From ce6c4cab81ea1b099db64513178b87784f0f6ddd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=BE=E5=B2=9B=E5=B0=91=E5=B8=85?= <342903488+xiaoyikunle0710-zy@users.noreply.github.com> Date: Thu, 15 Oct 2026 21:16:31 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat(llm):=20trace=20=E6=8A=A5=E6=96=87?= =?UTF-8?q?=E5=86=85=E8=81=94=20base64=20=E5=AA=92=E4=BD=93=E7=9A=84?= =?UTF-8?q?=E5=8D=95=E5=89=AF=E6=9C=AC=E5=AD=98=E5=82=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 落库前对 request_text / response_text / response_raw_text 各跑一遍截断器，达到门槛的 base64 载荷写入内容寻址 blob 库，正文替换为 [qq:blob sha256= bytes=] 占位符，替换后报文仍是合法 JSON。 llm_http_traces 新增 media_truncation 列记录 splice 清单与全文哈希；旧行 NULL 视作原样存储。读路径默认返回占位符文本，新增按 sha 的 blob 下载端点与 restore 参数。 每日留存清理之后按剩余行的引用集合 mark-sweep 无引用 blob，24h 保守带规避「blob 已写入、引用行尚未提交」的并发窗口。 由 LLM_TRACE_MEDIA_MIN_BYTES 控制，未设置或为 0 时原样存储，回滚即关开关。 Closes #291 --- docs/admin/configuration.md | 1 + docs/dev/llm-module.md | 2 + .../references/docs-admin-configuration.md | 1 + .../references/docs-dev-llm-module.md | 2 + skills.example/self-docs/references/index.md | 2 +- src/quickquip/app/web/routes/logs.py | 43 +- src/quickquip/llm/provider/media_truncator.py | 444 ++++++++++++++++++ src/quickquip/llm/provider/trace.py | 217 ++++++++- tests/unit/llm/provider/__init__.py | 0 .../unit/llm/provider/test_media_truncator.py | 205 ++++++++ 10 files changed, 892 insertions(+), 25 deletions(-) create mode 100644 src/quickquip/llm/provider/media_truncator.py create mode 100644 tests/unit/llm/provider/__init__.py create mode 100644 tests/unit/llm/provider/test_media_truncator.py diff --git a/docs/admin/configuration.md b/docs/admin/configuration.md index 3b8d4850..a3c23b8d 100644 --- a/docs/admin/configuration.md +++ b/docs/admin/configuration.md @@ -56,6 +56,7 @@ LLM 工具 `search_web` 与 `/search` 命令固定走项目内 SearXNG。普通 | 变量 | 说明 | 默认值 | |------|------|--------| | `LLM_TRACE_FLAG_FILE` | LLM HTTP Trace 持久开关文件路径；文件存在时按调用记录完整请求/响应 JSON 文本到 `data/llm_trace.db`，供 Web Admin 的 LLM Trace 页面按需读取 | — | +| `LLM_TRACE_MEDIA_MIN_BYTES` | 内联 base64 媒体载荷的截断门槛（字节）。达到门槛、且载荷构成被引号正确闭合的完整 JSON 字符串时，载荷本体写入 `data/trace_blobs/` 下的内容寻址副本，正文替换为 `[qq:blob sha256= bytes=]` 占位符；同一份载荷在请求、重组响应与原始 SSE 中只落盘一份。未设置或为 `0` 时原样存储。进程启动时读取 | `0` | ### 贴吧 diff --git a/docs/dev/llm-module.md b/docs/dev/llm-module.md index b580c963..d2a2e656 100644 --- a/docs/dev/llm-module.md +++ b/docs/dev/llm-module.md @@ -10,6 +10,8 @@ QuickQuip 的 LLM 模块为规则机器人提供对话扩展。管理员按群 LLM 运行时在 `LLM_TRACE_FLAG_FILE` 指向的开关文件存在时，把每次 HTTP 尝试写入 `data/llm_trace.db`。请求正文取自实际交给 HTTP 客户端的 UTF-8 JSON 序列化文本；普通响应保留 JSON 解析前的服务端文本；流式响应完整消费 SSE 后，由协议客户端重建 OpenAI Chat Completion、Claude Message、Gemini GenerateContent 或 OpenAI Responses 完整响应对象，同时保留 SSE 传输原文供管理员按需核对。索引、正文和单调递增的状态事件分开存储，Web Admin 先读取轻量调用元数据，管理员选择记录后再加载完整 Header 与正文。`run_tool_call_loop` 为一轮完整交互分配 Agent Loop ID，重试、故障切换和工具结果回送产生的 HTTP 调用按组内序号排列。 +多模态消息的内联 base64 载荷在落库前按内容寻址去重：`LLM_TRACE_MEDIA_MIN_BYTES` 达到门槛的载荷写入 `data/trace_blobs/`，正文只留 `[qq:blob sha256= bytes=]` 占位符，同一份载荷在请求、重组响应与原始 SSE 三处共用一个副本。每次截断把 splice 清单与全文哈希记入 `media_truncation` 列；读端默认返回占位符文本，`restore` 参数按清单回拼并逐项校验。每日留存清理之后，按剩余行的引用集合 mark-sweep 未被引用的 blob。截断只发生在落库侧，发往上游的报文不变；未达门槛、畸形或被截断的字符串一律原样存储。配置键与默认值见[管理员配置](../admin/configuration.md)。 + ### 1.1 执行记录的请求边界 群聊和私聊生成请求创建的执行记录按请求携带的 `trigger_kind` 保存触发类型；未显式指定时，群聊默认为 `group_direct`，私聊默认为 `private_direct`。被动群触发保存为 `group_passive`，供 Loop 详情与历史档案使用。历史记录保留已存分类，缺少原始触发证据时不推断回填。 diff --git a/skills.example/self-docs/references/docs-admin-configuration.md b/skills.example/self-docs/references/docs-admin-configuration.md index 738783f0..dffee6fb 100644 --- a/skills.example/self-docs/references/docs-admin-configuration.md +++ b/skills.example/self-docs/references/docs-admin-configuration.md @@ -58,6 +58,7 @@ LLM 工具 `search_web` 与 `/search` 命令固定走项目内 SearXNG。普通 | 变量 | 说明 | 默认值 | |------|------|--------| | `LLM_TRACE_FLAG_FILE` | LLM HTTP Trace 持久开关文件路径；文件存在时按调用记录完整请求/响应 JSON 文本到 `data/llm_trace.db`，供 Web Admin 的 LLM Trace 页面按需读取 | — | +| `LLM_TRACE_MEDIA_MIN_BYTES` | 内联 base64 媒体载荷的截断门槛（字节）。达到门槛、且载荷构成被引号正确闭合的完整 JSON 字符串时，载荷本体写入 `data/trace_blobs/` 下的内容寻址副本，正文替换为 `[qq:blob sha256= bytes=]` 占位符；同一份载荷在请求、重组响应与原始 SSE 中只落盘一份。未设置或为 `0` 时原样存储。进程启动时读取 | `0` | ### 贴吧 diff --git a/skills.example/self-docs/references/docs-dev-llm-module.md b/skills.example/self-docs/references/docs-dev-llm-module.md index aa2d1de4..86559974 100644 --- a/skills.example/self-docs/references/docs-dev-llm-module.md +++ b/skills.example/self-docs/references/docs-dev-llm-module.md @@ -12,6 +12,8 @@ QuickQuip 的 LLM 模块为规则机器人提供对话扩展。管理员按群 LLM 运行时在 `LLM_TRACE_FLAG_FILE` 指向的开关文件存在时，把每次 HTTP 尝试写入 `data/llm_trace.db`。请求正文取自实际交给 HTTP 客户端的 UTF-8 JSON 序列化文本；普通响应保留 JSON 解析前的服务端文本；流式响应完整消费 SSE 后，由协议客户端重建 OpenAI Chat Completion、Claude Message、Gemini GenerateContent 或 OpenAI Responses 完整响应对象，同时保留 SSE 传输原文供管理员按需核对。索引、正文和单调递增的状态事件分开存储，Web Admin 先读取轻量调用元数据，管理员选择记录后再加载完整 Header 与正文。`run_tool_call_loop` 为一轮完整交互分配 Agent Loop ID，重试、故障切换和工具结果回送产生的 HTTP 调用按组内序号排列。 +多模态消息的内联 base64 载荷在落库前按内容寻址去重：`LLM_TRACE_MEDIA_MIN_BYTES` 达到门槛的载荷写入 `data/trace_blobs/`，正文只留 `[qq:blob sha256= bytes=]` 占位符，同一份载荷在请求、重组响应与原始 SSE 三处共用一个副本。每次截断把 splice 清单与全文哈希记入 `media_truncation` 列；读端默认返回占位符文本，`restore` 参数按清单回拼并逐项校验。每日留存清理之后，按剩余行的引用集合 mark-sweep 未被引用的 blob。截断只发生在落库侧，发往上游的报文不变；未达门槛、畸形或被截断的字符串一律原样存储。配置键与默认值见[管理员配置](../admin/configuration.md)。 + ### 1.1 执行记录的请求边界 群聊和私聊生成请求创建的执行记录按请求携带的 `trigger_kind` 保存触发类型；未显式指定时，群聊默认为 `group_direct`，私聊默认为 `private_direct`。被动群触发保存为 `group_passive`，供 Loop 详情与历史档案使用。历史记录保留已存分类，缺少原始触发证据时不推断回填。 diff --git a/skills.example/self-docs/references/index.md b/skills.example/self-docs/references/index.md index 67c74aba..56ca0522 100644 --- a/skills.example/self-docs/references/index.md +++ b/skills.example/self-docs/references/index.md @@ -61,7 +61,7 @@ - `docs/dev/game-framework.md` → `references/docs-dev-game-framework.md` — 游戏框架开发者指南 ｜ 关键词：BaseGame、aliases、src/quickquip/app/message_pipeline.py、GameResult、finished、group_id、OrderedDict[str, Session]、src/quickquip/adapters/nonebot/commands.py、command_parts/、_ensure_schema() - `docs/dev/llm-configuration-api.md` → `references/docs-dev-llm-configuration-api.md` — LLM 配置后端接口 ｜ 关键词：data/llm_config.db、/ops/api/llm-configuration、quickquip_admin_session、Origin、Referer、/openapi.json、format_version、。顶层字段为、style_profiles、pricing - `docs/dev/llm-configuration-frontend.md` → `references/docs-dev-llm-configuration-frontend.md` — LLM 配置管理前端交付契约 ｜ 关键词：/ops/#/llm-config、draft_revision、published_revision、loaded_revision、draft_version、published_version、loaded_version、{saved_at, note}、PUT /note、202 -- `docs/dev/llm-module.md` → `references/docs-dev-llm-module.md` — QuickQuip LLM 模块说明 ｜ 关键词：/search、LLM_TRACE_FLAG_FILE、data/llm_trace.db、run_tool_call_loop、trigger_kind、group_direct、private_direct、group_passive、interrupted、request_cancelled +- `docs/dev/llm-module.md` → `references/docs-dev-llm-module.md` — QuickQuip LLM 模块说明 ｜ 关键词：/search、LLM_TRACE_FLAG_FILE、data/llm_trace.db、run_tool_call_loop、LLM_TRACE_MEDIA_MIN_BYTES、data/trace_blobs/、[qq:blob sha256= bytes=]、media_truncation、restore、trigger_kind - `docs/dev/mcp-integration.md` → `references/docs-dev-mcp-integration.md` — QuickQuip MCP 集成说明 ｜ 关键词：ToolRegistry、data/llm_config.db、settings.mcp、config.toml、search_web、tavily_search、tavily_crawl、tavily_research、docker、/var/run/docker.sock - `docs/dev/mcp-tutorial.md` → `references/docs-dev-mcp-tutorial.md` — 从零理解 MCP —— 以 QuickQuip 项目为例 ｜ 关键词：settings.mcp.servers、llm-module.md、src/quickquip/llm/mcp/、src/quickquip/llm/config_document.py、src/quickquip/llm/service_parts/mcp_lifecycle.py、。配置权威模板为、MCPClient、mcp/client.py、ToolRegistry、mcp_* - `docs/dev/README.md` → `references/docs-dev-readme.md` — QuickQuip 开发者文档 ｜ 关键词：docs/dev/、documenting.md、architecture.md、style.md、testing.md、branching.md、versioning.md、record-identities.md、llm-configuration-api.md、llm-configuration-frontend.md diff --git a/src/quickquip/app/web/routes/logs.py b/src/quickquip/app/web/routes/logs.py index f1535ef9..a9354056 100644 --- a/src/quickquip/app/web/routes/logs.py +++ b/src/quickquip/app/web/routes/logs.py @@ -8,7 +8,7 @@ import time from fastapi import APIRouter, HTTPException, Query, Request -from fastapi.responses import FileResponse, StreamingResponse +from fastapi.responses import FileResponse, Response, StreamingResponse from quickquip.common.paths import LOGS_DIR from quickquip.llm.provider import trace_store @@ -17,6 +17,7 @@ _LOGS_DIR = LOGS_DIR _LOG_FILE_RE = re.compile(r"^quickquip_\d{4}-\d{2}-\d{2}\.log$") +_SHA256_RE = re.compile(r"^[0-9a-f]{64}$") _STREAM_POLL_SECONDS = 1.0 _STREAM_HEARTBEAT_SECONDS = 20.0 _MAX_TAIL_LINES = 500 @@ -225,10 +226,42 @@ def list_trace_calls( @router.get("/logs/trace/calls/{call_id}") -def get_trace_call(call_id: str): - """Load the complete HTTP text payload for one selected Trace call.""" - - entry = trace_store.get_call(call_id) +def get_trace_call(call_id: str, restore: bool = False): + """Load one Trace call, optionally splicing media blobs back in. + + 默认返回含 ``[qq:blob sha256=... bytes=...]`` 占位符的文本（文本量小， + 前端不必渲染数 MB JSON）；``restore=true`` 时按 manifest 回拼并校验， + 返回字节级原始报文，重放验收场景直接可用。 + + ``restore`` 刻意用普通默认值而不是 ``Query(default=False)``：后者在 + FastAPI 之外被直接调用时拿到的是 Query 对象（真值），会把默认路径走成 + restore 分支。普通 ``bool`` 默认值 FastAPI 同样按查询参数解析。 + """ + + if not restore: + entry = trace_store.get_call(call_id) + if entry is None: + raise HTTPException(status_code=404, detail="trace call not found") + return entry + try: + entry = trace_store.restore_call(call_id) + except ValueError as error: + raise HTTPException( + status_code=409, detail=f"trace restore failed: {error}" + ) from error if entry is None: raise HTTPException(status_code=404, detail="trace call not found") return entry + + +@router.get("/logs/trace/blobs/{sha256}") +def get_trace_blob(sha256: str): + """Download one content-addressed media blob by its sha256 digest.""" + + if not _SHA256_RE.match(sha256): + raise HTTPException(status_code=422, detail="invalid blob digest") + result = trace_store.download_blob(sha256) + if result is None: + raise HTTPException(status_code=404, detail="blob not found") + data, mime = result + return Response(content=data, media_type=mime) diff --git a/src/quickquip/llm/provider/media_truncator.py b/src/quickquip/llm/provider/media_truncator.py new file mode 100644 index 00000000..2eb65141 --- /dev/null +++ b/src/quickquip/llm/provider/media_truncator.py @@ -0,0 +1,444 @@ +"""内联媒体载荷的单副本存储（issue #291）。 + +trace 落库前把报文里的 base64 媒体载荷摘出来，按内容哈希写进 blob 库， +正文里只留 ``[qq:blob sha256= bytes=]`` 占位符。三个膨胀源因此 +坍缩：同一次调用重组体与原始帧各含一份同一张图 → 两个占位符指向同一 +blob；历史重放 → N 个约 80 字节的占位符；DB 增长从此只与文本量相关， +与图片数、对话长度、重放次数解耦。 + +纪律是「存坏比存大严重得多」： + +- 只截断**被引号正确闭合的完整 JSON 字符串**；中途截断、畸形输入、 + 未达门槛一律原样保留； +- 回拼先校验输入文本、再逐项校验 blob 哈希，任一不符即抛 ``ValueError``； +- 嵌套 JSON 的 ``\\"`` 场景要求开闭引号的转义形态一致，否则视为畸形。 + +本模块只做文本变换与内容寻址存储，不碰请求发送路径：发往上游的报文 +一字节不变（见 issue #291「边界与非目标」）。 +""" + +from __future__ import annotations + +import base64 +import binascii +import hashlib +import json +import logging +import os +import re +import tempfile +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Iterator + +logger = logging.getLogger(__name__) + +MEDIA_MIN_BYTES_ENV = "LLM_TRACE_MEDIA_MIN_BYTES" +DEFAULT_MIN_BYTES = 65536 +MANIFEST_VERSION = 1 +BLOB_SUFFIX = ".b64" + +_TMP_PREFIX = ".tmp-" +# 裸 base64 的廉价初筛门槛：普通短字符串（UUID、token、路径）走原样路径， +# 只有足够长的连续 base64 游程才值得进一步做完整闭合判定。 +_MIN_BARE_BASE64_CHARS = 64 +_BASE64_BODY = r"[A-Za-z0-9+/=\s]" +# data URI 只认 ``data:/;base64,`` 这一种规范形态；带 charset +# 等参数或畸形声明的形态一律不匹配（保守，宁可原样存储）。 +_DATA_URI_HEAD = r"data:([A-Za-z0-9][A-Za-z0-9.+-]*/[A-Za-z0-9][A-Za-z0-9.+-]*);base64," +_DATA_URI_PREFIX_RE = re.compile("^" + _DATA_URI_HEAD) +_DATA_URI_RE = re.compile("^" + _DATA_URI_HEAD + _BASE64_BODY + r"+$") +_BARE_BASE64_RE = re.compile(_BASE64_BODY + r"+$") +# 完整 JSON 字符串（含转义规则）：``\\.`` 吞掉转义对，因此内层 ``\"`` 不结束 +# 字符串。未闭合的尾巴不匹配任何一项，天然走「原样保留」路径。 +_STRING_RE = re.compile(r'"((?:[^"\\]|\\.)*)"') +# 字符串内容里的内层 JSON 字符串：开闭都必须是转义形态，天然一致。 +# 内容用排除式字符类而非贪婪匹配——贪婪会让正则从内层第一个 ``\"`` 一路 +# 吃到最后一个，把 ``img\": \"`` 之类前缀并进载荷，导致判定失败。内层若 +# 自带转义（base64 载荷不会），此处不匹配，按畸形原样保留。 +_ESCAPED_STRING_RE = re.compile(r'\\"([^"\\]*)\\"') + + +def resolve_min_bytes(environ: dict[str, str] | None = None) -> int: + """读取 ``LLM_TRACE_MEDIA_MIN_BYTES``；未设、非法或 <= 0 表示关闭截断。""" + + raw = (environ if environ is not None else os.environ).get(MEDIA_MIN_BYTES_ENV, "") + text = str(raw).strip() + if not text: + return 0 + try: + value = int(text) + except ValueError: + logger.warning("%s=%r 不是整数，按关闭处理（原样存储）", MEDIA_MIN_BYTES_ENV, raw) + return 0 + return max(0, value) + + +def match_payload_mime(content: str) -> str | None: + """判定字符串内容是否为 base64 载荷。 + + 返回声明的 mime（``""`` 表示未声明，如裸 base64），非载荷返回 ``None``。 + """ + + if len(content) < _MIN_BARE_BASE64_CHARS: + return None + data_uri = _DATA_URI_RE.match(content) + if data_uri is not None: + return data_uri.group(1) + if _BARE_BASE64_RE.match(content): + return "" + return None + + +def placeholder_for(sha256: str, size: int) -> str: + """占位符是纯 ASCII 且不含 JSON 需转义的字符，替换后报文仍是合法 JSON。""" + + return f"[qq:blob sha256={sha256} bytes={size}]" + + +def decode_payload(payload: str) -> tuple[bytes, str] | None: + """把 blob 里的载荷解成二进制，返回 ``(bytes, mime)``。 + + blob 存的是载荷原文（data URI 含 ``data:...;base64,`` 前缀），下载端点 + 要把它还原成可供图片预览的二进制。解码失败返回 ``None``。 + """ + + mime = "" + body = payload + prefix = _DATA_URI_PREFIX_RE.match(payload) + if prefix is not None: + mime = prefix.group(1) + body = payload[prefix.end():] + try: + return base64.b64decode(body, validate=False), mime + except (binascii.Error, ValueError): + return None + + +@dataclass(slots=True) +class Splice: + """一处替换：截断后文本中的占位符位置，以及被摘走的载荷元信息。""" + + offset: int + span: int + orig_bytes: int + sha256: str + mime: str = "" + + def as_dict(self) -> dict[str, object]: + return { + "offset": self.offset, + "span": self.span, + "orig_bytes": self.orig_bytes, + "sha256": self.sha256, + "mime": self.mime, + } + + +@dataclass(slots=True) +class TruncationOutcome: + """一次截断的结果；``manifest`` 为 ``None`` 表示原样未动。""" + + text: str + manifest: dict[str, object] | None = None + saved_bytes: int = 0 + + +@dataclass(slots=True) +class FieldManifest: + """单个文本字段的 splice 清单与全文哈希。""" + + full_sha256: str + full_bytes: int + truncated_sha256: str + splices: list[Splice] = field(default_factory=list) + + def as_dict(self) -> dict[str, object]: + return { + "full_sha256": self.full_sha256, + "full_bytes": self.full_bytes, + "truncated_sha256": self.truncated_sha256, + "splices": [splice.as_dict() for splice in self.splices], + } + + +class MediaBlobStore: + """内容寻址 blob 库：``//.b64``，同哈希天然一份。 + + 写入用「临时文件 + fsync + hardlink」：hardlink 在目标已存在时失败， + 因此并发写同一哈希不会相互覆盖，也不会产生第二份副本（rename 会覆盖， + 白白多写一次盘）。 + """ + + def __init__(self, root: str | Path) -> None: + self.root = Path(root) + + def path_for(self, sha256: str) -> Path: + return self.root / sha256[:2] / f"{sha256}{BLOB_SUFFIX}" + + def store(self, payload: str) -> tuple[str, int]: + """写入载荷并返回 ``(sha256, 字节数)``；已存在则直接复用。""" + + encoded = payload.encode("utf-8") + digest = hashlib.sha256(encoded).hexdigest() + destination = self.path_for(digest) + if destination.exists(): + return digest, len(encoded) + + destination.parent.mkdir(parents=True, exist_ok=True) + handle, tmp_name = tempfile.mkstemp(dir=str(destination.parent), prefix=_TMP_PREFIX) + tmp_path = Path(tmp_name) + try: + with os.fdopen(handle, "wb") as stream: + stream.write(encoded) + stream.flush() + os.fsync(stream.fileno()) + try: + os.link(tmp_path, destination) + except FileExistsError: + # 并发写入同一哈希：盘上已有同一份内容，无需覆盖。 + pass + finally: + tmp_path.unlink(missing_ok=True) + return digest, len(encoded) + + def load(self, sha256: str) -> str | None: + try: + return self.path_for(sha256).read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + return None + + def collect_orphaned(self, referenced: set[str], grace_seconds: float) -> int: + """mark-sweep：删除未被引用、且 mtime 早于保守带的 blob。 + + 选扫描式而非引用计数：对任意删除路径（含运维手工清库）免疫，不存在 + 计数漂移。保守带规避「blob 已写入、引用行尚未提交」的并发窗口。 + """ + + if not self.root.is_dir(): + return 0 + cutoff = time.time() - grace_seconds + removed = 0 + try: + shards = list(self.root.iterdir()) + except OSError: + return 0 + for shard in shards: + if not shard.is_dir(): + continue + try: + entries = list(shard.iterdir()) + except OSError: + continue + for entry in entries: + try: + if entry.name.startswith(_TMP_PREFIX): + # 进程崩溃留下的临时文件：同样按保守带清理。 + if entry.stat().st_mtime < cutoff: + entry.unlink() + continue + if entry.suffix != BLOB_SUFFIX: + continue + if entry.stem in referenced: + continue + if entry.stat().st_mtime >= cutoff: + continue + entry.unlink() + removed += 1 + except OSError: + logger.debug("媒体 blob 清理失败：%s", entry) + try: + if not any(shard.iterdir()): + shard.rmdir() + except OSError: + pass + return removed + + +class MediaTruncator: + """把报文文本里的内联媒体载荷换成占位符，并可回拼原文。""" + + def __init__(self, blob_store: MediaBlobStore, *, min_bytes: int = 0) -> None: + self.blob_store = blob_store + self.min_bytes = max(0, int(min_bytes)) + + @property + def enabled(self) -> bool: + return self.min_bytes > 0 + + def truncate(self, text: str) -> TruncationOutcome: + """截断一处文本；无载荷、未达门槛或畸形时原样返回。""" + + if not self.enabled or not text: + return TruncationOutcome(text=text) + + cuts: list[tuple[int, int, Splice]] = [] + for start, end, mime in self._iter_payloads(text): + payload = text[start:end] + size = len(payload.encode("utf-8")) + if size < self.min_bytes: + continue + try: + digest, _ = self.blob_store.store(payload) + except OSError: + # 落盘失败就放弃这一处截断：宁可不存，也不产出无法回拼的文本。 + logger.exception("媒体 blob 写入失败，该处按原样保留") + continue + cuts.append( + ( + start, + end, + Splice(offset=0, span=0, orig_bytes=size, sha256=digest, mime=mime), + ) + ) + + if not cuts: + return TruncationOutcome(text=text) + + pieces: list[str] = [] + written = 0 + cursor = 0 + splices: list[Splice] = [] + for start, end, splice in cuts: + if start < cursor: + # 区间重叠不应出现；防御性跳过，绝不产出错位文本。 + continue + chunk = text[cursor:start] + pieces.append(chunk) + written += len(chunk) + placeholder = placeholder_for(splice.sha256, splice.orig_bytes) + splice.offset = written + splice.span = len(placeholder) + pieces.append(placeholder) + written += len(placeholder) + splices.append(splice) + cursor = end + pieces.append(text[cursor:]) + truncated = "".join(pieces) + + manifest = FieldManifest( + full_sha256=hashlib.sha256(text.encode("utf-8")).hexdigest(), + full_bytes=len(text.encode("utf-8")), + truncated_sha256=hashlib.sha256(truncated.encode("utf-8")).hexdigest(), + splices=splices, + ) + return TruncationOutcome( + text=truncated, + manifest=manifest.as_dict(), + saved_bytes=manifest.full_bytes - len(truncated.encode("utf-8")), + ) + + def restore(self, text: str, manifest: dict[str, object]) -> str: + """按清单回拼字节级原文；任一校验不通过即抛 ``ValueError``。""" + + expected_truncated = manifest.get("truncated_sha256") + if expected_truncated: + actual = hashlib.sha256(text.encode("utf-8")).hexdigest() + if actual != expected_truncated: + raise ValueError("trace 文本与 manifest 不匹配（truncated_sha256 校验失败）") + + raw_splices = manifest.get("splices") or [] + if not isinstance(raw_splices, list): + raise ValueError("manifest.splices 不是列表") + + pieces: list[str] = [] + cursor = 0 + for item in raw_splices: + if not isinstance(item, dict): + raise ValueError("manifest.splices 条目不是对象") + offset = int(item.get("offset", -1)) + span = int(item.get("span", -1)) + digest = str(item.get("sha256", "")) + if offset < cursor or span < 0 or not digest: + raise ValueError("manifest.splices 条目越界或字段缺失") + pieces.append(text[cursor:offset]) + payload = self.blob_store.load(digest) + if payload is None: + raise ValueError(f"媒体 blob 缺失：{digest}") + if hashlib.sha256(payload.encode("utf-8")).hexdigest() != digest: + raise ValueError(f"媒体 blob 哈希校验失败：{digest}") + pieces.append(payload) + cursor = offset + span + pieces.append(text[cursor:]) + restored = "".join(pieces) + + expected_full = manifest.get("full_sha256") + if expected_full: + actual = hashlib.sha256(restored.encode("utf-8")).hexdigest() + if actual != expected_full: + raise ValueError("回拼结果与 full_sha256 不符") + return restored + + def _iter_payloads(self, text: str) -> Iterator[tuple[int, int, str]]: + """产出可替换载荷的 ``(start, end, mime)``，按文本位置升序。""" + + for match in _STRING_RE.finditer(text): + content_start, content_end = match.start(1), match.end(1) + content = match.group(1) + mime = match_payload_mime(content) + if mime is not None: + yield content_start, content_end, mime + continue + if '\\"' not in content: + continue + for inner in _ESCAPED_STRING_RE.finditer(content): + inner_mime = match_payload_mime(inner.group(1)) + if inner_mime is not None: + yield ( + content_start + inner.start(1), + content_start + inner.end(1), + inner_mime, + ) + + +def _iter_field_manifests(payload: dict[str, object] | None) -> Iterator[dict[str, object]]: + """兼容两种落库形态，逐个产出字段级 manifest。 + + - 单字段：``truncate`` 的直接产出，顶层就带 ``splices``； + - 分字段包装：``{"v": 1, "request": {...}, "response": {...}}``， + 这是 trace 落库时把三个文本字段的清单合成一列后的形态。 + """ + + if payload is None: + return + if "splices" in payload: + yield payload + return + for value in payload.values(): + if isinstance(value, dict) and "splices" in value: + yield value + + +def collect_manifest_shas(manifest_json: str | None) -> set[str]: + """从落库的 manifest JSON 里收集全部 blob sha（GC 的 mark 集合）。""" + + collected: set[str] = set() + for field_manifest in _iter_field_manifests(_load_manifest(manifest_json)): + for item in field_manifest.get("splices") or []: + if isinstance(item, dict): + digest = item.get("sha256") + if isinstance(digest, str) and digest: + collected.add(digest) + return collected + + +def find_manifest_mime(manifest_json: str | None, sha256: str) -> str | None: + """在 manifest JSON 里查某个 blob 的 mime。""" + + for field_manifest in _iter_field_manifests(_load_manifest(manifest_json)): + for item in field_manifest.get("splices") or []: + if isinstance(item, dict) and item.get("sha256") == sha256: + mime = item.get("mime") + return mime if isinstance(mime, str) else "" + return None + + +def _load_manifest(manifest_json: str | None) -> dict[str, object] | None: + if not manifest_json: + return None + try: + payload = json.loads(manifest_json) + except (ValueError, TypeError): + return None + return payload if isinstance(payload, dict) else None diff --git a/src/quickquip/llm/provider/trace.py b/src/quickquip/llm/provider/trace.py index 72d1cdcd..d15e6a46 100644 --- a/src/quickquip/llm/provider/trace.py +++ b/src/quickquip/llm/provider/trace.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio +import json from contextlib import contextmanager from contextvars import ContextVar from quickquip.llm.session import SessionIdentity @@ -18,12 +19,24 @@ import logging from quickquip.common.paths import LLM_TRACE_DB_PATH, LOGS_DIR +from quickquip.llm.provider.media_truncator import ( + MANIFEST_VERSION, + MediaBlobStore, + MediaTruncator, + collect_manifest_shas, + decode_payload, + find_manifest_mime, + resolve_min_bytes, +) logger = logging.getLogger(__name__) _TRACE_FLAG_FILE = os.getenv("LLM_TRACE_FLAG_FILE", "") _TRACE_RETENTION_DAYS = 14 +_TRACE_BLOB_DIR_NAME = "trace_blobs" +# GC 保守带：blob 先落盘、引用行后提交，清扫必须避开这个并发窗口。 +_TRACE_BLOB_GRACE_SECONDS = 24 * 3600 _LEGACY_TRACE_GLOB = "quickquip_trace_????-??-??.jsonl" _TRACE_STALE_AFTER = timedelta(hours=1) _TRACE_STALE_CHECK_INTERVAL_SECONDS = 60 @@ -37,6 +50,42 @@ def _utc_now() -> str: return datetime.now(timezone.utc).isoformat() +def _parse_manifest(raw: str | None) -> dict[str, object] | None: + """解析落库的 manifest JSON；非法或为空返回 ``None``（视作原样存储）。""" + + if not raw: + return None + try: + payload = json.loads(raw) + except (ValueError, TypeError): + return None + return payload if isinstance(payload, dict) else None + + +def _merge_field_manifests( + existing_json: str | None, + additions: dict[str, dict[str, object] | None], +) -> str | None: + """把新字段的清单并入已有 manifest，避免后写的一次覆盖前一次。 + + ``begin_call`` 落 request，``finish_call`` 落 response 与 response_raw， + 三者共用一个 ``media_truncation`` 列，必须合并而非覆盖。全部字段都无 + 截断时返回 ``None``——该行就是「原样存储」。 + """ + + payload = _parse_manifest(existing_json) or {} + changed = False + for field_name, manifest in additions.items(): + if manifest is None: + continue + payload[field_name] = manifest + changed = True + if not changed: + return existing_json or None + payload["v"] = MANIFEST_VERSION + return json.dumps(payload, ensure_ascii=False) + + @dataclass(slots=True) class TraceCapture: """Collect provider call IDs for one explicitly observed operation.""" @@ -131,6 +180,8 @@ def __init__( path: str | Path = LLM_TRACE_DB_PATH, *, legacy_trace_dir: str | Path | None = None, + blob_dir: str | Path | None = None, + media_min_bytes: int | None = None, ): self.path = Path(path) self.legacy_trace_dir = ( @@ -138,6 +189,18 @@ def __init__( if legacy_trace_dir is not None else self.path.parent / "logs" ) + # blob 库刻意不按天分桶：滚动留存下，全局内容寻址库才能真正做到 + # 跨天严格单副本（issue #291）。min_bytes <= 0 表示截断关闭。 + self.blob_dir = ( + Path(blob_dir) + if blob_dir is not None + else self.path.parent / _TRACE_BLOB_DIR_NAME + ) + self.blob_store = MediaBlobStore(self.blob_dir) + self.truncator = MediaTruncator( + self.blob_store, + min_bytes=resolve_min_bytes() if media_min_bytes is None else media_min_bytes, + ) self._schema_ready = False self._schema_lock = threading.Lock() self._cleanup_lock = threading.Lock() @@ -209,6 +272,7 @@ def _ensure_schema(self) -> None: response_bytes INTEGER NOT NULL DEFAULT 0, response_raw_text TEXT, response_raw_bytes INTEGER NOT NULL DEFAULT 0, + media_truncation TEXT, duration_ms REAL, state TEXT NOT NULL, error_type TEXT, @@ -293,6 +357,11 @@ def _ensure_schema(self) -> None: "ALTER TABLE llm_http_traces ADD COLUMN response_raw_bytes " "INTEGER NOT NULL DEFAULT 0" ) + if "media_truncation" not in trace_columns: + # 旧行保持 NULL，读端视作「原样存储」，天然兼容。 + conn.execute( + "ALTER TABLE llm_http_traces ADD COLUMN media_truncation TEXT" + ) self._schema_ready = True @staticmethod @@ -340,14 +409,19 @@ def begin_call( self._ensure_schema() self._cleanup_if_due() call_id = uuid4().hex + # 唯一写入咽喉之一：三个大文本在落库前过一遍截断器，provider 层零改动。 + # request_bytes 仍记原始字节数（等同 full_bytes），语义不变。 + outcome = self.truncator.truncate(request_text) + manifest_json = _merge_field_manifests(None, {"request": outcome.manifest}) with self._connect() as conn: cursor = conn.execute( """ INSERT INTO llm_http_traces ( call_id, agent_loop_id, loop_sequence, started_at, provider_id, protocol, model, stream, - method, url, request_headers, request_text, request_bytes, session_id, state - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending') + method, url, request_headers, request_text, request_bytes, session_id, + media_truncation, state + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending') """, ( call_id, @@ -361,9 +435,10 @@ def begin_call( method, url, request_headers, - request_text, + outcome.text, request_bytes, session_id, + manifest_json, ), ) conn.execute( @@ -391,13 +466,31 @@ def finish_call( response_raw_bytes: int = 0, ) -> None: self._ensure_schema() + # 重组响应与原始 SSE 帧各含一份同样的 base64：两处截断后同哈希指向 + # 同一个 blob，这正是「同一次调用双份」膨胀源的坍缩点。 + response_outcome = self.truncator.truncate(response_text or "") + raw_outcome = self.truncator.truncate(response_raw_text or "") with self._connect() as conn: - cursor = conn.execute( + existing = conn.execute( + "SELECT id, media_truncation FROM llm_http_traces WHERE call_id = ?", + (call_id,), + ).fetchone() + if existing is None: + return + manifest_json = _merge_field_manifests( + existing["media_truncation"], + { + "response": response_outcome.manifest, + "response_raw": raw_outcome.manifest, + }, + ) + conn.execute( """ UPDATE llm_http_traces SET completed_at = ?, response_status = ?, response_headers = ?, response_text = ?, response_bytes = ?, duration_ms = ?, response_raw_text = ?, response_raw_bytes = ?, + media_truncation = ?, state = ?, error_type = ?, error_message = ? WHERE call_id = ? """, @@ -405,30 +498,25 @@ def finish_call( _utc_now(), response_status, response_headers, - response_text, + response_outcome.text, response_bytes, duration_ms, - response_raw_text or None, + raw_outcome.text or None, response_raw_bytes, + manifest_json, state, error_type or None, error_message or None, call_id, ), ) - if cursor.rowcount: - row = conn.execute( - "SELECT id FROM llm_http_traces WHERE call_id = ?", - (call_id,), - ).fetchone() - if row is not None: - conn.execute( - """ - INSERT INTO llm_http_trace_events (trace_id, state, created_at) - VALUES (?, ?, ?) - """, - (int(row["id"]), state, _utc_now()), - ) + conn.execute( + """ + INSERT INTO llm_http_trace_events (trace_id, state, created_at) + VALUES (?, ?, ?) + """, + (int(existing["id"]), state, _utc_now()), + ) def list_calls( self, @@ -487,8 +575,74 @@ def get_call(self, call_id: str) -> dict[str, object] | None: "response_headers": row["response_headers"] or "", "response_text": row["response_text"] or "", "response_raw_text": row["response_raw_text"] or "", + "media_truncation": _parse_manifest(row["media_truncation"]), } + def restore_call(self, call_id: str) -> dict[str, object] | None: + """按 manifest 回拼字节级原始报文；无清单的行原样返回。 + + 校验不通过时抛 ``ValueError``——调用方据此判断「存坏了」，而不是把 + 残缺内容当成原始报文交付。 + """ + + entry = self.get_call(call_id) + if entry is None: + return None + result: dict[str, object] = dict(entry) + manifest = entry.get("media_truncation") + if not isinstance(manifest, dict): + return result + for field_name, key in ( + ("request_text", "request"), + ("response_text", "response"), + ("response_raw_text", "response_raw"), + ): + field_manifest = manifest.get(key) + if not isinstance(field_manifest, dict): + continue + try: + result[field_name] = self.truncator.restore( + str(entry[field_name]), field_manifest + ) + except ValueError: + logger.exception( + "Trace 媒体回拼校验失败：call_id=%s field=%s", call_id, key + ) + raise + return result + + def find_blob_mime(self, sha256: str) -> str | None: + """反查某个 blob 的 mime，供下载端点作 Content-Type。""" + + self._ensure_schema() + with self._connect() as conn: + rows = conn.execute( + """ + SELECT media_truncation FROM llm_http_traces + WHERE media_truncation LIKE ? + LIMIT 20 + """, + (f"%{sha256}%",), + ).fetchall() + for row in rows: + mime = find_manifest_mime(row["media_truncation"], sha256) + if mime is not None: + return mime + return None + + def download_blob(self, sha256: str) -> tuple[bytes, str] | None: + """取出 blob 并解成二进制；返回 ``(bytes, mime)``，缺失返回 ``None``。""" + + payload = self.blob_store.load(sha256) + if payload is None: + return None + decoded = decode_payload(payload) + if decoded is None: + return None + data, mime = decoded + resolved = mime or self.find_blob_mime(sha256) or "" + return data, resolved or "application/octet-stream" + def list_events( self, *, @@ -601,6 +755,8 @@ def _cleanup_if_due(self) -> None: ) conn.execute("DELETE FROM llm_http_traces WHERE started_at < ?", (cutoff,)) self._cleanup_legacy_trace_files((now - retention).timestamp()) + # 先 DELETE 过期行再 sweep：此刻的引用集合才是准确的 mark。 + self._collect_orphaned_blobs() self._last_cleanup_date = today def _cleanup_legacy_trace_files(self, cutoff_timestamp: float) -> None: @@ -617,6 +773,29 @@ def _cleanup_legacy_trace_files(self, cutoff_timestamp: float) -> None: except OSError: logger.debug("Failed to clean legacy LLM trace file: %s", path) + def _collect_orphaned_blobs(self) -> None: + """mark-sweep 清扫无引用 blob；挂在每日留存清理末尾。 + + 选扫描式而非引用计数：对任意删除路径（含运维手工清库）免疫， + 不存在计数漂移。24h 保守带规避「blob 已写入、引用行尚未提交」。 + """ + + try: + with self._connect() as conn: + rows = conn.execute( + "SELECT media_truncation FROM llm_http_traces " + "WHERE media_truncation IS NOT NULL" + ).fetchall() + except sqlite3.Error: + logger.exception("LLM trace blob 引用扫描失败") + return + referenced: set[str] = set() + for row in rows: + referenced |= collect_manifest_shas(row["media_truncation"]) + removed = self.blob_store.collect_orphaned(referenced, _TRACE_BLOB_GRACE_SECONDS) + if removed: + logger.info("LLM trace blob GC 已清理 %d 个无引用 blob", removed) + def _expire_stale_pending_if_due(self) -> None: checked_at = time.monotonic() if ( diff --git a/tests/unit/llm/provider/__init__.py b/tests/unit/llm/provider/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/unit/llm/provider/test_media_truncator.py b/tests/unit/llm/provider/test_media_truncator.py new file mode 100644 index 00000000..0b9305ea --- /dev/null +++ b/tests/unit/llm/provider/test_media_truncator.py @@ -0,0 +1,205 @@ +"""内联媒体载荷单副本存储的单元测试（issue #291）。 + +覆盖：裸 base64 与 data URI 两种检测形态、嵌套 ``\\"`` 转义边界、畸形 +输入原样保留、``min_bytes`` 门槛、同内容去重、回拼双重校验、GC 引用 +集合与保守带、环境开关。 +""" + +from __future__ import annotations + +import base64 +import json + +import pytest + +from quickquip.llm.provider.media_truncator import ( + MediaBlobStore, + MediaTruncator, + collect_manifest_shas, + decode_payload, + find_manifest_mime, + resolve_min_bytes, +) + +_MIN_BYTES = 1024 + + +def _truncator(tmp_path, *, min_bytes: int = _MIN_BYTES) -> MediaTruncator: + return MediaTruncator(MediaBlobStore(tmp_path / "blobs"), min_bytes=min_bytes) + + +def _text(payload: str, prefix: str = "") -> str: + return '{"image": "' + prefix + payload + '"}' + + +def test_bare_base64_is_replaced_and_restores_byte_identically(tmp_path): + truncator = _truncator(tmp_path) + payload = "A" * 2000 + original = _text(payload) + + outcome = truncator.truncate(original) + + assert outcome.manifest is not None + assert payload not in outcome.text + assert outcome.saved_bytes > 0 + assert json.loads(outcome.text)["image"].startswith("[qq:blob sha256=") + assert truncator.restore(outcome.text, outcome.manifest) == original + + +def test_data_uri_mime_is_captured(tmp_path): + truncator = _truncator(tmp_path) + payload = "B" * 2000 + original = _text(payload, "data:image/png;base64,") + + outcome = truncator.truncate(original) + + assert outcome.manifest is not None + assert outcome.manifest["splices"][0]["mime"] == "image/png" + assert truncator.restore(outcome.text, outcome.manifest) == original + + +def test_payload_below_threshold_is_kept_as_is(tmp_path): + truncator = _truncator(tmp_path) + original = _text("C" * 200) + + outcome = truncator.truncate(original) + + assert outcome.manifest is None + assert outcome.text == original + + +def test_unterminated_string_is_kept_as_is(tmp_path): + truncator = _truncator(tmp_path) + original = '{"image": "' + "D" * 2000 + + outcome = truncator.truncate(original) + + assert outcome.manifest is None + assert outcome.text == original + + +def test_nested_escaped_json_payload_is_replaced(tmp_path): + truncator = _truncator(tmp_path) + payload = "E" * 2000 + original = '{"body": "{\\"image\\": \\"data:image/jpeg;base64,' + payload + '\\"}"}' + + outcome = truncator.truncate(original) + + assert outcome.manifest is not None + assert payload not in outcome.text + assert outcome.manifest["splices"][0]["mime"] == "image/jpeg" + assert isinstance(json.loads(outcome.text), dict) + assert truncator.restore(outcome.text, outcome.manifest) == original + + +def test_same_payload_is_stored_once(tmp_path): + truncator = _truncator(tmp_path) + payload = "F" * 2000 + original = '{"a": "' + payload + '", "b": "' + payload + '"}' + + outcome = truncator.truncate(original) + + splices = outcome.manifest["splices"] + assert len(splices) == 2 + assert splices[0]["sha256"] == splices[1]["sha256"] + blob_dir = truncator.blob_store.path_for(splices[0]["sha256"]).parent + assert len(list(blob_dir.glob("*.b64"))) == 1 + assert truncator.restore(outcome.text, outcome.manifest) == original + + +def test_splice_offsets_point_at_placeholders(tmp_path): + truncator = _truncator(tmp_path) + original = '{"a": "' + "L" * 2000 + '", "b": "' + "M" * 2000 + '"}' + + outcome = truncator.truncate(original) + + assert len(outcome.manifest["splices"]) == 2 + for splice in outcome.manifest["splices"]: + segment = outcome.text[splice["offset"]:splice["offset"] + splice["span"]] + assert segment.startswith("[qq:blob sha256=") + assert segment.endswith("]") + + +def test_placeholder_needs_no_json_escaping(tmp_path): + truncator = _truncator(tmp_path) + + outcome = truncator.truncate(_text("K" * 2000)) + + placeholder = json.loads(outcome.text)["image"] + assert placeholder.endswith("]") + assert '"' not in placeholder + assert "\\" not in placeholder + + +def test_restore_rejects_tampered_text(tmp_path): + truncator = _truncator(tmp_path) + outcome = truncator.truncate(_text("G" * 2000)) + + with pytest.raises(ValueError): + truncator.restore(outcome.text.replace("bytes=", "bytes =", 1), outcome.manifest) + + +def test_restore_rejects_tampered_blob(tmp_path): + truncator = _truncator(tmp_path) + outcome = truncator.truncate(_text("H" * 2000)) + digest = outcome.manifest["splices"][0]["sha256"] + truncator.blob_store.path_for(digest).write_text("tampered", encoding="utf-8") + + with pytest.raises(ValueError): + truncator.restore(outcome.text, outcome.manifest) + + +def test_disabled_truncator_never_touches_text(tmp_path): + truncator = _truncator(tmp_path, min_bytes=0) + original = _text("I" * 5000) + + outcome = truncator.truncate(original) + + assert outcome.manifest is None + assert outcome.text == original + + +def test_manifest_helpers_accept_field_and_wrapped_shapes(tmp_path): + truncator = _truncator(tmp_path) + outcome = truncator.truncate(_text("J" * 2000, "data:image/webp;base64,")) + field_manifest = outcome.manifest + digest = field_manifest["splices"][0]["sha256"] + wrapped = {"v": 1, "request": field_manifest} + + assert collect_manifest_shas(json.dumps(field_manifest)) == {digest} + assert collect_manifest_shas(json.dumps(wrapped)) == {digest} + assert find_manifest_mime(json.dumps(wrapped), digest) == "image/webp" + assert collect_manifest_shas("{not json") == set() + assert find_manifest_mime(None, digest) is None + + +def test_blob_gc_keeps_referenced_and_respects_grace_band(tmp_path): + store = MediaBlobStore(tmp_path / "blobs") + kept, _ = store.store("keep-me") + dropped, _ = store.store("drop-me") + + assert store.collect_orphaned({kept}, grace_seconds=0) == 1 + assert store.path_for(kept).exists() + assert not store.path_for(dropped).exists() + + # 保守带内即使无引用也不删：规避「blob 已写入、引用行尚未提交」窗口。 + assert store.collect_orphaned(set(), grace_seconds=3600) == 0 + assert store.path_for(kept).exists() + + +def test_decode_payload_strips_data_uri_prefix(tmp_path): + raw = b"\x89PNG\r\n\x1a\npayload" + encoded = base64.b64encode(raw).decode("ascii") + + assert decode_payload(f"data:image/png;base64,{encoded}") == (raw, "image/png") + assert decode_payload(encoded) == (raw, "") + assert decode_payload("AAA") is None + + +def test_resolve_min_bytes_reads_env_switch(): + assert resolve_min_bytes({}) == 0 + assert resolve_min_bytes({"LLM_TRACE_MEDIA_MIN_BYTES": ""}) == 0 + assert resolve_min_bytes({"LLM_TRACE_MEDIA_MIN_BYTES": "0"}) == 0 + assert resolve_min_bytes({"LLM_TRACE_MEDIA_MIN_BYTES": "65536"}) == 65536 + assert resolve_min_bytes({"LLM_TRACE_MEDIA_MIN_BYTES": "abc"}) == 0 + assert resolve_min_bytes({"LLM_TRACE_MEDIA_MIN_BYTES": "-5"}) == 0 From 194ad33c1e184d59866ec943b06e66f54a5d9c47 Mon Sep 17 00:00:00 2001 From: 3aKHP <178400596+3aKHP@users.noreply.github.com> Date: Fri, 23 Oct 2026 22:17:54 +0800 Subject: [PATCH 2/2] fix(trace): harden media storage lifecycle Coordinate blob writes and GC across processes, enforce attachment downloads, and preserve ordinary text with protocol-aware extraction. Validate restoration manifests, bound parser memory, clear orphan media, and cover storage and HTTP integration paths. --- docs/admin/configuration.md | 12 +- docs/dev/llm-module.md | 8 +- .../references/docs-admin-configuration.md | 12 +- .../references/docs-dev-llm-module.md | 8 +- skills.example/self-docs/references/index.md | 2 +- src/quickquip/app/web/routes/logs.py | 14 +- src/quickquip/llm/provider/media_truncator.py | 291 ++++++++++++++---- src/quickquip/llm/provider/trace.py | 103 +++---- .../unit/llm/provider/test_media_truncator.py | 76 ++++- tests/unit/test_trace_store.py | 168 ++++++++++ tests/unit/web/test_logs_routes.py | 92 ++++++ 11 files changed, 644 insertions(+), 142 deletions(-) diff --git a/docs/admin/configuration.md b/docs/admin/configuration.md index cb03db5d..250b2aad 100644 --- a/docs/admin/configuration.md +++ b/docs/admin/configuration.md @@ -56,7 +56,17 @@ LLM 工具 `search_web` 与 `/search` 命令固定走项目内 SearXNG。普通 | 变量 | 说明 | 默认值 | |------|------|--------| | `LLM_TRACE_FLAG_FILE` | LLM HTTP Trace 持久开关文件路径；文件存在时按调用记录完整请求/响应 JSON 文本到 `data/llm_trace.db`，供 Web Admin 的 LLM Trace 页面按需读取 | — | -| `LLM_TRACE_MEDIA_MIN_BYTES` | 内联 base64 媒体载荷的截断门槛（字节）。达到门槛、且载荷构成被引号正确闭合的完整 JSON 字符串时，载荷本体写入 `data/trace_blobs/` 下的内容寻址副本，正文替换为 `[qq:blob sha256= bytes=]` 占位符；同一份载荷在请求、重组响应与原始 SSE 中只落盘一份。未设置或为 `0` 时原样存储。进程启动时读取 | `0` | +| `LLM_TRACE_MEDIA_MIN_BYTES` | Trace 内联媒体抽取门槛（载荷原文 UTF-8 字节数），推荐启用值 `65536`；未设置或为 `0` 时关闭。Bot 和 Web 在启动时读取。操作见下方“Trace 媒体存储” | `0` | + +### Trace 媒体存储 + +Trace 开启期间，可在根 `.env` 设置 `LLM_TRACE_MEDIA_MIN_BYTES=65536` 并重启 Bot 与 Web，对新记录中达到门槛的内联媒体去重。正文显示占位符，原始载荷保存在 `data/trace_blobs/`；既有记录继续按原格式读取。完整 JSON 与 SSE 单行 JSON 中的 data URI 和明确媒体字段受支持。 + +备份与恢复时，将 `data/llm_trace.db` 和 `data/trace_blobs/` 作为一组：停服后同时备份，恢复时同时还原。部署模板挂载整个 `data/`，无需额外挂载。不要把多个独立 Trace 数据库指向同一个 blob 目录。 + +设置为 `0` 并重启会停止新记录的媒体抽取，当前版本仍能恢复已抽取记录。回退到不支持该清单格式的旧版本时，旧程序会显示占位符；需要完整历史报文时，应先在支持此功能的版本中导出原文，或恢复启用前的 Trace 备份。已有大记录的批量转换另行安排。 + +管理员可通过 `/ops/api/logs/trace/calls/{call_id}?restore=true` 获取经校验的完整正文；`call_id` 使用详情记录中的实际 ID。`/ops/api/logs/trace/blobs/{sha256}` 使用占位符中的摘要下载媒体附件，两个接口均要求有效的后台登录态。文件缺失或恢复校验失败时，先检查数据库与 blob 目录是否来自同一备份。清空 Trace 会移除记录并清理无引用媒体文件；下载只开放给仍被记录引用的载荷。 ### 贴吧 diff --git a/docs/dev/llm-module.md b/docs/dev/llm-module.md index 9bcf5e08..d0ab7181 100644 --- a/docs/dev/llm-module.md +++ b/docs/dev/llm-module.md @@ -12,7 +12,13 @@ QuickQuip 的 LLM 模块为规则机器人提供对话扩展。管理员按群 LLM 运行时在 `LLM_TRACE_FLAG_FILE` 指向的开关文件存在时，把每次 HTTP 尝试写入 `data/llm_trace.db`。请求正文取自实际交给 HTTP 客户端的 UTF-8 JSON 序列化文本；普通响应保留 JSON 解析前的服务端文本；流式响应完整消费 SSE 后，由协议客户端重建 OpenAI Chat Completion、Claude Message、Gemini GenerateContent 或 OpenAI Responses 完整响应对象，同时保留 SSE 传输原文供管理员按需核对。索引、正文和单调递增的状态事件分开存储，Web Admin 先读取轻量调用元数据，管理员选择记录后再加载完整 Header 与正文。`run_tool_call_loop` 为一轮完整交互分配 Agent Loop ID，重试、故障切换和工具结果回送产生的 HTTP 调用按组内序号排列。 -多模态消息的内联 base64 载荷在落库前按内容寻址去重：`LLM_TRACE_MEDIA_MIN_BYTES` 达到门槛的载荷写入 `data/trace_blobs/`，正文只留 `[qq:blob sha256= bytes=]` 占位符，同一份载荷在请求、重组响应与原始 SSE 三处共用一个副本。每次截断把 splice 清单与全文哈希记入 `media_truncation` 列；读端默认返回占位符文本，`restore` 参数按清单回拼并逐项校验。每日留存清理之后，按剩余行的引用集合 mark-sweep 未被引用的 blob。截断只发生在落库侧，发往上游的报文不变；未达门槛、畸形或被截断的字符串一律原样存储。配置键与默认值见[管理员配置](../admin/configuration.md)。 +启用 `LLM_TRACE_MEDIA_MIN_BYTES` 后，Trace 会在入库前抽取达到门槛的内联 base64 载荷。识别范围为完整 JSON、SSE 的完整单行 JSON data 记录及其中最多四层嵌套 JSON 字符串；data URI 须通过严格 base64 校验，裸编码还须处于明确的媒体字段，例如 `b64_json`、`partial_image_b64`、带媒体类型的 `data` 或图片生成结果。普通正文、未达门槛、畸形及不支持的编码形态按原文保存。 + +载荷原文按 SHA-256 存入 `data/trace_blobs/`，正文替换为 `[qq:blob sha256= bytes=]`。完全相同的载荷文本跨请求、重组响应和原始 SSE 共用一个文件；data URI 前缀参与哈希。每个正文的替换位置和哈希保存在同一条记录的 `media_truncation` 清单中，原始字节计数保持原义。请求发送路径使用原始载荷。 + +每份 blob 目录由一份 Trace 数据库拥有。Bot 和 Web 通过该数据库的 `BEGIN IMMEDIATE` 写事务协调载荷创建／复用、引用提交，以及 GC 引用扫描到文件删除的整个区间。每日留存清理后删除无引用且超过一天的文件；文件年龄用于保留失败写入的恢复窗口，并发安全由事务保证。清单无效时跳过该次文件清理。显式清空会删除记录并清理无引用文件，下载入口同时检查记录引用。 + +管理员详情接口默认返回含占位符的正文，`restore=true` 按清单恢复并核验全文与载荷哈希；文件缺失或校验失败返回 409。blob 下载继承管理员鉴权，统一返回 `application/octet-stream` 附件，并设置 `nosniff` 与限制性 CSP。抽取开关关闭后仍能恢复已抽取记录。配置、备份及升级操作见[管理员配置](../admin/configuration.md#trace-媒体存储)。 ### 1.1 执行记录的请求边界 diff --git a/skills.example/self-docs/references/docs-admin-configuration.md b/skills.example/self-docs/references/docs-admin-configuration.md index 9476cbfd..8d7fa667 100644 --- a/skills.example/self-docs/references/docs-admin-configuration.md +++ b/skills.example/self-docs/references/docs-admin-configuration.md @@ -58,7 +58,17 @@ LLM 工具 `search_web` 与 `/search` 命令固定走项目内 SearXNG。普通 | 变量 | 说明 | 默认值 | |------|------|--------| | `LLM_TRACE_FLAG_FILE` | LLM HTTP Trace 持久开关文件路径；文件存在时按调用记录完整请求/响应 JSON 文本到 `data/llm_trace.db`，供 Web Admin 的 LLM Trace 页面按需读取 | — | -| `LLM_TRACE_MEDIA_MIN_BYTES` | 内联 base64 媒体载荷的截断门槛（字节）。达到门槛、且载荷构成被引号正确闭合的完整 JSON 字符串时，载荷本体写入 `data/trace_blobs/` 下的内容寻址副本，正文替换为 `[qq:blob sha256= bytes=]` 占位符；同一份载荷在请求、重组响应与原始 SSE 中只落盘一份。未设置或为 `0` 时原样存储。进程启动时读取 | `0` | +| `LLM_TRACE_MEDIA_MIN_BYTES` | Trace 内联媒体抽取门槛（载荷原文 UTF-8 字节数），推荐启用值 `65536`；未设置或为 `0` 时关闭。Bot 和 Web 在启动时读取。操作见下方“Trace 媒体存储” | `0` | + +### Trace 媒体存储 + +Trace 开启期间，可在根 `.env` 设置 `LLM_TRACE_MEDIA_MIN_BYTES=65536` 并重启 Bot 与 Web，对新记录中达到门槛的内联媒体去重。正文显示占位符，原始载荷保存在 `data/trace_blobs/`；既有记录继续按原格式读取。完整 JSON 与 SSE 单行 JSON 中的 data URI 和明确媒体字段受支持。 + +备份与恢复时，将 `data/llm_trace.db` 和 `data/trace_blobs/` 作为一组：停服后同时备份，恢复时同时还原。部署模板挂载整个 `data/`，无需额外挂载。不要把多个独立 Trace 数据库指向同一个 blob 目录。 + +设置为 `0` 并重启会停止新记录的媒体抽取，当前版本仍能恢复已抽取记录。回退到不支持该清单格式的旧版本时，旧程序会显示占位符；需要完整历史报文时，应先在支持此功能的版本中导出原文，或恢复启用前的 Trace 备份。已有大记录的批量转换另行安排。 + +管理员可通过 `/ops/api/logs/trace/calls/{call_id}?restore=true` 获取经校验的完整正文；`call_id` 使用详情记录中的实际 ID。`/ops/api/logs/trace/blobs/{sha256}` 使用占位符中的摘要下载媒体附件，两个接口均要求有效的后台登录态。文件缺失或恢复校验失败时，先检查数据库与 blob 目录是否来自同一备份。清空 Trace 会移除记录并清理无引用媒体文件；下载只开放给仍被记录引用的载荷。 ### 贴吧 diff --git a/skills.example/self-docs/references/docs-dev-llm-module.md b/skills.example/self-docs/references/docs-dev-llm-module.md index 50f2c7bd..a9e308ed 100644 --- a/skills.example/self-docs/references/docs-dev-llm-module.md +++ b/skills.example/self-docs/references/docs-dev-llm-module.md @@ -14,7 +14,13 @@ QuickQuip 的 LLM 模块为规则机器人提供对话扩展。管理员按群 LLM 运行时在 `LLM_TRACE_FLAG_FILE` 指向的开关文件存在时，把每次 HTTP 尝试写入 `data/llm_trace.db`。请求正文取自实际交给 HTTP 客户端的 UTF-8 JSON 序列化文本；普通响应保留 JSON 解析前的服务端文本；流式响应完整消费 SSE 后，由协议客户端重建 OpenAI Chat Completion、Claude Message、Gemini GenerateContent 或 OpenAI Responses 完整响应对象，同时保留 SSE 传输原文供管理员按需核对。索引、正文和单调递增的状态事件分开存储，Web Admin 先读取轻量调用元数据，管理员选择记录后再加载完整 Header 与正文。`run_tool_call_loop` 为一轮完整交互分配 Agent Loop ID，重试、故障切换和工具结果回送产生的 HTTP 调用按组内序号排列。 -多模态消息的内联 base64 载荷在落库前按内容寻址去重：`LLM_TRACE_MEDIA_MIN_BYTES` 达到门槛的载荷写入 `data/trace_blobs/`，正文只留 `[qq:blob sha256= bytes=]` 占位符，同一份载荷在请求、重组响应与原始 SSE 三处共用一个副本。每次截断把 splice 清单与全文哈希记入 `media_truncation` 列；读端默认返回占位符文本，`restore` 参数按清单回拼并逐项校验。每日留存清理之后，按剩余行的引用集合 mark-sweep 未被引用的 blob。截断只发生在落库侧，发往上游的报文不变；未达门槛、畸形或被截断的字符串一律原样存储。配置键与默认值见[管理员配置](../admin/configuration.md)。 +启用 `LLM_TRACE_MEDIA_MIN_BYTES` 后，Trace 会在入库前抽取达到门槛的内联 base64 载荷。识别范围为完整 JSON、SSE 的完整单行 JSON data 记录及其中最多四层嵌套 JSON 字符串；data URI 须通过严格 base64 校验，裸编码还须处于明确的媒体字段，例如 `b64_json`、`partial_image_b64`、带媒体类型的 `data` 或图片生成结果。普通正文、未达门槛、畸形及不支持的编码形态按原文保存。 + +载荷原文按 SHA-256 存入 `data/trace_blobs/`，正文替换为 `[qq:blob sha256= bytes=]`。完全相同的载荷文本跨请求、重组响应和原始 SSE 共用一个文件；data URI 前缀参与哈希。每个正文的替换位置和哈希保存在同一条记录的 `media_truncation` 清单中，原始字节计数保持原义。请求发送路径使用原始载荷。 + +每份 blob 目录由一份 Trace 数据库拥有。Bot 和 Web 通过该数据库的 `BEGIN IMMEDIATE` 写事务协调载荷创建／复用、引用提交，以及 GC 引用扫描到文件删除的整个区间。每日留存清理后删除无引用且超过一天的文件；文件年龄用于保留失败写入的恢复窗口，并发安全由事务保证。清单无效时跳过该次文件清理。显式清空会删除记录并清理无引用文件，下载入口同时检查记录引用。 + +管理员详情接口默认返回含占位符的正文，`restore=true` 按清单恢复并核验全文与载荷哈希；文件缺失或校验失败返回 409。blob 下载继承管理员鉴权，统一返回 `application/octet-stream` 附件，并设置 `nosniff` 与限制性 CSP。抽取开关关闭后仍能恢复已抽取记录。配置、备份及升级操作见[管理员配置](../admin/configuration.md#trace-媒体存储)。 ### 1.1 执行记录的请求边界 diff --git a/skills.example/self-docs/references/index.md b/skills.example/self-docs/references/index.md index fbf720ac..0f4f8180 100644 --- a/skills.example/self-docs/references/index.md +++ b/skills.example/self-docs/references/index.md @@ -67,7 +67,7 @@ - `docs/dev/game-framework.md` → `references/docs-dev-game-framework.md` — 游戏框架开发者指南 ｜ 关键词：BaseGame、aliases、src/quickquip/app/message_pipeline.py、GameResult、finished、group_id、OrderedDict[str, Session]、src/quickquip/adapters/nonebot/commands.py、command_parts/、_ensure_schema() - `docs/dev/llm-configuration-api.md` → `references/docs-dev-llm-configuration-api.md` — LLM 配置后端接口 ｜ 关键词：data/llm_config.db、/ops/api/llm-configuration、quickquip_admin_session、Origin、Referer、在进程内构造应用导出，随约束或路由变更同步提交；运行中的管理 API 不提供、/redoc、/openapi.json、format_version、。顶层字段为 - `docs/dev/llm-configuration-frontend.md` → `references/docs-dev-llm-configuration-frontend.md` — LLM 配置管理前端交付契约 ｜ 关键词：/ops/#/llm-config、draft_revision、published_revision、loaded_revision、draft_version、published_version、loaded_version、{saved_at, note}、PUT /note、202 -- `docs/dev/llm-module.md` → `references/docs-dev-llm-module.md` — QuickQuip LLM 模块说明 ｜ 关键词：/search、LLM_TRACE_FLAG_FILE、data/llm_trace.db、run_tool_call_loop、LLM_TRACE_MEDIA_MIN_BYTES、data/trace_blobs/、[qq:blob sha256= bytes=]、media_truncation、restore、trigger_kind +- `docs/dev/llm-module.md` → `references/docs-dev-llm-module.md` — QuickQuip LLM 模块说明 ｜ 关键词：/search、LLM_TRACE_FLAG_FILE、data/llm_trace.db、run_tool_call_loop、LLM_TRACE_MEDIA_MIN_BYTES、b64_json、partial_image_b64、data、data/trace_blobs/、[qq:blob sha256= bytes=] - `docs/dev/mcp-integration.md` → `references/docs-dev-mcp-integration.md` — QuickQuip MCP 集成说明 ｜ 关键词：ToolRegistry、data/llm_config.db、settings.mcp、config.toml、search_web、tavily_search、tavily_crawl、tavily_research、docker、/var/run/docker.sock - `docs/dev/mcp-tutorial.md` → `references/docs-dev-mcp-tutorial.md` — 从零理解 MCP —— 以 QuickQuip 项目为例 ｜ 关键词：settings.mcp.servers、llm-module.md、src/quickquip/llm/mcp/、src/quickquip/llm/config_document.py、src/quickquip/llm/service_parts/mcp_lifecycle.py、。配置权威模板为、MCPClient、mcp/client.py、ToolRegistry、mcp_* - `docs/dev/README.md` → `references/docs-dev-readme.md` — QuickQuip 开发者文档 ｜ 关键词：docs/dev/、../glossary.md、adr/README.md、documenting.md、architecture.md、style.md、testing.md、branching.md、versioning.md、record-identities.md diff --git a/src/quickquip/app/web/routes/logs.py b/src/quickquip/app/web/routes/logs.py index a9354056..7337e0e7 100644 --- a/src/quickquip/app/web/routes/logs.py +++ b/src/quickquip/app/web/routes/logs.py @@ -258,10 +258,18 @@ def get_trace_call(call_id: str, restore: bool = False): def get_trace_blob(sha256: str): """Download one content-addressed media blob by its sha256 digest.""" - if not _SHA256_RE.match(sha256): + if not _SHA256_RE.fullmatch(sha256): raise HTTPException(status_code=422, detail="invalid blob digest") result = trace_store.download_blob(sha256) if result is None: raise HTTPException(status_code=404, detail="blob not found") - data, mime = result - return Response(content=data, media_type=mime) + data, _ = result + return Response( + content=data, + media_type="application/octet-stream", + headers={ + "Content-Disposition": f'attachment; filename="{sha256}.bin"', + "X-Content-Type-Options": "nosniff", + "Content-Security-Policy": "default-src 'none'; sandbox", + }, + ) diff --git a/src/quickquip/llm/provider/media_truncator.py b/src/quickquip/llm/provider/media_truncator.py index 2eb65141..c9010870 100644 --- a/src/quickquip/llm/provider/media_truncator.py +++ b/src/quickquip/llm/provider/media_truncator.py @@ -1,20 +1,8 @@ -"""内联媒体载荷的单副本存储（issue #291）。 +"""Trace 媒体载荷的内容寻址存储、结构化识别与原文恢复。 -trace 落库前把报文里的 base64 媒体载荷摘出来，按内容哈希写进 blob 库， -正文里只留 ``[qq:blob sha256= bytes=]`` 占位符。三个膨胀源因此 -坍缩：同一次调用重组体与原始帧各含一份同一张图 → 两个占位符指向同一 -blob；历史重放 → N 个约 80 字节的占位符；DB 增长从此只与文本量相关， -与图片数、对话长度、重放次数解耦。 - -纪律是「存坏比存大严重得多」： - -- 只截断**被引号正确闭合的完整 JSON 字符串**；中途截断、畸形输入、 - 未达门槛一律原样保留； -- 回拼先校验输入文本、再逐项校验 blob 哈希，任一不符即抛 ``ValueError``； -- 嵌套 JSON 的 ``\\"`` 场景要求开闭引号的转义形态一致，否则视为畸形。 - -本模块只做文本变换与内容寻址存储，不碰请求发送路径：发往上游的报文 -一字节不变（见 issue #291「边界与非目标」）。 +完整 JSON 或 SSE JSON 记录中的媒体载荷保留原始表达，正文替换为带摘要的 +占位符。恢复时验证正文与载荷哈希。引用提交与文件清理的跨进程事务由 +LLMTraceStore 管理；本模块不参与请求发送。 """ from __future__ import annotations @@ -35,7 +23,6 @@ logger = logging.getLogger(__name__) MEDIA_MIN_BYTES_ENV = "LLM_TRACE_MEDIA_MIN_BYTES" -DEFAULT_MIN_BYTES = 65536 MANIFEST_VERSION = 1 BLOB_SUFFIX = ".b64" @@ -43,21 +30,98 @@ # 裸 base64 的廉价初筛门槛：普通短字符串（UUID、token、路径）走原样路径， # 只有足够长的连续 base64 游程才值得进一步做完整闭合判定。 _MIN_BARE_BASE64_CHARS = 64 -_BASE64_BODY = r"[A-Za-z0-9+/=\s]" -# data URI 只认 ``data:/;base64,`` 这一种规范形态；带 charset -# 等参数或畸形声明的形态一律不匹配（保守，宁可原样存储）。 +_BASE64_BODY = r"[A-Za-z0-9+/]+={0,2}" _DATA_URI_HEAD = r"data:([A-Za-z0-9][A-Za-z0-9.+-]*/[A-Za-z0-9][A-Za-z0-9.+-]*);base64," _DATA_URI_PREFIX_RE = re.compile("^" + _DATA_URI_HEAD) -_DATA_URI_RE = re.compile("^" + _DATA_URI_HEAD + _BASE64_BODY + r"+$") -_BARE_BASE64_RE = re.compile(_BASE64_BODY + r"+$") -# 完整 JSON 字符串（含转义规则）：``\\.`` 吞掉转义对，因此内层 ``\"`` 不结束 -# 字符串。未闭合的尾巴不匹配任何一项，天然走「原样保留」路径。 -_STRING_RE = re.compile(r'"((?:[^"\\]|\\.)*)"') -# 字符串内容里的内层 JSON 字符串：开闭都必须是转义形态，天然一致。 -# 内容用排除式字符类而非贪婪匹配——贪婪会让正则从内层第一个 ``\"`` 一路 -# 吃到最后一个，把 ``img\": \"`` 之类前缀并进载荷，导致判定失败。内层若 -# 自带转义（base64 载荷不会），此处不匹配，按畸形原样保留。 -_ESCAPED_STRING_RE = re.compile(r'\\"([^"\\]*)\\"') +_BARE_BASE64_RE = re.compile(_BASE64_BODY) +_STRING_RE = re.compile(r'"([^"\\]*(?:\\.[^"\\]*)*)"') +_JSON_CHAR_RE = re.compile( + r"\\u[dD][89aAbB][0-9a-fA-F]{2}\\u[dD][c-fC-F][0-9a-fA-F]{2}" + r"|\\u[0-9a-fA-F]{4}|\\.|[^\\]+" +) + + +def _media_field(key: str, parent: dict) -> bool: + """仅在明确的媒体字段接纳裸编码，普通 content/text/result 保留原文。""" + + if key in {"image", "b64_json", "partial_image_b64"}: + return True + if key == "result": + return parent.get("type") == "image_generation_call" + if key == "data": + mime = parent.get("mimeType") or parent.get("mime_type") or parent.get("media_type") + return isinstance(mime, str) and mime.startswith(("image/", "audio/", "video/")) + return False + + +def _unique_object(pairs: list[tuple[str, object]]) -> dict: + result = {} + for key, value in pairs: + if key in result: + raise ValueError("duplicate JSON key") + result[key] = value + return result + + +def _string_offsets(raw: str, positions: set[int]) -> dict[int, int]: + """仅映射载荷边界，避免为大载荷的每个字符保留一个索引。""" + + pending = iter(sorted(positions)) + target = next(pending, None) + offsets = {} + decoded_start = 0 + for match in _JSON_CHAR_RE.finditer(raw): + piece = match.group() + escaped = piece.startswith("\\") + length = len(json.loads('"' + piece + '"')) if escaped else len(piece) + while target is not None and target < decoded_start + length: + offsets[target] = match.start() + (0 if escaped else target - decoded_start) + target = next(pending, None) + decoded_start += length + if target is None: + break + if target == decoded_start: + offsets[target] = len(raw) + return offsets + + +def _visit_json_values(node, tokens, text, depth, key="", parent=None): + if isinstance(node, dict): + for name, child in node.items(): + next(tokens) # JSON 对象键，遍历顺序与原文一致。 + yield from _visit_json_values(child, tokens, text, depth, name, node) + elif isinstance(node, list): + for child in node: + yield from _visit_json_values(child, tokens, text, depth) + elif isinstance(node, str): + token = next(tokens) + raw = token.group(1) + mime = match_payload_mime(node) + if mime is not None and (mime or _media_field(key, parent or {})): + if raw == node: + yield token.start(1), token.end(1), mime + elif node.lstrip().startswith(("{", "[")): + nested = _json_payloads(node, depth + 1) + if nested: + offsets = _string_offsets(raw, {point for cut in nested for point in cut[:2]}) + for start, end, nested_mime in nested: + source_start = token.start(1) + offsets[start] + source_end = token.start(1) + offsets[end] + if "\\" not in text[source_start:source_end]: + yield source_start, source_end, nested_mime + + +def _json_payloads(text: str, depth: int = 0) -> list[tuple[int, int, str]]: + """使用 JSON 结构确定媒体字段，再由字符串 token 保留原始字节表达。""" + + if depth > 4: + return [] + try: + value = json.loads(text, object_pairs_hook=_unique_object) + tokens = iter(_STRING_RE.finditer(text)) + return list(_visit_json_values(value, tokens, text, depth)) + except (ValueError, TypeError, RecursionError, StopIteration): + return [] def resolve_min_bytes(environ: dict[str, str] | None = None) -> int: @@ -83,12 +147,15 @@ def match_payload_mime(content: str) -> str | None: if len(content) < _MIN_BARE_BASE64_CHARS: return None - data_uri = _DATA_URI_RE.match(content) - if data_uri is not None: - return data_uri.group(1) - if _BARE_BASE64_RE.match(content): - return "" - return None + prefix = _DATA_URI_PREFIX_RE.match(content) + body = content[prefix.end():] if prefix else content + if not _BARE_BASE64_RE.fullmatch(body) or len(body) % 4: + return None + try: + base64.b64decode(body, validate=True) + except (binascii.Error, ValueError): + return None + return prefix.group(1) if prefix else "" def placeholder_for(sha256: str, size: int) -> str: @@ -111,7 +178,7 @@ def decode_payload(payload: str) -> tuple[bytes, str] | None: mime = prefix.group(1) body = payload[prefix.end():] try: - return base64.b64decode(body, validate=False), mime + return base64.b64decode(body, validate=True), mime except (binascii.Error, ValueError): return None @@ -163,6 +230,28 @@ def as_dict(self) -> dict[str, object]: } +def _sync_directory(path: Path) -> None: + # Python 在 Windows 上不提供可用于 fsync 的目录描述符。 + if os.name == "nt": + return + descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + + +def _ensure_blob_directory(path: Path) -> None: + missing = [] + current = path + while not current.exists(): + missing.append(current) + current = current.parent + path.mkdir(parents=True, exist_ok=True) + for created in reversed(missing): + _sync_directory(created.parent) + + class MediaBlobStore: """内容寻址 blob 库：``//.b64``，同哈希天然一份。 @@ -175,6 +264,8 @@ def __init__(self, root: str | Path) -> None: self.root = Path(root) def path_for(self, sha256: str) -> Path: + if not re.fullmatch(r"[0-9a-f]{64}", sha256): + raise ValueError("invalid blob digest") return self.root / sha256[:2] / f"{sha256}{BLOB_SUFFIX}" def store(self, payload: str) -> tuple[str, int]: @@ -186,7 +277,7 @@ def store(self, payload: str) -> tuple[str, int]: if destination.exists(): return digest, len(encoded) - destination.parent.mkdir(parents=True, exist_ok=True) + _ensure_blob_directory(destination.parent) handle, tmp_name = tempfile.mkstemp(dir=str(destination.parent), prefix=_TMP_PREFIX) tmp_path = Path(tmp_name) try: @@ -199,6 +290,7 @@ def store(self, payload: str) -> tuple[str, int]: except FileExistsError: # 并发写入同一哈希：盘上已有同一份内容，无需覆盖。 pass + _sync_directory(destination.parent) finally: tmp_path.unlink(missing_ok=True) return digest, len(encoded) @@ -213,7 +305,7 @@ def collect_orphaned(self, referenced: set[str], grace_seconds: float) -> int: """mark-sweep：删除未被引用、且 mtime 早于保守带的 blob。 选扫描式而非引用计数：对任意删除路径（含运维手工清库）免疫，不存在 - 计数漂移。保守带规避「blob 已写入、引用行尚未提交」的并发窗口。 + 计数漂移。调用者须在引用扫描至清理结束期间持有与写入相同的跨进程锁。 """ if not self.root.is_dir(): @@ -332,6 +424,7 @@ def truncate(self, text: str) -> TruncationOutcome: def restore(self, text: str, manifest: dict[str, object]) -> str: """按清单回拼字节级原文；任一校验不通过即抛 ``ValueError``。""" + _validate_field_manifest(manifest) expected_truncated = manifest.get("truncated_sha256") if expected_truncated: actual = hashlib.sha256(text.encode("utf-8")).hexdigest() @@ -350,7 +443,7 @@ def restore(self, text: str, manifest: dict[str, object]) -> str: offset = int(item.get("offset", -1)) span = int(item.get("span", -1)) digest = str(item.get("sha256", "")) - if offset < cursor or span < 0 or not digest: + if offset < cursor or span <= 0 or offset + span > len(text) or not digest: raise ValueError("manifest.splices 条目越界或字段缺失") pieces.append(text[cursor:offset]) payload = self.blob_store.load(digest) @@ -358,11 +451,17 @@ def restore(self, text: str, manifest: dict[str, object]) -> str: raise ValueError(f"媒体 blob 缺失：{digest}") if hashlib.sha256(payload.encode("utf-8")).hexdigest() != digest: raise ValueError(f"媒体 blob 哈希校验失败：{digest}") + if len(payload.encode("utf-8")) != item["orig_bytes"]: + raise ValueError("媒体 blob 字节数校验失败") + if text[offset:offset + span] != placeholder_for(digest, item["orig_bytes"]): + raise ValueError("媒体占位符与清单不匹配") pieces.append(payload) cursor = offset + span pieces.append(text[cursor:]) restored = "".join(pieces) + if len(restored.encode("utf-8")) != manifest["full_bytes"]: + raise ValueError("回拼正文的字节数不匹配") expected_full = manifest.get("full_sha256") if expected_full: actual = hashlib.sha256(restored.encode("utf-8")).hexdigest() @@ -371,25 +470,19 @@ def restore(self, text: str, manifest: dict[str, object]) -> str: return restored def _iter_payloads(self, text: str) -> Iterator[tuple[int, int, str]]: - """产出可替换载荷的 ``(start, end, mime)``，按文本位置升序。""" - - for match in _STRING_RE.finditer(text): - content_start, content_end = match.start(1), match.end(1) - content = match.group(1) - mime = match_payload_mime(content) - if mime is not None: - yield content_start, content_end, mime - continue - if '\\"' not in content: - continue - for inner in _ESCAPED_STRING_RE.finditer(content): - inner_mime = match_payload_mime(inner.group(1)) - if inner_mime is not None: - yield ( - content_start + inner.start(1), - content_start + inner.end(1), - inner_mime, - ) + """读取完整 JSON 或 SSE data 行；不改动不完整记录。""" + + if text.lstrip().startswith(("{", "[", '"')): + yield from _json_payloads(text) + return + offset = 0 + for line in text.splitlines(keepends=True): + if line.startswith("data:"): + body = line[5:].lstrip(" ") + start = offset + len(line) - len(body) + for begin, end, mime in _json_payloads(body): + yield start + begin, start + end, mime + offset += len(line) def _iter_field_manifests(payload: dict[str, object] | None) -> Iterator[dict[str, object]]: @@ -414,7 +507,7 @@ def collect_manifest_shas(manifest_json: str | None) -> set[str]: """从落库的 manifest JSON 里收集全部 blob sha（GC 的 mark 集合）。""" collected: set[str] = set() - for field_manifest in _iter_field_manifests(_load_manifest(manifest_json)): + for field_manifest in _iter_field_manifests(parse_manifest(manifest_json, allow_field=True)): for item in field_manifest.get("splices") or []: if isinstance(item, dict): digest = item.get("sha256") @@ -426,7 +519,7 @@ def collect_manifest_shas(manifest_json: str | None) -> set[str]: def find_manifest_mime(manifest_json: str | None, sha256: str) -> str | None: """在 manifest JSON 里查某个 blob 的 mime。""" - for field_manifest in _iter_field_manifests(_load_manifest(manifest_json)): + for field_manifest in _iter_field_manifests(parse_manifest(manifest_json, allow_field=True)): for item in field_manifest.get("splices") or []: if isinstance(item, dict) and item.get("sha256") == sha256: mime = item.get("mime") @@ -434,11 +527,77 @@ def find_manifest_mime(manifest_json: str | None, sha256: str) -> str | None: return None -def _load_manifest(manifest_json: str | None) -> dict[str, object] | None: - if not manifest_json: +def _validate_field_manifest(payload: dict) -> None: + for key in ("full_sha256", "truncated_sha256"): + if not isinstance(payload.get(key), str) or not re.fullmatch( + r"[0-9a-f]{64}", payload[key], + ): + raise ValueError("invalid manifest hash") + if type(payload.get("full_bytes")) is not int or payload["full_bytes"] < 0: + raise ValueError("invalid manifest byte count") + splices = payload.get("splices") + if not isinstance(splices, list) or not splices: + raise ValueError("invalid manifest splices") + for item in splices: + if not isinstance(item, dict): + raise ValueError("invalid manifest splice") + for key in ("offset", "span", "orig_bytes"): + if type(item.get(key)) is not int or item[key] < (0 if key == "offset" else 1): + raise ValueError("invalid manifest splice bounds") + if not isinstance(item.get("sha256"), str) or not re.fullmatch( + r"[0-9a-f]{64}", item["sha256"], + ): + raise ValueError("invalid manifest digest") + if not isinstance(item.get("mime", ""), str): + raise ValueError("invalid manifest MIME") + + +def parse_manifest( + raw: str | None, *, allow_field: bool = False, +) -> dict[str, object] | None: + """SQL NULL 表示原样记录；非空清单必须具有有效版本、字段和恢复校验。""" + + if raw is None: return None try: - payload = json.loads(manifest_json) - except (ValueError, TypeError): + payload = json.loads(raw) + except (ValueError, TypeError) as error: + raise ValueError("invalid media manifest JSON") from error + if not isinstance(payload, dict): + raise ValueError("invalid media manifest") + if allow_field and "splices" in payload: + _validate_field_manifest(payload) + return payload + fields = set(payload) - {"v"} + if payload.get("v") != MANIFEST_VERSION or not fields or not fields <= { + "request", "response", "response_raw", + }: + raise ValueError("invalid media manifest version or fields") + for key in fields: + if not isinstance(payload[key], dict): + raise ValueError("invalid media field manifest") + _validate_field_manifest(payload[key]) + return payload + + +def merge_field_manifests( + existing_json: str | None, + additions: dict[str, dict[str, object] | None], +) -> str | None: + """把新字段的清单并入已有 manifest，避免后写的一次覆盖前一次。 + + ``begin_call`` 落 request，``finish_call`` 落 response 与 response_raw， + 三者共用一个 ``media_truncation`` 列，必须合并而非覆盖。全部字段都无 + 截断时返回 ``None``——该行就是「原样存储」。 + """ + + payload = parse_manifest(existing_json) or {} + for field_name, manifest in additions.items(): + if manifest is None: + payload.pop(field_name, None) + else: + payload[field_name] = manifest + if not any(name != "v" for name in payload): return None - return payload if isinstance(payload, dict) else None + payload["v"] = MANIFEST_VERSION + return json.dumps(payload, ensure_ascii=False) diff --git a/src/quickquip/llm/provider/trace.py b/src/quickquip/llm/provider/trace.py index d15e6a46..bbea7ff7 100644 --- a/src/quickquip/llm/provider/trace.py +++ b/src/quickquip/llm/provider/trace.py @@ -2,7 +2,7 @@ from __future__ import annotations import asyncio -import json +import hashlib from contextlib import contextmanager from contextvars import ContextVar from quickquip.llm.session import SessionIdentity @@ -20,13 +20,14 @@ from quickquip.common.paths import LLM_TRACE_DB_PATH, LOGS_DIR from quickquip.llm.provider.media_truncator import ( - MANIFEST_VERSION, MediaBlobStore, MediaTruncator, collect_manifest_shas, decode_payload, find_manifest_mime, resolve_min_bytes, + parse_manifest, + merge_field_manifests, ) @@ -35,7 +36,7 @@ _TRACE_FLAG_FILE = os.getenv("LLM_TRACE_FLAG_FILE", "") _TRACE_RETENTION_DAYS = 14 _TRACE_BLOB_DIR_NAME = "trace_blobs" -# GC 保守带：blob 先落盘、引用行后提交，清扫必须避开这个并发窗口。 +# 孤立文件保留一天，给失败写入和运维检查留下恢复窗口。 _TRACE_BLOB_GRACE_SECONDS = 24 * 3600 _LEGACY_TRACE_GLOB = "quickquip_trace_????-??-??.jsonl" _TRACE_STALE_AFTER = timedelta(hours=1) @@ -50,42 +51,6 @@ def _utc_now() -> str: return datetime.now(timezone.utc).isoformat() -def _parse_manifest(raw: str | None) -> dict[str, object] | None: - """解析落库的 manifest JSON；非法或为空返回 ``None``（视作原样存储）。""" - - if not raw: - return None - try: - payload = json.loads(raw) - except (ValueError, TypeError): - return None - return payload if isinstance(payload, dict) else None - - -def _merge_field_manifests( - existing_json: str | None, - additions: dict[str, dict[str, object] | None], -) -> str | None: - """把新字段的清单并入已有 manifest，避免后写的一次覆盖前一次。 - - ``begin_call`` 落 request，``finish_call`` 落 response 与 response_raw， - 三者共用一个 ``media_truncation`` 列，必须合并而非覆盖。全部字段都无 - 截断时返回 ``None``——该行就是「原样存储」。 - """ - - payload = _parse_manifest(existing_json) or {} - changed = False - for field_name, manifest in additions.items(): - if manifest is None: - continue - payload[field_name] = manifest - changed = True - if not changed: - return existing_json or None - payload["v"] = MANIFEST_VERSION - return json.dumps(payload, ensure_ascii=False) - - @dataclass(slots=True) class TraceCapture: """Collect provider call IDs for one explicitly observed operation.""" @@ -409,11 +374,11 @@ def begin_call( self._ensure_schema() self._cleanup_if_due() call_id = uuid4().hex - # 唯一写入咽喉之一：三个大文本在落库前过一遍截断器，provider 层零改动。 - # request_bytes 仍记原始字节数（等同 full_bytes），语义不变。 - outcome = self.truncator.truncate(request_text) - manifest_json = _merge_field_manifests(None, {"request": outcome.manifest}) with self._connect() as conn: + # 与 GC 使用同一数据库写锁，覆盖文件创建/复用直到引用提交。 + conn.execute("BEGIN IMMEDIATE") + outcome = self.truncator.truncate(request_text) + manifest_json = merge_field_manifests(None, {"request": outcome.manifest}) cursor = conn.execute( """ INSERT INTO llm_http_traces ( @@ -466,18 +431,17 @@ def finish_call( response_raw_bytes: int = 0, ) -> None: self._ensure_schema() - # 重组响应与原始 SSE 帧各含一份同样的 base64：两处截断后同哈希指向 - # 同一个 blob，这正是「同一次调用双份」膨胀源的坍缩点。 - response_outcome = self.truncator.truncate(response_text or "") - raw_outcome = self.truncator.truncate(response_raw_text or "") with self._connect() as conn: + conn.execute("BEGIN IMMEDIATE") existing = conn.execute( "SELECT id, media_truncation FROM llm_http_traces WHERE call_id = ?", (call_id,), ).fetchone() if existing is None: return - manifest_json = _merge_field_manifests( + response_outcome = self.truncator.truncate(response_text or "") + raw_outcome = self.truncator.truncate(response_raw_text or "") + manifest_json = merge_field_manifests( existing["media_truncation"], { "response": response_outcome.manifest, @@ -575,7 +539,7 @@ def get_call(self, call_id: str) -> dict[str, object] | None: "response_headers": row["response_headers"] or "", "response_text": row["response_text"] or "", "response_raw_text": row["response_raw_text"] or "", - "media_truncation": _parse_manifest(row["media_truncation"]), + "media_truncation": parse_manifest(row["media_truncation"]), } def restore_call(self, call_id: str) -> dict[str, object] | None: @@ -612,7 +576,7 @@ def restore_call(self, call_id: str) -> dict[str, object] | None: return result def find_blob_mime(self, sha256: str) -> str | None: - """反查某个 blob 的 mime，供下载端点作 Content-Type。""" + """从最多 20 个含摘要的 JSON 清单中读取声明的 MIME，供诊断使用。""" self._ensure_schema() with self._connect() as conn: @@ -633,14 +597,19 @@ def find_blob_mime(self, sha256: str) -> str | None: def download_blob(self, sha256: str) -> tuple[bytes, str] | None: """取出 blob 并解成二进制；返回 ``(bytes, mime)``，缺失返回 ``None``。""" + self._ensure_schema() + self._cleanup_if_due() + declared_mime = self.find_blob_mime(sha256) + if declared_mime is None: + return None payload = self.blob_store.load(sha256) - if payload is None: + if payload is None or hashlib.sha256(payload.encode("utf-8")).hexdigest() != sha256: return None decoded = decode_payload(payload) if decoded is None: return None data, mime = decoded - resolved = mime or self.find_blob_mime(sha256) or "" + resolved = mime or declared_mime return data, resolved or "application/octet-stream" def list_events( @@ -719,9 +688,12 @@ def count_calls(self) -> int: def clear(self) -> int: self._ensure_schema() with self._connect() as conn: + conn.execute("BEGIN IMMEDIATE") count = int(conn.execute("SELECT COUNT(*) FROM llm_http_traces").fetchone()[0]) conn.execute("DELETE FROM llm_http_trace_events") conn.execute("DELETE FROM llm_http_traces") + # 先提交引用移除，再清理文件；即使清理失败，孤立文件也不可下载。 + self._collect_orphaned_blobs(grace_seconds=0) return count def storage_bytes(self) -> int: @@ -773,26 +745,27 @@ def _cleanup_legacy_trace_files(self, cutoff_timestamp: float) -> None: except OSError: logger.debug("Failed to clean legacy LLM trace file: %s", path) - def _collect_orphaned_blobs(self) -> None: - """mark-sweep 清扫无引用 blob；挂在每日留存清理末尾。 - - 选扫描式而非引用计数：对任意删除路径（含运维手工清库）免疫， - 不存在计数漂移。24h 保守带规避「blob 已写入、引用行尚未提交」。 - """ + def _collect_orphaned_blobs(self, *, grace_seconds=_TRACE_BLOB_GRACE_SECONDS) -> None: + """持有数据库写锁扫描引用并清理，协调 Bot/Web 的跨进程写入。""" try: with self._connect() as conn: + conn.execute("BEGIN IMMEDIATE") rows = conn.execute( "SELECT media_truncation FROM llm_http_traces " "WHERE media_truncation IS NOT NULL" - ).fetchall() - except sqlite3.Error: - logger.exception("LLM trace blob 引用扫描失败") + ) + referenced: set[str] = set() + for row in rows: + if parse_manifest(row["media_truncation"]) is None: + raise ValueError("invalid media manifest; skip blob GC") + referenced |= collect_manifest_shas(row["media_truncation"]) + removed = self.blob_store.collect_orphaned( + referenced, grace_seconds, + ) + except (sqlite3.Error, ValueError, TypeError): + logger.exception("LLM trace blob 引用扫描或清理失败") return - referenced: set[str] = set() - for row in rows: - referenced |= collect_manifest_shas(row["media_truncation"]) - removed = self.blob_store.collect_orphaned(referenced, _TRACE_BLOB_GRACE_SECONDS) if removed: logger.info("LLM trace blob GC 已清理 %d 个无引用 blob", removed) diff --git a/tests/unit/llm/provider/test_media_truncator.py b/tests/unit/llm/provider/test_media_truncator.py index 0b9305ea..48afafad 100644 --- a/tests/unit/llm/provider/test_media_truncator.py +++ b/tests/unit/llm/provider/test_media_truncator.py @@ -95,7 +95,7 @@ def test_nested_escaped_json_payload_is_replaced(tmp_path): def test_same_payload_is_stored_once(tmp_path): truncator = _truncator(tmp_path) payload = "F" * 2000 - original = '{"a": "' + payload + '", "b": "' + payload + '"}' + original = '{"image": "' + payload + '", "b64_json": "' + payload + '"}' outcome = truncator.truncate(original) @@ -109,7 +109,7 @@ def test_same_payload_is_stored_once(tmp_path): def test_splice_offsets_point_at_placeholders(tmp_path): truncator = _truncator(tmp_path) - original = '{"a": "' + "L" * 2000 + '", "b": "' + "M" * 2000 + '"}' + original = '{"image": "' + "L" * 2000 + '", "b64_json": "' + "M" * 2000 + '"}' outcome = truncator.truncate(original) @@ -169,7 +169,8 @@ def test_manifest_helpers_accept_field_and_wrapped_shapes(tmp_path): assert collect_manifest_shas(json.dumps(field_manifest)) == {digest} assert collect_manifest_shas(json.dumps(wrapped)) == {digest} assert find_manifest_mime(json.dumps(wrapped), digest) == "image/webp" - assert collect_manifest_shas("{not json") == set() + with pytest.raises(ValueError): + collect_manifest_shas("{not json") assert find_manifest_mime(None, digest) is None @@ -203,3 +204,72 @@ def test_resolve_min_bytes_reads_env_switch(): assert resolve_min_bytes({"LLM_TRACE_MEDIA_MIN_BYTES": "65536"}) == 65536 assert resolve_min_bytes({"LLM_TRACE_MEDIA_MIN_BYTES": "abc"}) == 0 assert resolve_min_bytes({"LLM_TRACE_MEDIA_MIN_BYTES": "-5"}) == 0 + + +@pytest.mark.parametrize("key", ["content", "text", "result", "data"]) +@pytest.mark.parametrize("value", ["These are ordinary English words " * 80, "abcd" * 600]) +def test_plain_text_stays_readable_even_if_valid_base64(tmp_path, key, value): + original = json.dumps({key: value}) + outcome = _truncator(tmp_path).truncate(original) + assert outcome.text == original + assert outcome.manifest is None + assert not list(tmp_path.rglob("*.b64")) + + +@pytest.mark.parametrize("shape", [ + {"inlineData": {"mimeType": "image/png", "data": "A" * 2000}}, + {"source": {"type": "base64", "media_type": "image/png", "data": "A" * 2000}}, + {"type": "image_generation_call", "result": "A" * 2000}, + {"type": "response.image_generation_call.partial_image", "partial_image_b64": "A" * 2000}, +]) +def test_protocol_media_fields_and_sse_restore(tmp_path, shape): + original = 'event: result\ndata: ' + json.dumps(shape) + '\n\ndata: [DONE]\n\n' + truncator = _truncator(tmp_path) + outcome = truncator.truncate(original) + assert outcome.manifest is not None + assert truncator.restore(outcome.text, outcome.manifest) == original + + +@pytest.mark.parametrize("payload", ["A" * 2001, "A" * 1999 + "=A", "A " * 1000]) +def test_invalid_base64_in_media_field_is_preserved(tmp_path, payload): + original = _text(payload) + outcome = _truncator(tmp_path).truncate(original) + assert outcome.text == original + assert outcome.manifest is None + + +@pytest.mark.parametrize("suffix", [",", ' trailing', ', "image": "duplicate"}']) +def test_malformed_or_duplicate_key_json_is_preserved(tmp_path, suffix): + original = _text("A" * 2000)[:-1] + suffix + outcome = _truncator(tmp_path).truncate(original) + assert outcome.text == original + assert outcome.manifest is None + + +def test_nested_json_unicode_offsets_and_same_text_in_non_media_field(tmp_path): + payload = "A" * 2000 + nested = json.dumps({"note": "中文😀\n", "image": payload, "content": payload}) + original = json.dumps({"body": nested, "text": payload}) + truncator = _truncator(tmp_path) + outcome = truncator.truncate(original) + assert len(outcome.manifest["splices"]) == 1 + assert json.loads(json.loads(outcome.text)["body"])["content"] == payload + assert json.loads(outcome.text)["text"] == payload + assert truncator.restore(outcome.text, outcome.manifest) == original + + +def test_multimegabyte_payload_scanning_has_bounded_memory(): + import tracemalloc + from quickquip.llm.provider.media_truncator import _json_payloads + + # 1 MiB 编码载荷曾引起约 150 MiB 正则回溯分配；容许线性副本开销。 + original = json.dumps({"image": "A" * (1024 * 1024)}) + tracemalloc.start() + try: + cuts = _json_payloads(original) + current, peak = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + assert len(cuts) == 1 + assert peak < 32 * len(original) + assert current < len(original) diff --git a/tests/unit/test_trace_store.py b/tests/unit/test_trace_store.py index 345e98e3..48f9d039 100644 --- a/tests/unit/test_trace_store.py +++ b/tests/unit/test_trace_store.py @@ -140,6 +140,7 @@ def capture_connect(): assert detail is not None assert detail["agent_loop_id"] == "legacy-call" + assert store.restore_call("legacy-call")["media_truncation"] is None assert detail["loop_sequence"] == 1 assert detail["response_raw_text"] == "" assert detail["response_raw_bytes"] == 0 @@ -408,3 +409,170 @@ async def run_loop(): details[0]["agent_loop_id"] # type: ignore[index] } assert [detail["loop_sequence"] for detail in details if detail] == [1, 2] + + +def _begin_media(store, text): + return store.begin_call( + provider_id="demo", protocol="openai", model="test", stream=True, + method="POST", url="https://example.test", request_headers="{}", + request_text=text, request_bytes=len(text.encode()), + ) + + +def _finish_media(store, call_id, text): + store.finish_call( + call_id, state="success", response_status=200, response_headers="{}", + response_text=text, response_bytes=len(text.encode()), duration_ms=1, + response_raw_text="data: " + text + "\n\n", + response_raw_bytes=len(("data: " + text + "\n\n").encode()), + ) + + +def test_media_three_field_restore_disable_and_missing_call(tmp_path): + import json + import pytest + + store = trace.LLMTraceStore(tmp_path / "trace.db", media_min_bytes=64) + text = json.dumps({"note": "中文", "image": "A" * 256}) + call_id = _begin_media(store, text) + _finish_media(store, call_id, text) + entry = store.get_call(call_id) + assert set(entry["media_truncation"]) == {"v", "request", "response", "response_raw"} + assert entry["request_bytes"] == entry["response_bytes"] == len(text.encode()) + assert entry["response_raw_bytes"] == len(("data: " + text + "\n\n").encode()) + assert len(list(store.blob_dir.rglob("*.b64"))) == 1 + + disabled = trace.LLMTraceStore(store.path, media_min_bytes=0) + restored = disabled.restore_call(call_id) + assert restored["request_text"] == restored["response_text"] == text + assert restored["response_raw_text"] == "data: " + text + "\n\n" + plain_id = _begin_media(disabled, text) + _finish_media(disabled, plain_id, text) + assert disabled.restore_call(plain_id)["media_truncation"] is None + assert disabled.get_call(plain_id)["request_text"] == text + assert disabled.restore_call("missing") is None + before_events = store.latest_event_id() + _finish_media(store, "missing", json.dumps({"image": "B" * 256})) + assert store.latest_event_id() == before_events + assert len(list(store.blob_dir.rglob("*.b64"))) == 1 + + blob = next(store.blob_dir.rglob("*.b64")) + blob.write_text("corrupt", encoding="utf-8") + with pytest.raises(ValueError): + store.restore_call(call_id) + blob.unlink() + with pytest.raises(ValueError): + store.restore_call(call_id) + + +def test_finish_replacement_removes_stale_response_manifest(tmp_path): + store = trace.LLMTraceStore(tmp_path / "trace.db", media_min_bytes=64) + text = '{"image":"' + "A" * 256 + '"}' + call_id = _begin_media(store, text) + _finish_media(store, call_id, text) + _finish_media(store, call_id, '{"text":"replacement"}') + restored = store.restore_call(call_id) + assert restored["request_text"] == text + assert restored["response_text"] == '{"text":"replacement"}' + assert set(restored["media_truncation"]) == {"v", "request"} + + +def _media_writer_process(path, text, ready, start, attempting, result): + store = trace.LLMTraceStore(path, media_min_bytes=64) + store._ensure_schema() + store._last_cleanup_date = datetime.now(timezone.utc).date().isoformat() + connect = store._connect + + def observe_connect(): + attempting.set() + return connect() + + store._connect = observe_connect + ready.set() + if not start.wait(15): + raise RuntimeError("writer start timed out") + result.put(_begin_media(store, text)) + + +def test_gc_and_new_reference_are_coordinated_across_processes(tmp_path, monkeypatch): + import multiprocessing + + ctx = multiprocessing.get_context("spawn") + ready, start, attempting = (ctx.Event() for _ in range(3)) + result = ctx.Queue() + store = trace.LLMTraceStore(tmp_path / "trace.db", media_min_bytes=64) + text = '{"image":"' + "A" * 256 + '"}' + old_call = _begin_media(store, text) + old_time = (datetime.now(timezone.utc) - timedelta(days=30)).isoformat() + with store._connect() as conn: + conn.execute( + "UPDATE llm_http_traces SET started_at=? WHERE call_id=?", (old_time, old_call), + ) + blob = next(store.blob_dir.rglob("*.b64")) + old_mtime = time.time() - 3 * 86400 + os.utime(blob, (old_mtime, old_mtime)) + scanned, release = threading.Event(), threading.Event() + sweep = store.blob_store.collect_orphaned + + def paused_sweep(referenced, grace_seconds): + assert not referenced + scanned.set() + assert release.wait(15) + return sweep(referenced, grace_seconds) + + monkeypatch.setattr(store.blob_store, "collect_orphaned", paused_sweep) + writer = ctx.Process( + target=_media_writer_process, + args=(store.path, text, ready, start, attempting, result), + ) + writer.start() + try: + assert ready.wait(15) + store._last_cleanup_date = None + with ThreadPoolExecutor(max_workers=1) as pool: + future = pool.submit(store.count_calls) + try: + assert scanned.wait(15) + start.set() + assert attempting.wait(15) + finally: + release.set() + future.result(timeout=15) + writer.join(timeout=15) + assert writer.exitcode == 0 + call_id = result.get(timeout=5) + assert store.restore_call(call_id)["request_text"] == text + assert blob.exists() + finally: + release.set() + start.set() + if writer.is_alive(): + writer.terminate() + writer.join(timeout=5) + result.close() + + +def test_clear_removes_media_and_orphan_downloads(tmp_path): + store = trace.LLMTraceStore(tmp_path / "trace.db", media_min_bytes=64) + call_id = _begin_media(store, '{"image":"' + "A" * 256 + '"}') + sha = store.get_call(call_id)["media_truncation"]["request"]["splices"][0]["sha256"] + assert store.download_blob(sha) is not None + assert store.clear() == 1 + assert store.count_calls() == 0 + assert store.download_blob(sha) is None + assert not list(store.blob_dir.rglob("*.b64")) + # 即使失败写入留下无引用文件，也不允许按摘要绕过记录生命周期下载。 + orphan, _ = store.blob_store.store("A" * 256) + assert store.download_blob(orphan) is None + + +def test_gc_skips_invalid_manifest_without_deleting_media(tmp_path): + store = trace.LLMTraceStore(tmp_path / "trace.db", media_min_bytes=64) + call_id = _begin_media(store, '{"image":"' + "A" * 256 + '"}') + blob = next(store.blob_dir.rglob("*.b64")) + with store._connect() as conn: + conn.execute( + "UPDATE llm_http_traces SET media_truncation=? WHERE call_id=?", ("{broken", call_id), + ) + store._collect_orphaned_blobs(grace_seconds=0) + assert blob.exists() diff --git a/tests/unit/web/test_logs_routes.py b/tests/unit/web/test_logs_routes.py index 6694953d..aff1b06e 100644 --- a/tests/unit/web/test_logs_routes.py +++ b/tests/unit/web/test_logs_routes.py @@ -94,3 +94,95 @@ def test_trace_sse_contains_metadata_as_one_event(): assert encoded.startswith("id: 12\n") assert '"call_id":"call-1"' in encoded assert encoded.endswith("\n\n") + + +@pytest.fixture +def media_client(monkeypatch, tmp_path): + from fastapi import FastAPI + from fastapi.testclient import TestClient + from quickquip.app.web import app as web_app, auth + from quickquip.llm.provider.trace import LLMTraceStore + + store = LLMTraceStore(tmp_path / "trace.db", media_min_bytes=64) + monkeypatch.setattr(logs, "trace_store", store) + app = FastAPI() + web_app._register_security_headers(app) + app.include_router(logs.router, prefix="/ops/api", dependencies=auth.protected_dependencies) + with TestClient(app) as client: + yield client, app, store + + +@pytest.mark.parametrize("mime", ["text/html", "image/svg+xml", "image/png"]) +def test_media_download_requires_auth_and_forces_attachment(media_client, mime): + import base64 + import json + from quickquip.app.web import auth + + client, app, store = media_client + raw = b"synthetic untrusted content " * 8 + payload = f"data:{mime};base64," + base64.b64encode(raw).decode() + text = json.dumps({"image": payload}) + call_id = store.begin_call( + provider_id="demo", protocol="openai", model="test", stream=False, + method="POST", url="https://example.test", request_headers="{}", + request_text=text, request_bytes=len(text.encode()), + ) + sha = store.get_call(call_id)["media_truncation"]["request"]["splices"][0]["sha256"] + path = f"/ops/api/logs/trace/blobs/{sha}" + assert client.get(path).status_code == 401 + assert client.get(f"/ops/api/logs/trace/calls/{call_id}?restore=true").status_code == 401 + app.dependency_overrides[auth.require_admin_session] = lambda: {"role": "admin"} + response = client.get(path) + assert response.status_code == 200 + assert response.content == raw + assert response.headers["content-type"] == "application/octet-stream" + assert response.headers["content-disposition"] == f'attachment; filename="{sha}.bin"' + assert response.headers["x-content-type-options"] == "nosniff" + assert response.headers["content-security-policy"] == "default-src 'none'; sandbox" + assert client.get("/ops/api/logs/trace/blobs/invalid").status_code == 422 + assert client.get("/ops/api/logs/trace/blobs/" + "0" * 64).status_code == 404 + + assert client.get(f"/ops/api/logs/trace/calls/{call_id}").json()["request_text"] != text + assert client.get(f"/ops/api/logs/trace/calls/{call_id}?restore=false").json()[ + "request_text" + ] != text + assert client.get(f"/ops/api/logs/trace/calls/{call_id}?restore=true").json()[ + "request_text" + ] == text + assert client.get("/ops/api/logs/trace/calls/missing?restore=true").status_code == 404 + blob = store.blob_store.path_for(sha) + blob.write_text("corrupt", encoding="utf-8") + assert client.get(f"/ops/api/logs/trace/calls/{call_id}?restore=true").status_code == 409 + blob.unlink() + assert client.get(f"/ops/api/logs/trace/calls/{call_id}?restore=true").status_code == 409 + + +@pytest.mark.parametrize("corruption", ["json", "field", "hashes", "offset", "unwrapped"]) +def test_restore_rejects_corrupt_manifest_with_409(media_client, corruption): + import json + from quickquip.app.web import auth + + client, app, store = media_client + app.dependency_overrides[auth.require_admin_session] = lambda: {"role": "admin"} + text = json.dumps({"image": "A" * 256}) + call_id = store.begin_call( + provider_id="demo", protocol="openai", model="test", stream=False, + method="POST", url="https://example.test", request_headers="{}", + request_text=text, request_bytes=len(text.encode()), + ) + manifest = store.get_call(call_id)["media_truncation"] + if corruption == "unwrapped": + manifest = manifest["request"] + elif corruption == "field": + manifest["request"] = [] + elif corruption == "hashes": + manifest["request"].pop("full_sha256") + manifest["request"].pop("truncated_sha256") + elif corruption == "offset": + manifest["request"]["splices"][0]["offset"] = None + serialized = "{broken" if corruption == "json" else json.dumps(manifest) + with store._connect() as conn: + conn.execute( + "UPDATE llm_http_traces SET media_truncation=? WHERE call_id=?", (serialized, call_id), + ) + assert client.get(f"/ops/api/logs/trace/calls/{call_id}?restore=true").status_code == 409