Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
165 changes: 165 additions & 0 deletions tests/tools/test_video_generation_artifacts.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,165 @@
"""Video artifact path annotations — mirror image_generate's agent_visible contract.

Under docker/ssh/modal local video bytes land in a relocated Hermes cache.
``video`` stays the host/gateway path; ``agent_visible_video`` is for
terminal/file follow-up inside the sandbox.
"""

import json
from types import SimpleNamespace


def test_postprocess_adds_agent_visible_video_for_active_ssh_env(monkeypatch, tmp_path):
from tools import video_generation_tool

hermes_home = tmp_path / ".hermes"
video_dir = hermes_home / "cache" / "videos"
video_dir.mkdir(parents=True)
video_path = video_dir / "deepinfra_clip.mp4"
video_path.write_bytes(b"ftyp")

sync_calls = []

class FakeSyncManager:
def sync(self, *, force=False):
sync_calls.append(force)

env = SimpleNamespace(
_remote_home="/home/remotesshuser",
_sync_manager=FakeSyncManager(),
)

monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setattr(
"tools.image_generation_tool._active_terminal_env",
lambda task_id: env,
)

raw = json.dumps({"success": True, "video": str(video_path)})
result = json.loads(
video_generation_tool._postprocess_video_generate_result(raw, task_id="task-1")
)

assert result["video"] == str(video_path)
assert result["host_video"] == str(video_path)
assert result["agent_visible_video"] == (
"/home/remotesshuser/.hermes/cache/videos/deepinfra_clip.mp4"
)
assert sync_calls == [True]


def test_postprocess_adds_docker_root_hermes_without_env(monkeypatch, tmp_path):
from tools import video_generation_tool

hermes_home = tmp_path / ".hermes"
video_dir = hermes_home / "cache" / "videos"
video_dir.mkdir(parents=True)
video_path = video_dir / "clip.mp4"
video_path.write_bytes(b"ftyp")

monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setenv("TERMINAL_ENV", "docker")
monkeypatch.setattr(
"tools.image_generation_tool._active_terminal_env",
lambda task_id: None,
)

raw = json.dumps({"success": True, "video": str(video_path)})
result = json.loads(video_generation_tool._postprocess_video_generate_result(raw))

assert result["video"] == str(video_path)
assert result["agent_visible_video"] == "/root/.hermes/cache/videos/clip.mp4"


def test_postprocess_noop_on_local_backend(monkeypatch, tmp_path):
from tools import video_generation_tool

hermes_home = tmp_path / ".hermes"
video_dir = hermes_home / "cache" / "videos"
video_dir.mkdir(parents=True)
video_path = video_dir / "local.mp4"
video_path.write_bytes(b"ftyp")

monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setenv("TERMINAL_ENV", "local")
monkeypatch.setattr(
"tools.image_generation_tool._active_terminal_env",
lambda task_id: None,
)

raw = json.dumps({"success": True, "video": str(video_path)})
result = json.loads(video_generation_tool._postprocess_video_generate_result(raw))

assert result == {"success": True, "video": str(video_path)}
assert "agent_visible_video" not in result


def test_postprocess_noop_on_http_url(monkeypatch):
from tools import video_generation_tool

monkeypatch.setenv("TERMINAL_ENV", "docker")
monkeypatch.setattr(
"tools.image_generation_tool._active_terminal_env",
lambda task_id: None,
)

raw = json.dumps({
"success": True,
"video": "https://cdn.example/clip.mp4",
})
result = json.loads(video_generation_tool._postprocess_video_generate_result(raw))

assert result == {
"success": True,
"video": "https://cdn.example/clip.mp4",
}
assert "agent_visible_video" not in result


def test_handle_video_generate_postprocesses_local_result(monkeypatch, tmp_path):
from tools import video_generation_tool

hermes_home = tmp_path / ".hermes"
video_dir = hermes_home / "cache" / "videos"
video_dir.mkdir(parents=True)
video_path = video_dir / "handled.mp4"
video_path.write_bytes(b"ftyp")

monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setenv("TERMINAL_ENV", "docker")

class FakeProvider:
name = "fake"

def default_model(self):
return "fake-model"

def generate(self, prompt, **kwargs):
return {"success": True, "video": str(video_path), "prompt": prompt}

monkeypatch.setattr(
video_generation_tool, "_resolve_active_provider", lambda: FakeProvider()
)
monkeypatch.setattr(
video_generation_tool, "_read_configured_video_provider", lambda: "fake"
)
monkeypatch.setattr(
video_generation_tool, "_read_configured_video_model", lambda: "fake-model"
)
monkeypatch.setattr(
"tools.image_generation_tool._confine_source_images",
lambda *a, **k: (None, None, None),
)
monkeypatch.setattr(
"tools.image_generation_tool._active_terminal_env",
lambda task_id: None,
)

out = json.loads(
video_generation_tool._handle_video_generate(
{"prompt": "a cat walks"},
task_id="t1",
)
)
assert out["video"] == str(video_path)
assert out["agent_visible_video"] == "/root/.hermes/cache/videos/handled.mp4"
55 changes: 50 additions & 5 deletions tools/video_generation_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -407,7 +407,49 @@ def _handle_video_generate(args: Dict[str, Any], **_kw: Any) -> str:
prompt=prompt,
))

return json.dumps(result)
return _postprocess_video_generate_result(json.dumps(result), task_id=task_id)


def _postprocess_video_generate_result(raw: str, task_id: str | None = None) -> str:
"""Annotate successful local video results with backend-visible paths.

``video`` remains the host/gateway-deliverable path (or URL). When the
active terminal backend has a different filesystem,
``agent_visible_video`` gives the path the agent can use with
terminal/file tools — same contract as ``image_generate``'s
``agent_visible_image`` and ``text_to_speech``'s
``agent_visible_file_path``.
"""
try:
payload = json.loads(raw) if isinstance(raw, str) else raw
except Exception:
return raw

if not isinstance(payload, dict) or not payload.get("success"):
return raw

from tools.image_generation_tool import (
_active_terminal_env,
_agent_visible_cache_path,
_force_artifact_sync,
_looks_like_absolute_file_path,
)

video = payload.get("video")
if not isinstance(video, str) or not _looks_like_absolute_file_path(video):
return raw

env = _active_terminal_env(task_id)
agent_path = _agent_visible_cache_path(video, env)
if not agent_path or agent_path == video:
return raw

if env is not None:
_force_artifact_sync(env)

payload.setdefault("host_video", video)
payload.setdefault("agent_visible_video", agent_path)
return json.dumps(payload, ensure_ascii=False)


# ---------------------------------------------------------------------------
Expand Down Expand Up @@ -437,10 +479,13 @@ def _handle_video_generate(args: Dict[str, Any], **_kw: Any) -> str:
"`hermes tools` → Video Generation; the agent does not pick them. "
"Long-running generations may take 30 seconds to several minutes — "
"the call blocks until the video is ready. Returns the result in the "
"`video` field — either an HTTP URL or an absolute file path. To show "
"it to the user, reference that path/URL in your response using the "
"file-delivery convention for the current platform (your platform "
"guidance describes how files are delivered here)."
"`video` field — either an HTTP URL or an absolute file path. Under a "
"sandbox terminal backend, local-file results may also include "
"`agent_visible_video` for terminal/file follow-up (`video` stays the "
"host/gateway path). To show it to the user, reference that path/URL "
"in your response using the file-delivery convention for the current "
"platform (your platform guidance describes how files are delivered "
"here)."
)


Expand Down
Loading