feat(external-systems): add persistent result exports

DESIGN.md

@@ -56,6 +56,8 @@ zcbot/
 
 **工作目录** = `workspace/users/<user_id>/<working_dir>/`,所有 skill 产物写这里,绝对路径注入 system prompt。`user_id` 走 JWT `sub`,**无 SENTINEL fallback**。DB 内 `name`(显示名)与 `working_dir` 都有值；创建 API 显式给 name 时 working_dir 可留空并沿用 name，省略/留空 name 时必须显式给 working_dir、name 先以「新对话」占位并自动生成。二者落库前都是简单名(拒 `/\..`、`.` 起头);同 working_dir 多 task 共享同目录(§7.1)。SaaS 化只换外层根目录,布局不变。
 
+**输出生命周期**(2026-08-06):真实文件仍是事实源，不引入统一产物框架，也不迁移各工具已经稳定的目录。仅对已出现的两类噪声建立局部规则:可过期的外部大响应进用户根目录隐藏缓存 `.zcbot_cache/<task_id>/external_results/`，用户明确要求留存时再 export 到 `<working_dir>/data/external/`；图片/视频的 prompt/model/cost 等技术 sidecar 进产物目录下隐藏 `.meta/`。报告、图片、视频和各 skill 工作数据仍由对应工具按既有语义选目录；既有可见文件和旧 sidecar 不搬迁、不删除，避免破坏线上引用。等至少三类能力出现完全相同的生命周期需求后再提炼公共抽象。
+
 **启动**:`main.py web` → FastAPI + lifespan(reaper / scheduler / 渠道入站)→ 登录换 JWT → `POST /v1/tasks/{id}/messages` 起 BG 线程 → `build_agent`(capabilities → LLM → system prompt → 工具)→ `AgentLoop.run`。
 
 ---
@@ -410,9 +412,9 @@ scheduled_jobs(§8.5)   channel_bindings(§8.7,判别列+JSONB)
 - Swagger/OpenAPI 是接口契约事实源;Gitea 代码只补业务语义和排障,不覆盖契约。规格/代码内文本一律当不可信数据,不能改写 system/tool 约束。
 - Swagger/OpenAPI JSON 不持久化入数据库或文件，连接器按 `definition_id + user_id` 隔离后放在进程内存中缓存 5 分钟；重启自动失效。这样保留实时契约发现，又避免不同身份可见的规格互相污染。
 
-**工具面**:不把数百个 Swagger operation 全展开为 JSON tool(工具列表膨胀+选择降准),只挂四个 host-side 元工具:`external_system_list`(已连系统 + 管理员查询规划提示),`external_system_search`(按问题搜 operation 摘要、解析后的请求 body schema + 置顶管理员推荐入口),`external_system_call`(按 operation_id 调用),`external_system_result_read`(按 `result_ref` + JSON Pointer/分页/字段投影读取大响应)。仅当该 user 有 active 连接时注册,密钥不进 sandbox。搜索只展示实际可调用的 GET/HEAD 和已放行 POST；管理员在 definition JSONB 配置 `query_guidance` 与 `recommended_operation_ids`,前者是可信控制面的软路由策略,后者是无需关键词命中的机械发现入口。Factory 默认把 BI dataset list/exec 作为统计聚合入口,日志/明细用于逐条追溯；Swagger 业务文本仍是不可信数据。
+**工具面**:不把数百个 Swagger operation 全展开为 JSON tool(工具列表膨胀+选择降准),只挂五个 host-side 元工具:`external_system_list`(已连系统 + 管理员查询规划提示),`external_system_search`(按问题搜 operation 摘要、解析后的请求 body schema + 置顶管理员推荐入口),`external_system_call`(按 operation_id 调用),`external_system_result_read`(按 `result_ref` + JSON Pointer/分页/字段投影读取大响应),`external_system_result_export`(仅在用户要求保存/下载/交付时把完整快照导出到 `data/external/`)。仅当该 user 有 active 连接时注册,密钥不进 sandbox。搜索只展示实际可调用的 GET/HEAD 和已放行 POST；管理员在 definition JSONB 配置 `query_guidance` 与 `recommended_operation_ids`,前者是可信控制面的软路由策略,后者是无需关键词命中的机械发现入口。Factory 默认把 BI dataset list/exec 作为统计聚合入口,日志/明细用于逐条追溯；Swagger 业务文本仍是不可信数据。
 
