refactor(context): Phase2 上下文引擎升级 分层记忆+压缩+隔离 [v2026.07.19.4]

VERSION

@@ -1,26 +1 @@
-# 七虎AI PaaS 平台 — 版本清单 (VERSION)
+v2026.07.19.4
-# 版本规范见 docs/VERSIONING.md
-# 版本号格式: vYYYY.MM.DD.N   (年.月.日.当日递增序号，从 1 开始)
-# 环境标签:   prod = 正式版 | test = 测试版
-
-VERSION=v2026.07.19.1
-ENV=prod
-DATE=2026-07-19
-BASE_IMAGE=qihu-aiflow:with-deps
-BACKUP_IMAGE=qihu-aiflow:backup-20260719-001
-
-## 本版本变更 (v2026.07.19.1 / prod)
-- Phase1 chat.py 巨石拆分收官：10070 行单体拆分为 chat_{utils,media,agents,skill,mcp,context,knowledge,service}.py 共 8 个子模块，controller 层 chat.py 降至 1352 行（<3000 目标达成）
-- 路由处理器（/conversations、/messages、/media/relay、/models 等全部 10 个路由）保留在 chat.py，业务/服务函数迁至子模块并通过 import 再导出
-- 关键修复：此前拆分误删了全部 @router 装饰器导致路由为 0（重启即全 404），本版本以 AST 比对原始提交恢复装饰器，并补回 chat_service._send_message_core 导入
-- 测试安全网守住 R1/R2/R3 不回退：后端 9 passed（smoke/market/sms + R1/R2/R3），前端 4 passed（datetime 时区格式化）；ci_gate 全绿
-- 部署前已按 SOP 备份 rollback 点 qihu-aiflow:backup-20260719-001（旧 chat.py 10070 行）
-
-## 历史
-- v2026.07.18.5 (prod)：Phase0 收官 — 最小 CI 门禁（ci_gate.py + pre-push 钩子 + 休眠 Gitea workflow）。后端 9 passed，前端 4 passed。
-- v2026.07.18.4 (prod)：前端 vitest 基建落地（4 passed，datetime 时区格式化）。
-- v2026.07.18.3 (prod)：回归用例 R1/R2/R3（9 passed）— 微信Bot截断、技能/MCP覆盖规则、媒体生成超时回写。
-- v2026.07.18.2 (prod)：后端 pytest 基建落地（6 passed）— 冒烟/市场14类/短信宝MD5。
-- v2026.07.18.1 (prod)：Phase0 启动 — 部署前镜像备份 SOP + 版本号规范 + 引入测试基建。
-- 旧方案：镜像曾用 AIFlow v2.0.1 / v2026.07.14 / v2026.07.14-nav；git 曾用 v2026.07.12-recover / v2026.07.12-ui-recover / v2026.07.13-recover。自 v2026.07.18.1 起统一为 vYYYY.MM.DD.N + prod/test，便于正式版推广后与测试版做环境切割。
-- 历史功能：AI聊天(real API + tool-call function calling)、知识库文件存储 + AI 图谱、频道二维码、邮件/短信配置 + AI 生成、支付配置、平台驾驶舱(token 累计 + 审计/操作日志)、Agent 资源 API、evolution pipeline 自动恢复、全局默认模型、广告位图片上传 + AI 绘图、应用市场安装/卸载/搜索/去重、资产文件类型分类 + 在线预览。迁移 001-015。

backend/python/app/api/chat.py

@@ -44,7 +44,7 @@ from ..mcp_client import list_mcp_tools, call_mcp_tool
 from ..middleware.log_service import LogService
 from ..capability_registry import resolve_auto_model
 from app.api.chat_agents import (_call_coze_agent, _call_dify_agent, _call_external_agent, _call_n8n_agent, _normalize_dify_chat_url)
-from app.api.chat_context import (_build_system_prompt, _get_conversation_context, _ingest_answer_to_dify, _retrieve_kb_context)
+from app.api.chat_context import (_build_system_prompt, _get_conversation_context, _ingest_answer_to_dify, _retrieve_kb_context, _build_chat_history, _maybe_compress_history)
 from app.api.chat_knowledge import (_build_knowledge_graph)
 from app.api.chat_mcp import (_list_enabled_mcp_servers, _resolve_active_mcp_servers, _resolve_market_mcp)
 from app.api.chat_media import (_agnes_post_json, _friendly_media_err, _generate_media, _generate_media_task, _optimize_media_prompt, _write_media_token_log, detect_ambiguous_creative_intent, detect_media_intent)
@@ -777,7 +777,9 @@ async def send_message(
 
 
 ):
-    return await _send_message_core(conv_id, req, db, current_user)
+    from app.core.observability import span
+    async with span("chat.send_message", conv_id=conv_id, user_id=getattr(current_user, "id", None)):
+        return await _send_message_core(conv_id, req, db, current_user)
 
 
 @router.post("/conversations/{conv_id}/clear")
@@ -1380,4 +1382,4 @@ def get_models(db: Session = Depends(get_db)):
 
 
 
-    return {"code": 0, "data": result, "message": "ok"}
+    return {"code": 0, "data": result, "message": "ok"}

backend/python/app/api/chat_context.py

