From 62b55ac3f289eaed2edf71e7989d8ee0f2eb4592 Mon Sep 17 00:00:00 2001 From: chengyongru <2755839590@qq.com> Date: Tue, 19 May 2026 01:11:53 +0800 Subject: [PATCH] refactor: remove dead image media attachment code - Remove generated_image_paths_from_messages() and _extract_text_payload() from artifacts.py (no runtime callers) - Remove session_attachments.py entirely (merge_turn_media_into_last_assistant and stage_media_paths_for_session_replay had no runtime callers) - Remove test_session_media_persist.py and the orphaned test in test_artifacts.py --- nanobot/utils/artifacts.py | 39 ------------ nanobot/utils/session_attachments.py | 74 ----------------------- tests/agent/test_session_media_persist.py | 34 ----------- tests/utils/test_artifacts.py | 21 ------- 4 files changed, 168 deletions(-) delete mode 100644 nanobot/utils/session_attachments.py delete mode 100644 tests/agent/test_session_media_persist.py diff --git a/nanobot/utils/artifacts.py b/nanobot/utils/artifacts.py index f01e08942..5f127f44c 100644 --- a/nanobot/utils/artifacts.py +++ b/nanobot/utils/artifacts.py @@ -21,8 +21,6 @@ _MIME_EXTENSIONS = { "image/webp": ".webp", "image/gif": ".gif", } -_GENERATE_IMAGE_TOOL_NAME = "generate_image" - class ArtifactError(ValueError): """Raised when an artifact cannot be safely decoded or stored.""" @@ -124,40 +122,3 @@ def generated_image_tool_result(artifacts: list[dict[str, Any]]) -> str: ) -def _extract_text_payload(content: Any) -> str | None: - if isinstance(content, str): - return content - if isinstance(content, list): - parts: list[str] = [] - for block in content: - if isinstance(block, dict) and isinstance(block.get("text"), str): - parts.append(block["text"]) - return "\n".join(parts) if parts else None - return None - - -def generated_image_paths_from_messages(messages: list[dict[str, Any]]) -> list[str]: - """Collect generated image artifact paths from generate_image tool results.""" - paths: list[str] = [] - seen: set[str] = set() - for message in messages: - if message.get("role") != "tool" or message.get("name") != _GENERATE_IMAGE_TOOL_NAME: - continue - payload = _extract_text_payload(message.get("content")) - if not payload: - continue - try: - data = json.loads(payload) - except json.JSONDecodeError: - continue - artifacts = data.get("artifacts") if isinstance(data, dict) else None - if not isinstance(artifacts, list): - continue - for artifact in artifacts: - if not isinstance(artifact, dict): - continue - path = artifact.get("path") - if isinstance(path, str) and path and path not in seen: - paths.append(path) - seen.add(path) - return paths diff --git a/nanobot/utils/session_attachments.py b/nanobot/utils/session_attachments.py deleted file mode 100644 index d761d33b3..000000000 --- a/nanobot/utils/session_attachments.py +++ /dev/null @@ -1,74 +0,0 @@ -"""Session replay: ensure assistant ``media`` paths are under the media root. - -WebUI history signing (``/api/.../messages``) only works for files inside -``get_media_dir``. Tool-driven attachments may live in the workspace; stage -copies into the websocket media bucket before persisting message JSON. -""" - -from __future__ import annotations - -import shutil -import uuid -from pathlib import Path -from typing import Any - -from loguru import logger - -from nanobot.config.paths import get_media_dir -from nanobot.utils.helpers import safe_filename - - -def stage_media_paths_for_session_replay(paths: list[str]) -> list[str]: - """Keep local files only; copy anything outside the media root into ``media/websocket``.""" - root = get_media_dir().resolve() - out: list[str] = [] - seen: set[str] = set() - for raw in paths: - if not isinstance(raw, str) or not raw.strip(): - continue - if raw.startswith(("http://", "https://")): - continue - try: - p = Path(raw).expanduser().resolve() - except OSError: - continue - if not p.is_file(): - continue - try: - p.relative_to(root) - key = str(p) - except ValueError: - try: - media_dir = get_media_dir("websocket") - staged = media_dir / f"{uuid.uuid4().hex[:12]}-{safe_filename(p.name) or 'attachment'}" - shutil.copyfile(p, staged) - key = str(staged.resolve()) - except OSError as exc: - logger.warning("failed to stage session media from {}: {}", raw, exc) - continue - if key not in seen: - out.append(key) - seen.add(key) - return out - - -def merge_turn_media_into_last_assistant( - all_messages: list[dict[str, Any]], - generated_image_paths: list[str], - extra_attachment_paths: list[str], -) -> None: - """Attach staged paths to the last assistant row in *all_messages* (in-place).""" - merged = list( - dict.fromkeys( - [ - *stage_media_paths_for_session_replay(generated_image_paths), - *stage_media_paths_for_session_replay(extra_attachment_paths), - ] - ) - ) - last = all_messages[-1] if all_messages else None - if not merged or not last or last.get("role") != "assistant": - return - existing = last.get("media") - base = existing if isinstance(existing, list) else [] - last["media"] = list(dict.fromkeys([*base, *merged])) diff --git a/tests/agent/test_session_media_persist.py b/tests/agent/test_session_media_persist.py deleted file mode 100644 index 98b77ffd1..000000000 --- a/tests/agent/test_session_media_persist.py +++ /dev/null @@ -1,34 +0,0 @@ -"""Tests for staging attachment paths into the media bucket for session replay.""" - -from pathlib import Path - -from nanobot.config.loader import set_config_path -from nanobot.config.paths import get_media_dir -from nanobot.utils.session_attachments import stage_media_paths_for_session_replay - - -def test_persist_media_stages_workspace_file(tmp_path: Path) -> None: - set_config_path(tmp_path / "config.json") - outside = tmp_path / "workspace" / "report.md" - outside.parent.mkdir(parents=True) - outside.write_text("body", encoding="utf-8") - - out = stage_media_paths_for_session_replay([str(outside)]) - - assert len(out) == 1 - staged = Path(out[0]) - assert staged.is_file() - assert staged.read_text(encoding="utf-8") == "body" - assert staged.resolve().is_relative_to(get_media_dir().resolve()) - - -def test_persist_media_keeps_files_already_under_media_root(tmp_path: Path) -> None: - set_config_path(tmp_path / "config.json") - media = get_media_dir("websocket") - media.mkdir(parents=True, exist_ok=True) - inside = media / "keep-me.txt" - inside.write_text("x", encoding="utf-8") - - out = stage_media_paths_for_session_replay([str(inside.resolve())]) - - assert out == [str(inside.resolve())] diff --git a/tests/utils/test_artifacts.py b/tests/utils/test_artifacts.py index 54c9b222a..941c1a40d 100644 --- a/tests/utils/test_artifacts.py +++ b/tests/utils/test_artifacts.py @@ -10,8 +10,6 @@ from nanobot.config.loader import set_config_path from nanobot.utils.artifacts import ( ArtifactError, decode_image_data_url, - generated_image_paths_from_messages, - generated_image_tool_result, store_generated_image_artifact, ) @@ -66,22 +64,3 @@ def test_store_generated_image_artifact_rejects_unsafe_save_dir(tmp_path: Path) model="m", save_dir="../outside", ) - - -def test_generated_image_paths_from_tool_results() -> None: - result = generated_image_tool_result( - [ - {"id": "img_1", "path": "/tmp/one.png"}, - {"id": "img_2", "path": "/tmp/two.png"}, - ] - ) - payload = json.loads(result) - - assert generated_image_paths_from_messages( - [ - {"role": "tool", "name": "generate_image", "content": result}, - {"role": "tool", "name": "other", "content": result}, - ] - ) == ["/tmp/one.png", "/tmp/two.png"] - assert "Call the message tool" in payload["next_step"] - assert "media parameter" in payload["next_step"]