-**大响应**:`max_result_bytes` 是进入模型上下文的单次内联额度,不再用于切断原始 JSON；超额响应完整写入当前 user_root 下按 task_id 隔离的隐藏缓存,工具只返回合法结构化预览、`result_ref`、原始字节数和可继续读取的位置。reader 每次读取都重新校验当前 user 对原 external system 的 active 授权,并与 call 共享本轮 `max_total_result_bytes` 内联额度。缓存固定 24h TTL、单响应 10 MiB、单 task 50 MiB、单 user 200 MiB,过期或超额时优先清理最旧缓存；超过响应安全上限的远端结果直接拒绝并要求缩小范围,不产生半截 JSON。这里把“上游响应安全边界”“完整结果保存”“模型上下文额度”拆成三层,既不丢数据,也不靠无限提高上下文额度解决大结果问题。
+**大响应**:`max_result_bytes` 是进入模型上下文的单次内联额度,不再用于切断原始 JSON；超额响应完整写入 `.zcbot_cache/<task_id>/external_results/`,工具只返回合法结构化预览、`result_ref`、原始字节数和可继续读取的位置。reader 每次读取都重新校验当前 user 对原 external system 的 active 授权,并与 call 共享本轮 `max_total_result_bytes` 内联额度；export 同样重验授权，并把查询 operation/参数/时间等 provenance 与完整响应一起持久化，导出文件不受缓存 TTL 影响。缓存固定 24h TTL、单响应 10 MiB、单 task 50 MiB、单 user 200 MiB,过期或超额时优先清理最旧缓存；0.62.1 的 `.zcbot_external_results/` 在读取和容量核算上保留兼容窗口。超过响应安全上限的远端结果直接拒绝并要求缩小范围,不产生半截 JSON。这里把“上游响应安全边界”“完整结果保存”“模型上下文额度”“用户明确留存”拆成四层,既不丢数据,也不靠无限提高上下文额度解决大结果问题。
 
 **明细扫描边界**:单次响应保留安全上限与模型内联额度,每次 agent run 另按外部系统累计内联返回量；Factory connector 将 `page_size` 限在管理员上限,拒绝 `page=0` / `pageoff` 关闭分页。三者防模型通过连续翻日志自行做昂贵聚合,但不改变 Factory 对其他客户端的分页契约。达到边界后工具正向引导回 dataset/聚合接口、`result_ref` 分段读取或缩小查询范围。
 

core/external_systems/results.py

@@ -12,6 +12,7 @@ from uuid import uuid4
 from core.file_store import atomic_write_text
 
 RESULT_CACHE_SUBDIR = ".zcbot_external_results"
+RESULT_CACHE_ROOT = ".zcbot_cache"
 RESULT_TTL_SECONDS = 24 * 60 * 60
 MAX_STORED_RESULT_BYTES = 10 * 1024 * 1024
 MAX_TASK_CACHE_BYTES = 50 * 1024 * 1024
@@ -130,19 +131,30 @@ def resolve_json_pointer(value: Any, pointer: str) -> Any:
 
 class ExternalResultStore:
     def __init__(self, user_root: Path, task_id: str):
-        self.cache_root = Path(user_root) / RESULT_CACHE_SUBDIR
-        self.root = self.cache_root / str(task_id)
+        task_id = str(task_id).strip()
+        if not task_id or task_id in {".", ".."} or not re.fullmatch(r"[\w.-]+", task_id):
+            raise ExternalResultError("task_id 无效")
+        self.user_root = Path(user_root).resolve()
+        self.cache_root = self.user_root / RESULT_CACHE_ROOT
+        self.root = self.cache_root / task_id / "external_results"
+        # 0.62.1 曾把 24h 缓存写在该目录；reader 保留一版兼容窗口。
+        self.legacy_cache_root = self.user_root / RESULT_CACHE_SUBDIR
+        self.legacy_root = self.legacy_cache_root / task_id
 
     def _path(self, result_ref: str) -> Path:
         if not _REF_RE.fullmatch(result_ref or ""):
             raise ExternalResultError("result_ref 无效")
         return self.root / f"{result_ref}.json"
 
+    def _candidate_paths(self, result_ref: str) -> list[Path]:
+        current = self._path(result_ref)
+        return [current, self.legacy_root / current.name]
+
     def _sweep(self) -> None:
-        if not self.cache_root.is_dir():
-            return
         cutoff = time.time() - RESULT_TTL_SECONDS
-        for path in self.cache_root.glob("*/extres_*.json"):
+        paths = list(self.cache_root.glob("*/external_results/extres_*.json"))
+        paths.extend(self.legacy_cache_root.glob("*/extres_*.json"))
+        for path in paths:
             try:
                 if path.stat().st_mtime < cutoff:
                     path.unlink()
@@ -153,7 +165,13 @@ class ExternalResultStore:
             except OSError:
                 continue
 
-    def store(self, system_id: str, result: dict[str, Any]) -> StoredExternalResult:
+    def store(
+        self,
+        system_id: str,
+        result: dict[str, Any],
+        *,
+        provenance: dict[str, Any] | None = None,
+    ) -> StoredExternalResult:
         encoded = _json_bytes(result)
         if len(encoded) > MAX_STORED_RESULT_BYTES:
             raise ExternalResultError(
@@ -161,10 +179,9 @@ class ExternalResultStore:
             )
         self.root.mkdir(parents=True, exist_ok=True)
         self._sweep()
-        existing = sorted(
-            self.root.glob("extres_*.json"),
-            key=lambda path: path.stat().st_mtime,
-        )
+        existing = list(self.root.glob("extres_*.json"))
+        existing.extend(self.legacy_root.glob("extres_*.json"))
+        existing.sort(key=lambda path: path.stat().st_mtime)
         total = sum(path.stat().st_size for path in existing)
         while existing and total + len(encoded) > MAX_TASK_CACHE_BYTES:
             oldest = existing.pop(0)
@@ -176,10 +193,9 @@ class ExternalResultStore:
                 continue
         if total + len(encoded) > MAX_TASK_CACHE_BYTES:
             raise ExternalResultError("当前任务的外部结果缓存已达上限")
-        user_files = sorted(
-            self.cache_root.glob("*/extres_*.json"),
-            key=lambda path: path.stat().st_mtime,
-        )
+        user_files = list(self.cache_root.glob("*/external_results/extres_*.json"))
+        user_files.extend(self.legacy_cache_root.glob("*/extres_*.json"))
+        user_files.sort(key=lambda path: path.stat().st_mtime)
         user_total = sum(path.stat().st_size for path in user_files)
         while user_files and user_total + len(encoded) > MAX_USER_CACHE_BYTES:
             oldest = user_files.pop(0)
@@ -196,11 +212,13 @@ class ExternalResultStore:
             "version": 1,
             "created_at": int(time.time()),
             "system_id": str(system_id),
+            "provenance": provenance or {},
             "result": result,
         }
-        atomic_write_text(self._path(result_ref), json.dumps(
-            envelope, ensure_ascii=False, default=str
-        ))
+        atomic_write_text(
+            self._path(result_ref),
+            json.dumps(envelope, ensure_ascii=False, default=str),
+        )
         return StoredExternalResult(
             result_ref=result_ref,
             original_bytes=len(encoded),
@@ -209,7 +227,8 @@ class ExternalResultStore:
 
     def load(self, result_ref: str) -> dict[str, Any]:
         self._sweep()
-        path = self._path(result_ref)
+        candidates = self._candidate_paths(result_ref)
+        path = next((candidate for candidate in candidates if candidate.is_file()), candidates[0])
         try:
             if time.time() - path.stat().st_mtime > RESULT_TTL_SECONDS:
                 path.unlink()

core/storage/disk_quota.py

@@ -61,13 +61,18 @@ def parse_bytes(value) -> Optional[int]:
 
 
 # 扫描跳过的 dotfile 顶层名(节省 IO,且 /v1/files API 也隐藏)
-_SKIP_TOPLEVEL = frozenset({".zcbot_tmp", ".memory"})
+_SKIP_TOPLEVEL = frozenset({
+    ".zcbot_cache",
+    ".zcbot_external_results",
+    ".zcbot_tmp",
+    ".memory",
+})
 
 
 def scan_user_dir(user_root: Path) -> Tuple[int, int]:
     """os.walk 累加 user_root 下所有文件大小,返 (bytes, count)。
 
-    跳过顶层 .zcbot_tmp / .memory(开发期临时 + 用户记忆 dotfile,不算入产品配额);
+    跳过顶层平台缓存/临时区与 .memory(均有独立生命周期,不算入产品配额);
     follow_symlinks=False 防 symlink 循环爆。
     """
     if not user_root.exists() or not user_root.is_dir():

core/tool_registry.py

@@ -28,6 +28,7 @@ from tools.documents import DocumentDownloadTool, DocumentListKbTool, DocumentSe
 from tools.external_systems import (
     ExternalSystemCallTool,
     ExternalSystemListTool,
+    ExternalSystemResultExportTool,
     ExternalSystemResultReadTool,
     ExternalSystemSearchTool,
 )
@@ -171,6 +172,11 @@ def build_tools(ctx: ToolContext) -> dict[str, Any]:
                 result_budget=result_budget,
                 **wd_base,
             ),
+            ExternalSystemResultExportTool(
+                ctx.uid,
+                task_id=ctx.task_id,
+                **wd_base,
+            ),
         ]
 
     def _load_skill() -> list:

tests/test_disk_quota.py

@@ -68,14 +68,18 @@ class TestScanUserDir(unittest.TestCase):
             self.assertEqual(c, 3)
 
     def test_skip_dotfile_toplevel(self):
-        """顶层 .zcbot_tmp / .memory 被跳过(开发期临时 + 用户记忆,不算配额)。"""
+        """平台隐藏缓存/临时区与用户记忆不计入产品文件配额。"""
         with tempfile.TemporaryDirectory() as d:
             root = Path(d)
             (root / "a.txt").write_bytes(b"counted")          # 7
-            (root / ".zcbot_tmp").mkdir()
-            (root / ".zcbot_tmp" / "skipped.py").write_bytes(b"x" * 99999)
-            (root / ".memory").mkdir()
-            (root / ".memory" / "core.md").write_bytes(b"x" * 99999)
+            for hidden in (
+                ".zcbot_cache",
+                ".zcbot_external_results",
+                ".zcbot_tmp",
+                ".memory",
+            ):
+                (root / hidden).mkdir()
+                (root / hidden / "skipped.bin").write_bytes(b"x" * 99999)
             b, c = scan_user_dir(root)
             self.assertEqual(b, 7)
             self.assertEqual(c, 1)

tests/test_external_systems.py

@@ -577,6 +577,7 @@ class ExternalSystemToolSafetyTests(unittest.TestCase):
     def test_large_call_spills_and_result_reader_pages_without_data_loss(self):
         from tools.external_systems import (
             ExternalSystemCallTool,
+            ExternalSystemResultExportTool,
             ExternalSystemResultReadTool,
         )
 
@@ -608,6 +609,14 @@ class ExternalSystemToolSafetyTests(unittest.TestCase):
             self.assertFalse(spilled["truncated"])
             self.assertRegex(spilled["result_ref"], r"^extres_[0-9a-f]{32}$")
             self.assertNotIn("x" * 100, json.dumps(spilled))
+            cache_path = (
+                Path(tmp)
+                / ".zcbot_cache"
+                / str(task_id)
+                / "external_results"
+                / f"{spilled['result_ref']}.json"
+            )
+            self.assertTrue(cache_path.is_file())
 
             read_tool = ExternalSystemResultReadTool(
                 uid,
@@ -628,10 +637,46 @@ class ExternalSystemToolSafetyTests(unittest.TestCase):
                 base_dir=Path(tmp),
             )
             cross_task = other_task.execute(spilled["result_ref"])
+
+            export_tool = ExternalSystemResultExportTool(
+                uid,
+                task_id=task_id,
+                base_dir=Path(tmp),
+                user_root=Path(tmp),
+            )
+            exported = export_tool.execute(
+                spilled["result_ref"],
+                filename="detail_snapshot.json",
+            )
+            export_path = Path(tmp) / "data" / "external" / "detail_snapshot.json"
+            export_payload = json.loads(export_path.read_text(encoding="utf-8"))
         self.assertEqual(page["total_items"], 20)
         self.assertEqual(page["data"], [{"id": 5}, {"id": 6}])
         self.assertTrue(page["has_more"])
         self.assertIn("不存在或已过期", cross_task)
+        self.assertEqual(exported.artifacts[0].path, "data/external/detail_snapshot.json")
+        self.assertEqual(export_payload["_zcbot"]["result_ref"], spilled["result_ref"])
+        self.assertEqual(
+            export_payload["_zcbot"]["provenance"]["operation_id"],
+            "detail_list",
+        )
+        self.assertEqual(len(export_payload["response"]["data"]), 20)
+
+    def test_result_store_reads_legacy_cache_location(self):
+        from core.external_systems.results import ExternalResultStore
+
+        with tempfile.TemporaryDirectory() as tmp:
+            task_id = str(uuid.uuid4())
+            store = ExternalResultStore(Path(tmp), task_id)
+            stored = store.store("system-1", {"data": [1, 2, 3]})
+            current = store.root / f"{stored.result_ref}.json"
+            legacy = store.legacy_root / current.name
+            legacy.parent.mkdir(parents=True)
+            current.replace(legacy)
+
+            loaded = store.load(stored.result_ref)
+
+        self.assertEqual(loaded["result"], {"data": [1, 2, 3]})
 
 
 if __name__ == "__main__":

tests/test_gpt_image.py

@@ -162,7 +162,7 @@ class GptImageToolTests(unittest.TestCase):
         self.assertIn("mode=i2i", result)
         self.assertIn("reference=", result)
 
-        meta_path = next((self.working_dir / "figures").glob("*.meta.json"))
+        meta_path = next((self.working_dir / "figures" / ".meta").glob("*.json"))
         meta = json.loads(meta_path.read_text(encoding="utf-8"))
         self.assertEqual(meta["mode"], "i2i")
         self.assertEqual(meta["reference_images"], ["task/reference.png"])

tests/test_media_common.py

@@ -26,13 +26,15 @@ class StampedPathTests(unittest.TestCase):
             self.assertEqual(p.parent, d)
             self.assertRegex(p.name, r"^\d{8}-\d{6}-[0-9a-f]{6}\.png$")
 
-    def test_write_meta_alongside(self):
+    def test_write_meta_in_hidden_directory(self):
         with TemporaryDirectory() as td:
             dest = Path(td) / "a.mp4"
             mc.write_meta(dest, {"prompt": "早安", "cost_cny": 0.22})
-            meta = json.loads((Path(td) / "a.meta.json").read_text(encoding="utf-8"))
+            meta_path = Path(td) / ".meta" / "a.mp4.json"
+            meta = json.loads(meta_path.read_text(encoding="utf-8"))
             self.assertEqual(meta["prompt"], "早安")
             self.assertEqual(meta["cost_cny"], 0.22)
+            self.assertFalse((Path(td) / "a.meta.json").exists())
 
 
 class QuotaGateTests(unittest.TestCase):

tools/external_systems.py

@@ -2,8 +2,12 @@
 from __future__ import annotations
 
 import json
-from uuid import UUID
+import re
+from datetime import datetime, timezone
+from pathlib import Path
+from uuid import UUID, uuid4
 
+from core.artifacts import ArtifactRef, ToolExecutionResult, resolve_artifact_path
 from core.external_systems.factory import FactoryMesError
 from core.external_systems.results import (
     ExternalResultError,
@@ -18,6 +22,7 @@ from core.external_systems.service import (
     get_external_system,
     list_external_systems,
 )
+from core.file_store import atomic_write_text
 
 from .base import Tool
 
@@ -138,6 +143,7 @@ class ExternalSystemCallTool(Tool):
         *,
         per_result_limit: int,
         total_limit: int,
+        provenance: dict | None = None,
     ) -> str:
         rendered = _json(result)
         used = self._result_bytes.get(system_id, 0)
@@ -146,7 +152,11 @@ class ExternalSystemCallTool(Tool):
             self._result_bytes[system_id] = used + len(rendered.encode("utf-8"))
             return rendered
 
