diff --git a/DESIGN.md b/DESIGN.md index 2a38151..bbfbb7f 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -56,6 +56,8 @@ zcbot/ **工作目录** = `workspace/users///`,所有 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//external_results/`，用户明确要求留存时再 export 到 `/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//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` 分段读取或缩小查询范围。 diff --git a/core/external_systems/results.py b/core/external_systems/results.py index f973a53..35e20f8 100644 --- a/core/external_systems/results.py +++ b/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() diff --git a/core/storage/disk_quota.py b/core/storage/disk_quota.py index f13adbd..e890a6b 100644 --- a/core/storage/disk_quota.py +++ b/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(): diff --git a/core/tool_registry.py b/core/tool_registry.py index d931d90..7eecb96 100644 --- a/core/tool_registry.py +++ b/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: diff --git a/tests/test_disk_quota.py b/tests/test_disk_quota.py index 5d927f2..b37c36a 100644 --- a/tests/test_disk_quota.py +++ b/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) diff --git a/tests/test_external_systems.py b/tests/test_external_systems.py index 813106c..feaa4e9 100644 --- a/tests/test_external_systems.py +++ b/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__": diff --git a/tests/test_gpt_image.py b/tests/test_gpt_image.py index 03e9c6c..cf7d1d5 100644 --- a/tests/test_gpt_image.py +++ b/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"]) diff --git a/tests/test_media_common.py b/tests/test_media_common.py index 8c7b9e5..f9be99a 100644 --- a/tests/test_media_common.py +++ b/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): diff --git a/tools/external_systems.py b/tools/external_systems.py index 82e42bc..f91f641 100644 --- a/tools/external_systems.py +++ b/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}" diff --git a/tools/gpt_image.py b/tools/gpt_image.py index 088cede..3e930f7 100644 --- a/tools/gpt_image.py +++ b/tools/gpt_image.py @@ -7,7 +7,7 @@ - 响应直接返 b64_json,无需二次下载; - 复杂图片可能需 ~2min。 完成后: -- 图片落 `/figures/-.png` + 同名 `.meta.json` +- 图片落 `/figures/-.png`，技术元数据进隐藏 `.meta/` - usage_events 写 kind="image" 一行(model_profile="unifyllm.", usage tokens 记进 units → 网关价目公布后可回填对账) """ diff --git a/tools/media_common.py b/tools/media_common.py index 40463f6..d05b25e 100644 --- a/tools/media_common.py +++ b/tools/media_common.py @@ -2,7 +2,7 @@ 五处同构析出,2026-07-23)。 收进来的判据:**逐字或参数化后逐字**的重复 —— 每日配额闸、`-` 落盘命名、 -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), ) diff --git a/tools/seedance.py b/tools/seedance.py index 5cc30e4..bc5870f 100644 --- a/tools/seedance.py +++ b/tools/seedance.py @@ -12,7 +12,7 @@ W×H 由 resolution + ratio 推算(横版 height=resolution_num,竖版 width=res 完成后: - 视频落到 `/videos/-.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 diff --git a/tools/seedream.py b/tools/seedream.py index 6afa9c6..26ec248 100644 --- a/tools/seedream.py +++ b/tools/seedream.py @@ -3,7 +3,7 @@ 模型 ID + 单价 + 默认参数全在 `config/media/doubao.yaml`,本 tool 只装配。 完成后: - 图片落到 `/figures/-.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/-.png + .meta.json + # 落盘 figures/-.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: