mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-05-19 16:12:30 +00:00
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
This commit is contained in:
chengyongru
parent
8e5715face
commit
62b55ac3f2
@ -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
|
||||
|
||||
@ -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]))
|
||||
@ -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())]
|
||||
@ -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"]
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user