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:
@@ -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]))
|
||||
Reference in New Issue
Block a user