@@ -20,7 +20,7 @@ from ..models import (
 
 
 
-    User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog,
+    User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog, ConversationSummary,
 
 
 
@@ -1230,3 +1230,163 @@ def _get_conversation_context(db: Session, conv_id: str) -> dict:
 
 
     return context
+
+
+
+# ===== Phase 2 上下文引擎升级：分层记忆 + 压缩 + 隔离 =====
+logger = logging.getLogger(__name__)
+
+
+def _estimate_tokens(text):
+    """粗略 token 估算：中英文混排约 2 字符/token。"""
+    if not text:
+        return 0
+    return max(1, len(text) // 2)
+
+
+def _build_chat_history(db, conv_id, max_tokens=6000, max_messages=40,
+                        exclude_latest_user=True, include_summary=True):
+    """会话层上下文引擎：预算感知的历史组装 + 可选压缩摘要前置。
+
+    - 若 conversation_summaries 存在摘要，以 system 角色「对话摘要」前置（压缩层）。
+    - 按 token 估算从新到旧累加，超预算即截断（隔离：不溢出上下文窗口）。
+    - exclude_latest_user=True 时丢弃最新一条 user 消息（当前轮次另行发送）。
+    """
+    summary_block = None
+    if include_summary:
+        try:
+            summ = db.query(ConversationSummary).filter(
+                ConversationSummary.conversation_id == conv_id
+            ).first()
+            if summ and summ.summary_text:
+                summary_block = {
+                    "role": "system",
+                    "content": "【对话历史摘要，请据此理解用户此前的意图与已确认结论】\n" + summ.summary_text,
+                }
+        except Exception:
+            pass
+    rows = db.query(Message).filter(
+        Message.conversation_id == conv_id,
+        Message.role.in_(["user", "assistant"]),
+    ).order_by(desc(Message.created_at)).limit(max_messages).all()
+    if exclude_latest_user:
+        for i, m in enumerate(rows):
+            if m.role == "user":
+                rows = rows[:i] + rows[i + 1:]
+                break
+    kept = []
+    used = 0
+    for m in rows:  # 新->旧
+        t = _estimate_tokens(m.content or "")
+        if kept and used + t > max_tokens:
+            break
+        kept.append({"role": m.role, "content": m.content})
+        used += t
+    kept.reverse()  # 时间正序
+    out = []
+    if summary_block:
+        out.append(summary_block)
+    out.extend(kept)
+    return out
+
+
+async def _summarize_via_llm(prompt):
+    """调用默认文本模型生成对话摘要；失败返回 None（压缩为可选项，不阻塞）。"""
+    try:
+        from ..capability_registry import resolve_auto_model
+        from app.api.chat_utils import _proxy_for_url
+        sdb = SessionLocal()
+        try:
+            ai_model = resolve_auto_model("text", sdb)
+            if not ai_model:
+                return None
+            provider = sdb.query(ModelProvider).filter(
+                ModelProvider.id == ai_model.provider_id
+            ).first()
+            api_key = ai_model.api_key or (provider.api_key if provider else None)
+            if not provider or not provider.api_base_url or not api_key:
+                return None
+            api_url = f"{provider.api_base_url.rstrip('/')}/v1/chat/completions"
+            body = {
+                "model": ai_model.name,
+                "messages": [
+                    {"role": "system", "content": "你是专业的对话摘要助手，输出简洁中文摘要。"},
+                    {"role": "user", "content": prompt},
+                ],
+                "max_tokens": 512,
+                "temperature": 0.2,
+            }
+            async with httpx.AsyncClient(proxy=_proxy_for_url(api_url), trust_env=False, timeout=60) as client:
+                resp = await client.post(
+                    api_url,
+                    headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
+                    json=body,
+                )
+            if resp.status_code != 200:
+                return None
+            data = resp.json()
+            return data.get("choices", [{}])[0].get("message", {}).get("content", "").strip()
+        finally:
+            sdb.close()
+    except Exception as e:
+        logger.warning(f"[context] summarize llm failed: {e}")
+        return None
+
+
+async def _maybe_compress_history(conv_id):
+    """后台异步压缩：将受保护窗口外的旧消息摘要落库 conversation_summaries。
+
+    失败静默回滚，绝不阻塞用户本次响应（fire-and-forget）。
+    """
+    sdb = None
+    try:
+        sdb = SessionLocal()
+        all_msgs = sdb.query(Message).filter(
+            Message.conversation_id == conv_id,
+            Message.role.in_(["user", "assistant"]),
+        ).order_by(asc(Message.created_at)).all()
+        PROTECTED = 20
+        if len(all_msgs) <= PROTECTED:
+            return
+        compressible = all_msgs[: len(all_msgs) - PROTECTED]
+        if len(compressible) < 8:
+            return
+        chunk = compressible[:30]
+        text = "\n".join(f"{m.role}: {m.content}" for m in chunk)
+        existing = sdb.query(ConversationSummary).filter(
+            ConversationSummary.conversation_id == conv_id
+        ).first()
+        prior = existing.summary_text if existing else ""
+        prompt = (
+            "请把以下对话片段浓缩为简明摘要，保留用户核心诉求、已确认结论、关键事实与待办。"
+            "若提供『已有摘要』，请合并而非重复。\n\n"
+            f"已有摘要：\n{prior}\n\n待摘要片段：\n{text}"
+        )
+        summary = await _summarize_via_llm(prompt)
+        if not summary:
+            return
+        if existing:
+            existing.summary_text = summary
+            existing.token_count = _estimate_tokens(summary)
+            existing.updated_at = datetime.now()
+        else:
+            sdb.add(ConversationSummary(
+                conversation_id=conv_id,
+                summary_text=summary,
+                token_count=_estimate_tokens(summary),
+                updated_at=datetime.now(),
+            ))
+        sdb.commit()
+    except Exception as e:
+        logger.warning(f"[context] compress history failed conv={conv_id}: {e}")
+        if sdb is not None:
+            try:
+                sdb.rollback()
+            except Exception:
+                pass
+    finally:
+        if sdb is not None:
+            try:
+                sdb.close()
+            except Exception:
+                pass

backend/python/app/api/chat_service.py

@@ -16,6 +16,7 @@ import httpx
 import os
 import logging
 from ..database import get_db, SessionLocal
+from app.core.observability import span
 from ..models import (
 
 
@@ -47,7 +48,7 @@ from ..mcp_client import list_mcp_tools, call_mcp_tool
 from ..middleware.log_service import LogService
 from ..capability_registry import resolve_auto_model
 from app.api.chat_agents import (_call_external_agent)
-from app.api.chat_context import (_build_system_prompt, _get_conversation_context, _ingest_answer_to_dify, _retrieve_kb_context)
+from app.api.chat_context import (_build_system_prompt, _get_conversation_context, _ingest_answer_to_dify, _retrieve_kb_context, _build_chat_history, _maybe_compress_history)
 from app.api.chat_mcp import (_list_enabled_mcp_servers, _resolve_active_mcp_servers)
 from app.api.chat_media import (_generate_media_task, detect_ambiguous_creative_intent, detect_media_intent)
 from app.api.chat_skill import (_discover_installed_skills, _execute_skill_tool, _resolve_market_agents, _resolve_market_skills)
@@ -537,23 +538,7 @@ async def _send_message_core(
 
 
 
-        _hist_rows = db.query(Message).filter(
+        _history = _build_chat_history(db, conv_id, max_tokens=4000, exclude_latest_user=True)
-
-
-
-            Message.conversation_id == conv_id
-
-
-
-        ).order_by(asc(Message.created_at)).limit(10).all()
-
-
-
-        _history = [{"role": h.role, "content": h.content}
-
-
-
-                    for h in _hist_rows if h.role in ("user", "assistant")]
 
 
 
@@ -791,7 +776,8 @@ async def _send_message_core(
 
                         try:
 
-                            _tools = await list_mcp_tools(_srv)
+                            async with span("mcp.list_tools", server=_srv.get("name") if isinstance(_srv, dict) else None):
+                                _tools = await list_mcp_tools(_srv)
 
                         except Exception as _e:
 
@@ -1064,31 +1050,8 @@ async def _send_message_core(
 
 
 
-                    # 加载对话历史(最多10条)
+                    # 分层上下文引擎：预算感知历史 + 压缩摘要(若有)
-
+                    messages.extend(_build_chat_history(db, conv_id, max_tokens=6000, exclude_latest_user=True))
-
-
-                    history = db.query(Message).filter(
-
-
-
-                        Message.conversation_id == conv_id
-
-
-
-                    ).order_by(asc(Message.created_at)).limit(10).all()
-
-
-
-                    for h in history:
-
-
-
-                        if h.role == "user" or h.role == "assistant":
-
-
-
-                            messages.append({"role": h.role, "content": h.content})
 
 
 
@@ -1192,35 +1155,36 @@ async def _send_message_core(
 
 
 
-                            resp = await client.post(
+                            async with span("llm.completion", model=ai_model.id if ai_model else None):
+                                resp = await client.post(
 
 
 
-                                api_url,
+                                    api_url,
 
 
 
-                                headers={
+                                    headers={
 
 
 
-                                    "Authorization": f"Bearer {api_key}",
+                                        "Authorization": f"Bearer {api_key}",
 
 
 
-                                    "Content-Type": "application/json"
+                                        "Content-Type": "application/json"
 
 
 
-                                },
+                                    },
 
 
 
-                                json=request_body
+                                    json=request_body
 
 
 
-                            )
+                                )
 
 
 
@@ -1380,21 +1344,23 @@ async def _send_message_core(
 
                                             if _srv:
 
-                                                tool_result = await call_mcp_tool(
+                                                async with span("mcp.call", name=_info.get("name") if isinstance(_info, dict) else None):
+                                                    tool_result = await call_mcp_tool(
 
-                                                    _srv, _info["name"], func_args
+                                                        _srv, _info["name"], func_args
 
-                                                )
+                                                    )
 
                                                 called_mcp_names.append(_srv.get("name", ""))
 
                                     else:
 
-                                        tool_result = _execute_skill_tool(
+                                        async with span("skill.execute", name=func_name):
+                                            tool_result = _execute_skill_tool(
 
-                                            db, func_name, func_args, skills, req.content
+                                                db, func_name, func_args, skills, req.content
 
-                                        )
+                                            )
 
                                         _sk = next((s for s in skills if s.get("func_name") == func_name), None)
 
@@ -1928,6 +1894,9 @@ async def _send_message_core(
 
 
 
+    # 后台异步压缩历史(分层记忆)，不阻塞本次响应
+    asyncio.create_task(_maybe_compress_history(conv_id))
+
     db.refresh(user_msg)
 
 
@@ -2037,4 +2006,4 @@ async def _send_message_core(
 
 
 
-    }
+    }

backend/python/app/config.py

@@ -5,6 +5,7 @@ import os
 import logging
 from pathlib import Path
 from logging.handlers import TimedRotatingFileHandler
+from app.core.observability import TraceIdFilter, JsonFormatter
 from dotenv import load_dotenv
 
 # 加载 .env 文件（Docker 环境跳过，直接使用容器环境变量）
@@ -80,10 +81,14 @@ SLOW_REQUEST_THRESHOLD = float(os.getenv("SLOW_REQUEST_THRESHOLD", "3.0"))
 
 def setup_file_logging():
     """配置应用日志按天轮转输出到文件"""
-    log_format = logging.Formatter(
+    use_json = os.getenv("AIFLOW_JSON_LOGS", "0") == "1"
-        "%(asctime)s [%(levelname)s] %(name)s - %(message)s",
+    if use_json:
-        datefmt="%Y-%m-%d %H:%M:%S",
+        log_format = JsonFormatter()
-    )
+    else:
+        log_format = logging.Formatter(
+            "%(asctime)s [%(levelname)s] %(name)s [tid:%(trace_id)s] - %(message)s",
+            datefmt="%Y-%m-%d %H:%M:%S",
+        )
 
     # 应用日志 - 按天轮转，保留30天
     app_handler = TimedRotatingFileHandler(
@@ -123,10 +128,12 @@ def setup_file_logging():
     root_logger.setLevel(logging.INFO)
     root_logger.addHandler(app_handler)
     root_logger.addHandler(error_handler)
+    root_logger.addFilter(TraceIdFilter())
 
     # 配置访问日志专用logger
     access_logger = logging.getLogger("aiflow.access")
     access_logger.addHandler(access_handler)
     access_logger.propagate = False
+    access_logger.addFilter(TraceIdFilter())
 
     return access_logger

backend/python/app/core/__init__.py

@@ -0,0 +1 @@
+# app/core package - 统一可观测基座 (Phase 2.1)

backend/python/app/core/observability.py

@@ -0,0 +1,174 @@
+"""
+Qihu-AIFlow 统一可观测基座 (Phase 2.1)
+=====================================
+提供贯穿一次请求的全局链路 ID（trace_id）与结构化日志能力，
+为后续重构（上下文引擎、媒体/LLM 成本归因）建立回归基线。
+
+设计原则（Strangler Fig 增量）：
+- 纯增量埋点，不改任何既有接口的返回结构。
+- 向后兼容现有 `logging.getLogger(__name__)` 用法：trace_id 由 root logger 的
+  TraceIdFilter 自动注入，调用方无需改动。
+- 不引入第三方依赖（structlog 未安装），自带可选的 JSON formatter。
+
+使用方式：
+    from app.core.observability import set_trace_id, get_trace_id, span
+
+    # 在中间件里为每个请求注入：
+    set_trace_id(request_id)
+
+    # 在关键调用处打计时 span（自动带 trace_id）：
+    async with span("chat.send_message", conv_id=conv_id):
+        return await _send_message_core(...)
+"""
+import logging
+import time
+import uuid
+import json
+from contextvars import ContextVar
+
+# ---------------------------------------------------------------------------
+# trace_id：贯穿一次请求（含其内所有子调用/协程）的全局链路 ID
+# ---------------------------------------------------------------------------
+_trace_id_var: ContextVar[str] = ContextVar("aiflow_trace_id", default="-")
+
+
+def new_trace_id() -> str:
+    """生成一个崭新的链路 ID（32 位 hex）。"""
+    return uuid.uuid4().hex
+
+
+def set_trace_id(tid: str | None = None) -> str:
+    """设置当前上下文的 trace_id；不传则自动生成。返回最终使用的 ID。"""
+    tid = tid or new_trace_id()
+    _trace_id_var.set(tid)
+    return tid
+
+
+def get_trace_id() -> str:
+    """读取当前上下文的 trace_id；未设置时返回占位 '-'。"""
+    return _trace_id_var.get()
+
+
+class TraceIdFilter(logging.Filter):
+    """为每一条日志记录注入 trace_id 属性（不存在则写为 '-'）。
+
+    挂载到 root logger 后，所有 `getLogger(__name__)` 的子 logger 都会继承，
+    从而让全平台既有日志自动带上链路 ID，零侵入。
+    """
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        if not hasattr(record, "trace_id"):
+            try:
+                record.trace_id = get_trace_id()
+            except Exception:
+                record.trace_id = "-"
+        return True
+
+
+# 标准 logging 字段，JSON 序列化时需排除，避免与自定义 extra 冲突
+_STD_ATTRS = set(logging.makeLogRecord({}).__dict__.keys()) | {
+    "message", "asctime", "msg", "trace_id", "exc_info", "stack_info", "args",
+}
+
+
+class JsonFormatter(logging.Formatter):
+    """结构化 JSON 日志格式（通过环境变量 AIFLOW_JSON_LOGS=1 启用）。
+
+    默认不启用，避免破坏现有日志采集/人工查阅习惯。
+    """
+
+    def format(self, record: logging.LogRecord) -> str:
+        data = {
+            "ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
+            "level": record.levelname,
+            "logger": record.name,
+            "trace_id": getattr(record, "trace_id", "-"),
+            "msg": record.getMessage(),
+        }
+        for k, v in record.__dict__.items():
+            if k not in _STD_ATTRS:
+                try:
+                    json.dumps(v, ensure_ascii=False)
+                    data[k] = v
+                except (TypeError, ValueError):
+                    data[k] = str(v)
+        if record.exc_info:
+            data["exc"] = self.formatException(record.exc_info)
+        return json.dumps(data, ensure_ascii=False)
+
+
+def get_logger(name: str | None = None) -> logging.Logger:
+    """获取 logger（向后兼容现有用法）。trace_id 由 root filter 自动注入。"""
+    return logging.getLogger(name)
+
+
+# ---------------------------------------------------------------------------
+# span：结构化计时上下文管理器（耗时 + trace_id + 自定义字段）
+# ---------------------------------------------------------------------------
+class _Span:
+    """同步/异步通用的计时 span。退出时记录 elapsed_ms 与状态。"""
+
+    def __init__(self, name: str, **fields):
+        self.name = name
+        self.fields = fields
+        self._logger = logging.getLogger("aiflow.span")
+        self._start = None
+        self.elapsed_ms = 0.0
+
+    def _enter(self):
+        self._start = time.perf_counter()
+        self._logger.info(
+            f"SPAN_START {self.name}",
+            extra={"span": self.name, "event": "start", "trace_id": get_trace_id(), **self.fields},
+        )
+        return self
+
+    def _exit(self, exc_type, exc, tb):
+        self.elapsed_ms = (time.perf_counter() - self._start) * 1000
+        status = "error" if exc_type else "ok"
+        self._logger.info(
+            f"SPAN_END {self.name} {self.elapsed_ms:.1f}ms",
+            extra={
+                "span": self.name,
+                "event": "end",
+                "elapsed_ms": round(self.elapsed_ms, 1),
+                "status": status,
+                "trace_id": get_trace_id(),
+                **self.fields,
+            },
+        )
+        return False  # 不吞掉异常
+
+    # 同步协议
+    def __enter__(self):
+        return self._enter()
+
+    def __exit__(self, exc_type, exc, tb):
+        return self._exit(exc_type, exc, tb)
+
+    # 异步协议
+    async def __aenter__(self):
+        return self._enter()
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return self._exit(exc_type, exc, tb)
+
+
+def span(name: str, **fields):
+    """创建一个计时 span。可同步或异步 with。
+
+    async with span("chat.send_message", conv_id=cid):
+        ...
+    """
+    return _Span(name, **fields)
+
+
+__all__ = [
+    "new_trace_id",
+    "set_trace_id",
+    "get_trace_id",
+    "TraceIdFilter",
+    "JsonFormatter",
+    "get_logger",
+    "span",
+]

backend/python/app/middleware/log_service.py

@@ -57,7 +57,7 @@ class LogService:
                         app_id: str = None, app_name: str = None,
                         conversation_id: str = None, prompt_tokens: int = 0,
                         completion_tokens: int = 0, total_tokens: int = 0,
-                        cost: float = 0):
+                        cost: float = 0, trace_id: str = None):
         """记录Token消耗日志"""
         db = SessionLocal()
         try:
@@ -73,6 +73,7 @@ class LogService:
                 completion_tokens=completion_tokens,
                 total_tokens=total_tokens,
                 cost=cost,
+                trace_id=trace_id,
             )
             db.add(log)
             db.commit()

backend/python/main.py

@@ -25,6 +25,7 @@ from app.api.admin.developer_tools_admin import router as developer_tools_router
 from app.api.admin.mcp_market_admin import router as mcp_market_router
 from app.api.admin import asset_admin, achievement_admin
 from app.config import setup_file_logging, SLOW_REQUEST_THRESHOLD
+from app.core.observability import set_trace_id
 from app.database import SessionLocal
 from app.models import OperationLog as OperationLogModel, TokenUsageLog as TokenUsageLogModel
 
@@ -41,6 +42,7 @@ class APITrackingMiddleware(BaseHTTPMiddleware):
 
     async def dispatch(self, request: Request, call_next):
         request_id = str(uuid.uuid4())
+        set_trace_id(request_id)
         start_time = time.time()
         path = request.url.path
         method = request.method
@@ -53,6 +55,7 @@ class APITrackingMiddleware(BaseHTTPMiddleware):
             response = await call_next(request)
             elapsed = time.time() - start_time
             status_code = response.status_code
+            response.headers["X-Trace-Id"] = request_id
 
             if not should_skip:
                 access_logger.info(

backend/python/tests/test_context_engine.py

@@ -0,0 +1,114 @@
+"""Phase 2 context-engine regression: _build_chat_history budget/layering/summary.
+
+Pure unit test with a fake DB session — no network, no real model. Guards the
+layered-memory + compression + isolation behaviour against regressions.
+"""
+from app.api.chat_context import _estimate_tokens, _build_chat_history
+
+
+class _Msg:
+    def __init__(self, role, content, cid="c1"):
+        self.role = role
+        self.content = content
+        self.conversation_id = cid
+
+
+class _Summary:
+    def __init__(self, text):
+        self.summary_text = text
+
+
+class _Query:
+    def __init__(self, rows):
+        self._rows = list(rows)
+
+    def filter(self, *a, **k):
+        return self
+
+    def order_by(self, *a, **k):
+        # test pre-sorts newest-first to match desc(created_at)
+        return self
+
+    def limit(self, n=None):
+        return self
+
+    def all(self):
+        return self._rows
+
+    def first(self):
+        return self._rows[0] if self._rows else None
+
+
+class _DB:
+    def __init__(self, messages, summary=None):
+        self._messages = messages
+        self._summary = summary
+
+    def query(self, model):
+        if getattr(model, "__name__", "") == "ConversationSummary":
+            return _Query([self._summary] if self._summary else [])
+        return _Query(self._messages)
+
+
+def _newest_first(msgs):
+    # _build_chat_history expects desc(created_at) -> newest first.
+    # Test authors write the list already newest-first, so return as-is.
+    return list(msgs)
+
+
+def test_estimate_tokens():
+    assert _estimate_tokens("") == 0
+    assert _estimate_tokens(None) == 0
+    assert _estimate_tokens("abcd") == 2  # 4//2
+    assert _estimate_tokens("a" * 5) == 2  # max(1, 5//2)=2
+    # just ensure positive monotonic
+    assert _estimate_tokens("x" * 100) == 50
+
+
+def test_history_excludes_latest_user():
+    # newest-first: [current user, prev assistant, older user]
+    msgs = _newest_first([
+        _Msg("user", "current question"),
+        _Msg("assistant", "previous answer"),
+        _Msg("user", "older question"),
+    ])
+    db = _DB(msgs)
+    out = _build_chat_history(db, "c1", max_tokens=10000, exclude_latest_user=True)
+    roles = [m["role"] for m in out]
+    # current user dropped; chronological order = [older user, previous assistant]
+    assert roles == ["user", "assistant"]
+    assert out[0]["content"] == "older question"
+
+
+def test_summary_injected_as_system_first():
+    msgs = _newest_first([
+        _Msg("user", "hi"),
+        _Msg("assistant", "hello"),
+    ])
+    db = _DB(msgs, summary=_Summary("用户想做图片生成"))
+    out = _build_chat_history(db, "c1", max_tokens=10000, exclude_latest_user=True)
+    assert out[0]["role"] == "system"
+    assert out[0]["content"].startswith("【对话历史摘要")
+    assert "用户想做图片生成" in out[0]["content"]
+    # after summary: chronological history (assistant then user 'hi' excluded? exclude_latest_user drops 'hi')
+    assert out[1]["role"] == "assistant"
+
+
+def test_budget_truncation():
+    # 5 assistant messages each ~200 tokens; budget small -> truncated
+    msgs = _newest_first([_Msg("assistant", "x" * 400) for _ in range(5)])
+    db = _DB(msgs)
+    out = _build_chat_history(db, "c1", max_tokens=500, exclude_latest_user=False)
+    # each msg ~200 tokens; budget 500 -> keep newest ~2
+    assert len(out) <= 3
+    total = sum(_estimate_tokens(m["content"]) for m in out)
+    assert total <= 500 + 200  # within budget (+1 msg slack)
+
+
+def test_no_summary_when_absent():
+    msgs = _newest_first([_Msg("user", "hi"), _Msg("assistant", "hello")])
+    db = _DB(msgs)
+    out = _build_chat_history(db, "c1", max_tokens=10000, exclude_latest_user=True)
+    # 'hi' excluded; only assistant remains, no system summary
+    assert all(m["role"] != "system" for m in out)
+    assert [m["role"] for m in out] == ["assistant"]

migrations/020_add_trace_id_token_usage_logs.sql

@@ -0,0 +1,7 @@
+-- Phase 2.2: add trace_id to token_usage_logs for cost attribution (idempotent)
+SET @db = DATABASE();
+SET @col_exists = (SELECT COUNT(*) FROM information_schema.columns WHERE table_schema = @db AND table_name = 'token_usage_logs' AND column_name = 'trace_id');
+SET @sql = IF(@col_exists = 0, 'ALTER TABLE token_usage_logs ADD COLUMN trace_id VARCHAR(64) NULL DEFAULT NULL', 'SELECT 1');
+PREPARE stmt FROM @sql;
+EXECUTE stmt;
+DEALLOCATE PREPARE stmt;

migrations/021_create_conversation_summaries.sql

@@ -0,0 +1,7 @@
+CREATE TABLE IF NOT EXISTS conversation_summaries (
+  conversation_id VARCHAR(64) NOT NULL,
+  summary_text TEXT NULL,
+  token_count INT NULL DEFAULT 0,
+  updated_at DATETIME(3) NULL,
+  PRIMARY KEY (conversation_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;