# Per-Thread Memory Brainstorm

Date: 2026-05-07

## Background

Deerflow 现有的记忆功能是单租户的——不同会话都属于同一个用户，所有对话共享一份全局 `memory.json`。

要做一个新的记忆功能：不同对话属于不同用户，每个会话都有一个长期记忆，内容包括用户的使用习惯、个人信息、个人喜好和偏好语气。

## 现有记忆系统

- **存储**：单一全局 `backend/.deer-flow/memory.json`，所有会话共享
- **认证**：没有用户认证，没有用户隔离（better-auth 已搭建但未启用）
- **结构**：
  - `user`: workContext / personalContext / topOfMind
  - `history`: recentMonths / earlierContext / longTermBackground
  - `facts[]`: id, content, category, confidence, source
- **读路径**：system prompt 生成时注入 `<memory>...</memory>` XML 标签
- **写路径**：MemoryMiddleware 在对话后过滤消息 → MemoryUpdateQueue debounce 30s → MemoryUpdater 调 LLM 提取更新 → 原子写入
- **配置**：`config.yaml > memory`（enabled, debounce_seconds, max_facts, max_injection_tokens 等）

---

## 决策记录

### 存储方式: 数据库

~~文件存储 `threads/{thread_id}/profile-memory.json`~~ → **改为数据库表**，通过 `thread_id` 区分用户。

### 数据库: SQLite（本地/测试） + MySQL（生产环境）

### 表结构: 单表 + JSON 列（Option A）

### 依赖: 最小化，不引入 SQLAlchemy

SQLite 用标准库 `sqlite3`，MySQL 用 `pymysql`（纯 Python，轻量）。

### 与全局记忆关系: 策略 B（fallback）

Per-thread 有记忆就用 per-thread 的，没有就 fallback 到全局记忆。

### 首次对话: 不主动询问用户偏好

---

## 1. 数据库表设计

```sql
-- SQLite
CREATE TABLE IF NOT EXISTS thread_memory (
    thread_id TEXT PRIMARY KEY,
    profile TEXT NOT NULL DEFAULT '{}',
    preferences TEXT NOT NULL DEFAULT '{}',
    facts TEXT NOT NULL DEFAULT '[]',
    last_updated TEXT NOT NULL DEFAULT (datetime('now'))
);

-- MySQL
CREATE TABLE IF NOT EXISTS thread_memory (
    thread_id VARCHAR(64) PRIMARY KEY,
    profile JSON NOT NULL,
    preferences JSON NOT NULL,
    facts JSON NOT NULL,
    last_updated TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);
```

**profile** ({})：

| 字段 | 类型 | 说明 |
|------|------|------|
| `name` | `string \| null` | 用户称呼 |
| `role` | `string \| null` | 职业/角色 |
| `expertise` | `string[]` | 技术栈/专业领域 |
| `language` | `"zh-CN" \| "en-US" \| null` | 使用的语言 |
| `context` | `string \| null` | 其他上下文（自由文本） |

**preferences** ({})：

| 字段 | 类型 | 说明 |
|------|------|------|
| `tone` | `"casual" \| "formal" \| "technical" \| "friendly" \| null` | 语气偏好 |
| `verbosity` | `"concise" \| "detailed" \| null` | 回答详细程度 |
| `codeStyle` | `string \| null` | 代码风格偏好 |
| `other` | `string \| null` | 其他偏好（自由文本） |

**facts** ([])：复用现有全局记忆的 fact 结构

```json
{
  "id": "fact_abc123",
  "content": "用户在使用 React + TypeScript",
  "category": "tech_stack | preference | personal | context | goal",
  "confidence": 0.9,
  "createdAt": "2026-05-07T...",
  "source": "thread_id"
}
```

**说明**：三个 JSON 字段在 SQLite 中存为 TEXT（sqlite3 标准库没有原生 JSON 类型），在 MySQL 中存为 JSON。代码层面读写时做 `json.dumps` / `json.loads`，对上层透明。

## 2. config.yaml 新增配置段

```yaml
thread_memory:
  enabled: true
  debounce_seconds: 30
  model_name: null            # null = 使用默认模型
  max_facts: 100
  fact_confidence_threshold: 0.7
  injection_enabled: true
  max_injection_tokens: 2000

  database:
    type: sqlite              # sqlite | mysql
    sqlite:
      path: "thread_memory.db"
    mysql:
      host: "localhost"
      port: 3306
      user: "root"
      password: "$MYSQL_PASSWORD"
      database: "deerflow"
```

大部分字段和现有 `memory` 配置段语义相同，可以在两个配置段之间复用。`database` 段按 type 取子段，工厂函数只读自己需要的部分。

## 3. 存储层设计

### 3.1 抽象接口

```python
# deerflow/agents/memory/thread_storage.py

import abc
import json
import sqlite3
from datetime import datetime
from typing import Any


class ThreadMemoryStorage(abc.ABC):

    @abc.abstractmethod
    def load(self, thread_id: str) -> dict[str, Any] | None:
        """加载指定 thread 的记忆，不存在返回 None。"""
        ...

    @abc.abstractmethod
    def save(self, thread_id: str, data: dict[str, Any]) -> bool:
        """保存指定 thread 的记忆（upsert）。"""
        ...

    @abc.abstractmethod
    def delete(self, thread_id: str) -> bool:
        """删除指定 thread 的记忆（thread 被删除时联动）。"""
        ...


def _create_empty_memory() -> dict[str, Any]:
    """Per-thread 记忆的初始空结构。"""
    return {
        "profile": {
            "name": None,
            "role": None,
            "expertise": [],
            "language": None,
            "context": None,
        },
        "preferences": {
            "tone": None,
            "verbosity": None,
            "codeStyle": None,
            "other": None,
        },
        "facts": [],
    }


def _row_to_memory(row: tuple) -> dict[str, Any]:
    """将数据库行转为 memory dict。SQLite 的 JSON 列存的是 TEXT，需要 parse。"""
    return {
        "threadId": row[0],
        "profile": json.loads(row[1]),
        "preferences": json.loads(row[2]),
        "facts": json.loads(row[3]),
        "lastUpdated": row[4],
    }
```

### 3.2 SQLite 实现（本地测试）

```python
class SqliteThreadMemoryStorage(ThreadMemoryStorage):

    def __init__(self, db_path: str):
        self._conn = sqlite3.connect(db_path)
        self._conn.execute("""
            CREATE TABLE IF NOT EXISTS thread_memory (
                thread_id TEXT PRIMARY KEY,
                profile TEXT NOT NULL DEFAULT '{}',
                preferences TEXT NOT NULL DEFAULT '{}',
                facts TEXT NOT NULL DEFAULT '[]',
                last_updated TEXT NOT NULL DEFAULT (datetime('now'))
            )
        """)
        self._conn.commit()

    def load(self, thread_id: str) -> dict | None:
        row = self._conn.execute(
            "SELECT thread_id, profile, preferences, facts, last_updated "
            "FROM thread_memory WHERE thread_id = ?",
            (thread_id,)
        ).fetchone()
        return _row_to_memory(row) if row else None

    def save(self, thread_id: str, data: dict) -> bool:
        now = datetime.utcnow().isoformat() + "Z"
        self._conn.execute("""
            INSERT INTO thread_memory (thread_id, profile, preferences, facts, last_updated)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(thread_id) DO UPDATE SET
                profile = excluded.profile,
                preferences = excluded.preferences,
                facts = excluded.facts,
                last_updated = excluded.last_updated
        """, (
            thread_id,
            json.dumps(data["profile"], ensure_ascii=False),
            json.dumps(data["preferences"], ensure_ascii=False),
            json.dumps(data["facts"], ensure_ascii=False),
            now,
        ))
        self._conn.commit()
        return True

    def delete(self, thread_id: str) -> bool:
        self._conn.execute("DELETE FROM thread_memory WHERE thread_id = ?", (thread_id,))
        self._conn.commit()
        return True
```

### 3.3 MySQL 实现（生产环境）

```python
class MysqlThreadMemoryStorage(ThreadMemoryStorage):

    def __init__(self, host: str, port: int, user: str, password: str, database: str):
        import pymysql
        self._conn = pymysql.connect(
            host=host, port=port, user=user, password=password, database=database,
            charset="utf8mb4",
        )
        with self._conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS thread_memory (
                    thread_id VARCHAR(64) PRIMARY KEY,
                    profile JSON NOT NULL,
                    preferences JSON NOT NULL,
                    facts JSON NOT NULL,
                    last_updated TIMESTAMP NOT NULL
                        DEFAULT CURRENT_TIMESTAMP
                        ON UPDATE CURRENT_TIMESTAMP
                )
            """)
        self._conn.commit()

    def load(self, thread_id: str) -> dict | None:
        with self._conn.cursor() as cur:
            cur.execute(
                "SELECT thread_id, profile, preferences, facts, last_updated "
                "FROM thread_memory WHERE thread_id = %s",
                (thread_id,)
            )
            row = cur.fetchone()
        return _row_to_memory(row) if row else None

    def save(self, thread_id: str, data: dict) -> bool:
        now = datetime.utcnow()
        with self._conn.cursor() as cur:
            cur.execute("""
                INSERT INTO thread_memory (thread_id, profile, preferences, facts, last_updated)
                VALUES (%s, %s, %s, %s, %s)
                ON DUPLICATE KEY UPDATE
                    profile = VALUES(profile),
                    preferences = VALUES(preferences),
                    facts = VALUES(facts),
                    last_updated = VALUES(last_updated)
            """, (
                thread_id,
                json.dumps(data["profile"], ensure_ascii=False),
                json.dumps(data["preferences"], ensure_ascii=False),
                json.dumps(data["facts"], ensure_ascii=False),
                now,
            ))
        self._conn.commit()
        return True

    def delete(self, thread_id: str) -> bool:
        with self._conn.cursor() as cur:
            cur.execute("DELETE FROM thread_memory WHERE thread_id = %s", (thread_id,))
        self._conn.commit()
        return True
```

### 3.4 工厂函数

```python
def get_thread_memory_storage() -> ThreadMemoryStorage:
    """从 config 读取 database 配置，构建对应的 storage 实例（单例）。"""
    config = get_thread_memory_config()
    db = config.database

    if db.type == "sqlite":
        return SqliteThreadMemoryStorage(db.sqlite.path)
    elif db.type == "mysql":
        return MysqlThreadMemoryStorage(
            host=db.mysql.host,
            port=db.mysql.port,
            user=db.mysql.user,
            password=db.mysql.password,
            database=db.mysql.database,
        )
    else:
        raise ValueError(f"Unknown thread_memory database type: {db.type}")
```

### 3.5 注意事项

- **JSON 在 SQLite 中存为 TEXT**：`sqlite3` 标准库没有 JSON 类型，用 TEXT 存储 `json.dumps` 的结果。读写时做序列化/反序列化。MySQL 用原生 JSON 列，`pymysql` 自动处理。
- **upsert 语法差异**：SQLite 用 `ON CONFLICT ... DO UPDATE SET`，MySQL 用 `ON DUPLICATE KEY UPDATE`，语义等价。
- **连接管理**：两个实现都在 `__init__` 创建连接并持有。单线程场景没问题。如果将来需要并发，可以加连接池或改为每次操作创建连接。

---

## 4. upsert 语义：全量替换 vs 合并更新

### 两种模式

**模式 A — 增量合并**（LLM 出 delta，应用层合并）：

```
LLM 输入:  现有记忆 + 新对话
LLM 输出:  { profile: { name: "新值", shouldUpdate: true }, newFacts: [...], factsToRemove: [...] }
应用层:    读取现有记忆 → 按 delta 逐字段合并 → 写入
```

现有全局记忆用的就是这个模式。LLM 输出里带 `shouldUpdate` 标记和 `factsToRemove` 列表，应用代码做合并。

**模式 B — 全量替换**（LLM 出完整状态，应用层直接覆盖）：

```
LLM 输入:  现有记忆 + 新对话
LLM 输出:  { profile: { name: "...", role: "...", ... }, preferences: {...}, facts: [...] }
应用层:    INSERT ... ON CONFLICT DO UPDATE（整行覆盖）
```

### 选择模式 B 的理由

1. **profile 和 preferences 本身很小**。每个对象 5-6 个字段，全部输出最多几十个 token，增量节省的 token 可以忽略。

2. **去重和淘汰由 LLM 负责，应用层零逻辑**。LLM 看到了完整的现有记忆，在 prompt 中就能决定哪些 facts 要保留、哪些过时了要删、哪些要合并。应用代码只需要 `json.dumps` + upsert。

3. **避免字段删除的尴尬**。如果 LLM 想把 `profile.context` 从 `"前端开发者"` 改成 `null`（表示不再确定这个信息），增量模式需要额外表达"显式置 null"还是"不变"，全量替换没有歧义。

4. **和现有全局记忆的模式不同是合理的**。全局记忆的 `history` 有大量的对话摘要文本，不适合全量替换。Per-thread 记忆的 profile/preferences 是结构化的元数据，全量输出成本低。

### 具体流程

```
用户对话结束
  ↓
MemoryMiddleware.after_agent() 提取 user + final AI 消息
  ↓
queue.add(thread_id, messages)   # debounce 30s
  ↓
ThreadMemoryUpdater.update()
  1. 从 DB 读取现有记忆（不存在就用 _create_empty_memory()）
  2. 构建 prompt: "以下是用户的现有画像和偏好：{existing_memory}，以下是新的对话：{conversation}，请更新用户画像。"
  3. LLM 返回完整的 profile + preferences + facts
  4. storage.save(thread_id, data)  # upsert 整行覆盖
```

**关键点**：LLM 的 prompt 里放了**现有记忆**，LLM 看到之后自己决定：
- 保留哪些 facts
- 更新哪些 profile 字段
- 新增什么偏好
- 删除过时的信息（不输出就是删除）

应用代码不做任何合并判断，只负责把 LLM 输出写入数据库。

---

## 5. 更新路径

### 5.1 MemoryMiddleware 改造（最小改动）

在现有 `MemoryMiddleware.after_agent()` 中加一段逻辑，当 `thread_id` 存在时，同时向 per-thread 记忆的 queue 推一条：

```python
# 现有逻辑：全局记忆
queue = get_memory_queue()
queue.add(thread_id=thread_id, messages=filtered_messages, ...)

# 新增：per-thread 记忆
if thread_id:
    thread_queue = get_thread_memory_queue()
    thread_queue.add(thread_id=thread_id, messages=filtered_messages)
```

### 5.2 ThreadMemoryUpdater

新类，结构类似现有的 `MemoryUpdater`，但使用不同的 prompt 和存储后端：

```python
class ThreadMemoryUpdater:
    def update(self, messages, thread_id):
        storage = get_thread_memory_storage()
        existing = storage.load(thread_id) or _create_empty_memory()

        prompt = THREAD_MEMORY_UPDATE_PROMPT.format(
            existing_memory=json.dumps(existing, ensure_ascii=False),
            conversation=format_conversation(messages),
        )

        response = model.invoke(prompt)
        new_memory = parse_llm_output(response)  # { profile, preferences, facts }

        storage.save(thread_id, new_memory)
```

### 5.3 Prompt 设计要点

与全局记忆 prompt 的关键区别：

| | 全局记忆 prompt | Per-thread 记忆 prompt |
|---|---|---|
| **目标** | "对话中发生了什么" | "这个人是谁、喜欢什么" |
| **输出** | user context 摘要 + history 摘要 + facts | profile + preferences + facts |
| **侧重** | 保留对话内容的事实性信息 | 推断用户的身份、偏好、风格 |
| **语气影响** | 无 | 输出 `preferences.tone` 直接影响后续回复风格 |

---

## 6. 读取路径（注入 System Prompt）

```python
def inject_thread_memory(system_prompt: str, thread_id: str) -> str:
    storage = get_thread_memory_storage()
    memory = storage.load(thread_id)

    if memory is None:
        # fallback 到全局记忆
        return inject_global_memory(system_prompt)

    # 生成 <memory profile="..."> 标签注入 system prompt
    profile_xml = _format_profile_xml(memory)
    return system_prompt + "\n" + profile_xml
```

注入内容的 XML 结构示例：

```xml
<memory>
  <profile>
    <name>张三</name>
    <role>全栈工程师</role>
    <expertise>React, TypeScript, Python</expertise>
    <language>zh-CN</language>
    <context>在做一个电商项目</context>
  </profile>
  <preferences>
    <tone>casual</tone>
    <verbosity>detailed</verbosity>
    <codeStyle>prefers functional components with hooks</codeStyle>
  </preferences>
</memory>
```

语气偏好（`preferences.tone`）不直接改 system prompt 模板，而是放在 `<preferences>` XML 里让 LLM 自己理解。方式简单，不用维护 prompt 模板的分支逻辑。如果发现 LLM 不遵循，再考虑动态改写 prompt 模板。

---

## 7. Thread 删除时的联动

Gateway 已有 `DELETE /api/threads/{id}`。在现有 handler 中加一行：

```python
# app/gateway/routers/threads.py
@router.delete("/api/threads/{thread_id}")
async def delete_thread(thread_id: str):
    # ... 现有清理逻辑 ...

    # 新增：删除 per-thread 记忆
    get_thread_memory_storage().delete(thread_id)
```

---

## 8. 实施步骤

1. **新增配置模型** — `thread_memory_config.py`（参考现有 `memory_config.py`）
2. **新增存储层** — `thread_storage.py`（`ThreadMemoryStorage` + `SqliteThreadMemoryStorage` + `MysqlThreadMemoryStorage`）
3. **新增 prompt** — `thread_memory_prompt.py`（用于 LLM 提取用户画像）
4. **新增 updater** — 或扩展现有 `MemoryUpdater`，根据 `thread_id` 参数路由到不同逻辑
5. **改造 middleware** — `MemoryMiddleware` 中加 per-thread 记忆的 queue 逻辑
6. **改造注入** — system prompt 生成时注入 `<memory>` 标签
7. **扩展 thread 删除 handler** — 联动删除 DB 记录
8. **写入测试** — `test_thread_memory_storage.py`, `test_thread_memory_updater.py`

## 9. 待确认事项

- [ ] pymysql 作为新依赖是否 OK？
- [ ] `database` 配置段结构是否合适？
- [ ] upsert 使用全量替换模式（模式 B）是否认同？

## 10. 第二轮脑暴（风险前置）

下面这轮不是改大方向，而是把容易在落地时踩坑的点先钉住。

### 10.1 隔离键：`thread_id` 是否足够？

当前设计用 `thread_id` 作为主键隔离用户记忆，简单可行。但有一个隐含前提：
- 一个 thread 永远只对应一个真实用户

如果未来支持“同一用户多 thread 共享画像”或“thread 可能转移 owner”，只用 `thread_id` 会限制扩展。

可选路径：

- 路径 A（维持现状，推荐短期）：主键 `thread_id`，最快上线。
- 路径 B（兼容未来）：增加 `owner_id`（可空），并加索引 `(owner_id, thread_id)`。

建议：
- 第一版继续 `thread_id`，但在表里预留 `owner_id` nullable 字段，避免后续大迁移。

### 10.2 并发一致性：同一 thread 的并发写覆盖问题

场景：同一 thread 在短时间内触发多次 update，后到达的旧结果可能覆盖先到达的新结果。

可选保护：

- 方案 A：`last_updated` 乐观锁（更新时带 where 条件）
- 方案 B：`memory_version` 整数版本号（推荐）
- 方案 C：严格串行队列（单 thread 单 worker）

建议：
- 加 `memory_version`（默认 0）。`save` 时做 compare-and-swap 语义：
  - 读取 version = n
  - 写入时要求 version 仍为 n，成功后 version = n+1
  - 失败则重试一次（重新 load + merge prompt 再写）

这样不需要分布式锁，也能规避“旧结果回写”。

### 10.3 记忆质量控制：防止噪声和幻觉固化

LLM 抽取用户画像时，最大风险是把一次性表达当长期偏好。

建议加三道门：

1. 事实类别阈值
- `preference` 类阈值可略低（如 0.7）
- `personal` 类阈值更高（如 0.85）

2. 稳定性规则
- 同类偏好至少被 2 次独立对话支持，才提升为 profile/preference 的强字段

3. 冲突降级
- 新旧事实冲突时，不立刻删旧值
- 先把旧值降权并标记 `supersededBy`，下一轮再淘汰

### 10.4 隐私与合规：先定义“不能记”的边界

建议在 prompt 与代码都加 denylist（双保险）：

- 默认不写入：身份证号、手机号、邮箱、住址、银行卡、密码/API Key 等敏感信息
- 允许写入：技术偏好、工作语境、沟通风格、项目目标

实现上：
- 在 `ThreadMemoryUpdater` parse 后做一次 server-side scrub
- 命中敏感模式就丢弃并打审计日志（不落库原文）

### 10.5 注入预算：避免 memory 挤爆上下文

当前有 `max_injection_tokens`，但还缺“裁剪策略”。

建议固定优先级：
1. profile（最高）
2. preferences
3. facts（按 confidence + recency 排序后截断）

当超预算时：
- 永远保留 profile/preference
- 只裁剪 facts

### 10.6 可观测性：上线后如何判断有效

建议最小指标集：

- `thread_memory_update_total{status=ok|error}`
- `thread_memory_injection_tokens`
- `thread_memory_fact_count`
- `thread_memory_update_latency_ms`
- `thread_memory_conflict_retry_total`

加两条抽样日志：
- 更新前后摘要 diff（脱敏后）
- 注入片段长度与截断原因

### 10.7 迁移与回滚策略（从全局记忆过渡）

你已选 fallback 策略，这很好。建议再补两个机制：

- 冷启动导入（可选）
  - 首次访问 thread 且无 per-thread 记录时，从全局记忆抽取一份“弱画像”写入
  - 打 `bootstrapped_from_global=true`

- 一键回滚
  - 配置开关 `thread_memory.injection_enabled=false` 时，立刻只走全局注入
  - 更新链路可继续跑，便于回滚期间保留数据

### 10.8 API 语义建议（便于后续运维）

即使第一版 UI 不暴露，也建议预留内部接口：

- `GET /internal/thread-memory/{thread_id}`（脱敏视图）
- `DELETE /internal/thread-memory/{thread_id}`
- `POST /internal/thread-memory/{thread_id}/rebuild`

这样排障时不用直接查库。

---

## 11. 第三轮决策清单（进入实现前最后拍板）

- [ ] 表结构是否预留 `owner_id` 与 `memory_version`？
- [ ] 是否采用 `memory_version` 方案处理并发覆盖？
- [ ] 敏感信息 denylist 范围是否按 10.4 执行？
- [ ] 注入裁剪优先级是否固定为 profile > preferences > facts？
- [ ] 是否需要“冷启动导入”全局记忆到 per-thread？
- [ ] 是否要在首版就加内部运维接口？

如果以上 6 项确定，基本就能把实现风险压到可控范围内。

## 12. 默认拍板方案（建议直接采用）

目标：在不显著增加复杂度的前提下，拿到“可上线 + 可回滚 + 可演进”的第一版。

### 12.1 表结构默认值

采用：**预留 `owner_id` + 引入 `memory_version`**。

SQLite：

```sql
CREATE TABLE IF NOT EXISTS thread_memory (
    thread_id TEXT PRIMARY KEY,
    owner_id TEXT NULL,
    profile TEXT NOT NULL DEFAULT '{}',
    preferences TEXT NOT NULL DEFAULT '{}',
    facts TEXT NOT NULL DEFAULT '[]',
    memory_version INTEGER NOT NULL DEFAULT 0,
    last_updated TEXT NOT NULL DEFAULT (datetime('now'))
);

CREATE INDEX IF NOT EXISTS idx_thread_memory_owner_id ON thread_memory(owner_id);
```

MySQL：

```sql
CREATE TABLE IF NOT EXISTS thread_memory (
    thread_id VARCHAR(64) PRIMARY KEY,
    owner_id VARCHAR(64) NULL,
    profile JSON NOT NULL,
    preferences JSON NOT NULL,
    facts JSON NOT NULL,
    memory_version INT NOT NULL DEFAULT 0,
    last_updated TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_owner_id (owner_id)
);
```

### 12.2 并发一致性默认值

采用：**`memory_version` 乐观并发控制 + 失败重试 1 次**。

保存逻辑：
- `load()` 读出 `memory_version=n`
- `save()` 时执行条件更新（`WHERE thread_id=? AND memory_version=n`）
- 成功则 `memory_version=n+1`
- 如果受影响行数为 0，说明被并发写抢先，重读并重试一次

这能防止“旧更新覆盖新更新”，同时实现复杂度可控。

### 12.3 隐私策略默认值

采用：**默认拒绝敏感信息入库（代码层 hard filter）**。

默认 denylist：
- 手机号
- 邮箱
- 身份证号/护照号
- 银行卡号
- 密码/API Key/Token
- 详细住址

规则：
- 命中则从 `profile/preferences/facts` 中删除该片段
- 仅记录脱敏审计信息（类型 + 时间 + thread_id），不记录原文

### 12.4 注入裁剪默认值

采用固定优先级：**`profile > preferences > facts`**。

当超过 `max_injection_tokens`：
- 必保留：`profile`、`preferences`
- 裁剪：`facts`（按 `confidence DESC, createdAt DESC` 排序后截断）

这能保证人格与风格信息稳定注入，不被历史 facts 挤掉。

### 12.5 冷启动策略默认值

采用：**首版不开启自动冷启动导入**（`bootstrap_from_global=false`）。

理由：
- 降低“全局脏数据复制到 thread”风险
- 逻辑更清晰，便于观察 per-thread 记忆真实质量

补充：
- 保留 fallback（你当前已定）
- 后续若需要可加后台任务做可控回填

### 12.6 内部运维接口默认值

采用：**首版只加读接口，写接口延后**。

第一版建议：
- `GET /internal/thread-memory/{thread_id}`（脱敏后返回）

暂不做：
- `DELETE /internal/thread-memory/{thread_id}`（已有 thread delete 联动可覆盖主场景）
- `POST /internal/thread-memory/{thread_id}/rebuild`（二期再加）

这样可以先满足排障可见性，避免过早扩大运维面。

---

## 13. 实施前冻结版 Checklist（可直接转开发）

- [ ] DDL 按 12.1 落地（含 `owner_id`, `memory_version`, index）
- [ ] Storage `save()` 改为 compare-and-swap 语义
- [ ] Updater 增加一次冲突重试
- [ ] parse 后执行敏感信息 scrub
- [ ] 注入模块按 `profile > preferences > facts` 裁剪
- [ ] fallback 保持开启，冷启动导入保持关闭
- [ ] 增加最小指标与脱敏 diff 日志
- [ ] 增加内部只读排障接口

到这一步，方案已经可以进入实现，不需要再做大改。