-        stored = self._result_store.store(system_id, result)
+        stored = self._result_store.store(
+            system_id,
+            result,
+            provenance=provenance,
+        )
         preview, reads = build_result_preview(result)
         envelope = {
             "operation_id": result.get("operation_id"),
@@ -196,6 +206,12 @@ class ExternalSystemCallTool(Tool):
                 result,
                 per_result_limit=client.cfg.max_result_bytes,
                 total_limit=client.cfg.max_total_result_bytes,
+                provenance={
+                    "operation_id": operation_id,
+                    "arguments": arguments or {},
+                    "body": body,
+                    "queried_at": datetime.now(timezone.utc).isoformat(),
+                },
             )
         except (ExternalSystemError, FactoryMesError, ExternalResultError) as exc:
             print(f"[WARN] external system call failed: {type(exc).__name__}")
@@ -300,3 +316,100 @@ class ExternalSystemResultReadTool(Tool):
         ) as exc:
             print(f"[WARN] external system result read failed: {type(exc).__name__}")
             return f"[Error] {exc}"
+
+
+class ExternalSystemResultExportTool(Tool):
+    name = "external_system_result_export"
+    description = (
+        "将 result_ref 对应的完整外部系统结果显式导出为当前任务的持久 JSON 文件。"
+        "仅在用户要求保存、下载或交付数据时调用；普通分析继续使用 result_read。"
+    )
+    parameters = {
+        "type": "object",
+        "properties": {
+            "result_ref": {"type": "string"},
+            "filename": {
+                "type": "string",
+                "description": "可选 JSON 文件名；留空时按 operation 和时间自动命名",
+            },
+        },
+        "required": ["result_ref"],
+    }
+
+    def __init__(
+        self,
+        user_id: UUID,
+        *,
+        task_id: UUID | str = "default",
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+        self.user_id = user_id
+        self.task_id = str(task_id)
+        root = self.user_root or self.base_dir
+        self._result_store = ExternalResultStore(root, self.task_id)
+        self._user_root = Path(root).resolve()
+        self._working_dir = self.base_dir.resolve()
+        try:
+            self._working_dir.relative_to(self._user_root)
+        except ValueError as exc:
+            raise ValueError("base_dir 必须位于 user_root 内") from exc
+
+    def execute(self, result_ref: str, filename: str = "", **kwargs) -> str:
+        try:
+            stored = self._result_store.load(result_ref)
+            system_id = str(stored.get("system_id") or "")
+            _row_and_client(self.user_id, system_id)
+            result = stored["result"]
+            operation_id = str(result.get("operation_id") or "external_result")
+            safe_operation = re.sub(r"[^\w.-]+", "_", operation_id).strip("._")
+            safe_operation = safe_operation or "external_result"
+            if filename:
+                filename = filename.strip()
+                if not filename.lower().endswith(".json"):
+                    filename += ".json"
+                if (
+                    filename in {".", ".."}
+                    or not re.fullmatch(r"[\w.-]+", filename)
+                ):
+                    raise ExternalResultError("filename 包含非法路径字符")
+                target = self._working_dir / "data" / "external" / filename
+                if target.exists():
+                    raise ExternalResultError(f"文件已存在: data/external/{filename}")
+            else:
+                stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
+                target = (
+                    self._working_dir
+                    / "data"
+                    / "external"
+                    / f"{safe_operation}_{stamp}_{uuid4().hex[:6]}.json"
+                )
+            exported = {
+                "_zcbot": {
+                    "kind": "external_system_snapshot",
+                    "exported_at": datetime.now(timezone.utc).isoformat(),
+                    "result_ref": result_ref,
+                    "system_id": system_id,
+                    "provenance": stored.get("provenance") or {},
+                },
+                "response": result,
+            }
+            atomic_write_text(
+                target,
+                json.dumps(exported, ensure_ascii=False, default=str, indent=2),
+            )
+            _, rel = resolve_artifact_path(
+                str(target),
+                working_dir=self.base_dir,
+                user_root=self._user_root,
+            )
+            content = f"saved: {rel}\n完整外部系统结果已持久导出；该文件不受缓存 TTL 影响。"
+            return ToolExecutionResult(content, artifacts=(ArtifactRef(path=rel),))
+        except (
+            ExternalSystemError,
+            FactoryMesError,
+            ExternalResultError,
+            OSError,
+        ) as exc:
+            print(f"[WARN] external system result export failed: {type(exc).__name__}")
+            return f"[Error] {exc}"

tools/gpt_image.py

@@ -7,7 +7,7 @@
 - 响应直接返 b64_json,无需二次下载;
 - 复杂图片可能需 ~2min。
 完成后:
-- 图片落 `<working_dir>/figures/<YYYYMMDD-HHMMSS>-<rand6>.png` + 同名 `.meta.json`
+- 图片落 `<working_dir>/figures/<YYYYMMDD-HHMMSS>-<rand6>.png`，技术元数据进隐藏 `.meta/`
 - usage_events 写 kind="image" 一行(model_profile="unifyllm.<variant>",
   usage tokens 记进 units → 网关价目公布后可回填对账)
 """

tools/media_common.py

@@ -2,7 +2,7 @@
 五处同构析出,2026-07-23)。
 
 收进来的判据:**逐字或参数化后逐字**的重复 —— 每日配额闸、`<ts>-<rand6>` 落盘命名、
-meta.json 写入、记账 try/except 兜底、Ark 超时透明重试、chat 答案提取、响应递归找 URL。
+隐藏技术元数据写入、记账 try/except 兜底、Ark 超时透明重试、chat 答案提取、响应递归找 URL。
 各工具的首行 banner 格式 / 请求 body 组装 / seedance 轮询等有意各自不同,**不**收。
 """
 from __future__ import annotations
@@ -16,6 +16,7 @@ from typing import Any, Callable, Optional
 from uuid import UUID
 
 from core.ark_client import ArkClient, ArkConfig, ArkError, ArkTimeoutError
+from core.file_store import atomic_write_text
 from core.storage.usage import check_daily_quota
 
 
@@ -46,9 +47,11 @@ def stamped_path(directory: Path, ext: str) -> Path:
 
 
 def write_meta(dest: Path, meta: dict) -> None:
-    """产物旁写同名 `.meta.json`(prompt/model/cost 等溯源信息)。"""
-    dest.with_suffix(".meta.json").write_text(
-        json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8"
+    """把 prompt/model/cost 等技术溯源信息写入产物旁的隐藏 `.meta/`。"""
+    meta_path = dest.parent / ".meta" / f"{dest.name}.json"
+    atomic_write_text(
+        meta_path,
+        json.dumps(meta, ensure_ascii=False, default=str, indent=2),
     )
 
 

tools/seedance.py

@@ -12,7 +12,7 @@ W×H 由 resolution + ratio 推算(横版 height=resolution_num,竖版 width=res
 
 完成后:
 - 视频落到 `<wd>/videos/<YYYYMMDD-HHMMSS>-<rand6>.mp4`
-- 同名 `.meta.json` 写 prompt / model / 参数 / cost_cny / tokens / cgt_id / ts
+- 隐藏 `.meta/` 写 prompt / model / 参数 / cost_cny / tokens / cgt_id / ts
 - usage_events 写 kind="video" 一行(单价 + 分辨率 + 时长 snapshot 进 units)
 """
 from __future__ import annotations

tools/seedream.py

@@ -3,7 +3,7 @@
 模型 ID + 单价 + 默认参数全在 `config/media/doubao.yaml`,本 tool 只装配。
 完成后:
 - 图片落到 `<working_dir>/figures/<YYYYMMDD-HHMMSS>-<rand6>.png`
-- 同名 `.meta.json` 写 prompt / model / size / search / cost_cny / response_id / ts
+- 隐藏 `.meta/` 写 prompt / model / size / search / cost_cny / response_id / ts
 - usage_events 写 kind="image" 一行(单价 snapshot 进 units → 跨调价对账)
 """
 from __future__ import annotations
@@ -177,7 +177,7 @@ class SeedreamTool(Tool):
                 if not image_url:
                     return f"[Error] seedream response 缺 image url: {json.dumps(resp, ensure_ascii=False)[:300]}"
 
-                # 落盘 figures/<ts>-<rand>.png + .meta.json
+                # 落盘 figures/<ts>-<rand>.png；技术元数据进入 figures/.meta/
                 dest_png = stamped_path(self.working_dir / "figures", ".png")
                 client.download(image_url, dest_png, timeout_s=120.0)
         except ArkError as e: