diff --git a/agent/verification_stop.py b/agent/verification_stop.py index ac5bc852d0b..dc66944883c 100644 --- a/agent/verification_stop.py +++ b/agent/verification_stop.py @@ -72,64 +72,24 @@ def _filter_verifiable_paths(paths: Iterable[str]) -> list[str]: return [p for p in paths if p and not _is_non_code_path(p)] -# Session identities (platform or source) that are NOT human conversational -# messaging surfaces: interactive coding surfaces (CLI, TUI, desktop, codex, -# local, gateway) and programmatic callers (API server, webhooks, tools). -# Verify-on-stop stays ON by default for these. Any other resolved gateway -# platform is a conversational messaging surface (Telegram, Discord, WhatsApp, -# Signal, Slack, etc.) where the verification narrative would reach a human as -# chat noise, so it defaults OFF. Mirrors LOCAL_SESSION_SOURCE_IDS in -# apps/desktop/src/lib/session-source.ts; keep roughly in sync when adding a -# local or programmatic surface. Default-deny by design: an unrecognized -# identity is treated as messaging (OFF) so a new chat platform never leaks the -# verification receipt before this set is updated. -_NON_MESSAGING_SESSION_SURFACES = frozenset( - { - "", - "cli", - "codex", - "desktop", - "gateway", - "local", - "tui", - "tool", - "api_server", - "webhook", - "msgraph_webhook", - } -) - - def _session_is_messaging_surface() -> bool: - """Return whether this turn is delivered over a human messaging channel. + """Whether this turn is delivered over a human messaging channel. - The gateway binds the platform value (e.g. ``telegram``) to - ``HERMES_SESSION_PLATFORM``; the CLI and TUI set ``HERMES_SESSION_SOURCE`` - (e.g. ``cli``, ``tui``) instead. Both are consulted via the session-context - helper (with an ``os.environ`` fallback), alongside the ``HERMES_PLATFORM`` - override, matching the sibling platform resolution in - ``agent/skill_commands.py`` and ``agent/prompt_builder.py``. A turn is a - messaging surface when a resolved identity is present and is not a known - non-messaging surface. + Verify-on-stop defaults ON for the interactive coding surfaces and + programmatic callers, and OFF on a conversational platform (Telegram, + Discord, Slack, ...) where the verification narrative reaches a human as + chat noise. The surface classification itself is shared with the other + consumers of this distinction — see + ``gateway.session_context.session_is_messaging_surface``. """ try: - from gateway.session_context import get_session_env + from gateway.session_context import session_is_messaging_surface - platform = ( - os.getenv("HERMES_PLATFORM") - or get_session_env("HERMES_SESSION_PLATFORM", "") - ) - source = get_session_env("HERMES_SESSION_SOURCE", "") + return session_is_messaging_surface() except Exception: - platform = os.getenv("HERMES_PLATFORM", "") or os.environ.get( - "HERMES_SESSION_PLATFORM", "" - ) - source = os.environ.get("HERMES_SESSION_SOURCE", "") - for identity in (platform, source): - identity = str(identity or "").strip().lower() - if identity and identity not in _NON_MESSAGING_SESSION_SURFACES: - return True - return False + # The gateway package is unreachable, so there is no messaging channel + # to be on. Reporting a local surface keeps verify-on-stop enabled. + return False def verify_on_stop_enabled(config: dict[str, Any] | None = None) -> bool: diff --git a/gateway/session_context.py b/gateway/session_context.py index cf227fac121..24d556d2cba 100644 --- a/gateway/session_context.py +++ b/gateway/session_context.py @@ -331,6 +331,57 @@ def get_session_env(name: str, default: str = "") -> str: return os.getenv(name, default) +# Surfaces that are not a human chat channel. The gateway binds a platform +# value (``telegram``) to HERMES_SESSION_PLATFORM, while the CLI, TUI, and +# desktop bind HERMES_SESSION_SOURCE (``cli``, ``tui``, ``desktop``) and leave +# the platform empty — so both have to be consulted. ``local``, ``api_server``, +# ``webhook``, and ``msgraph_webhook`` are real Platform values that reach +# HERMES_SESSION_PLATFORM but have no attachment channel behind them. +# Default-deny: an unrecognized identity counts as messaging so a newly added +# chat platform is never treated as a private surface before this set is +# updated. Mirrors LOCAL_SESSION_SOURCE_IDS in +# apps/desktop/src/lib/session-source.ts; keep roughly in sync when adding a +# local or programmatic surface. +NON_MESSAGING_SESSION_SURFACES = frozenset( + { + "", + "api_server", + "cli", + "codex", + "desktop", + "gateway", + "local", + "msgraph_webhook", + "tool", + "tui", + "webhook", + } +) + + +def session_is_messaging_surface() -> bool: + """Whether this turn is delivered over a human messaging channel. + + Callers use this to decide anything that differs between "the user is + reading a chat message" and "the user is at a machine they own": whether + to emit a delivery tag, whether a file has to land somewhere the gateway + is allowed to send from, whether narration would read as chat noise. + + Resolves ``HERMES_PLATFORM``, then the session platform, then the session + source, and reports messaging when any of them names a surface outside + :data:`NON_MESSAGING_SESSION_SURFACES`. + """ + import os + + platform = os.getenv("HERMES_PLATFORM") or get_session_env("HERMES_SESSION_PLATFORM", "") + source = get_session_env("HERMES_SESSION_SOURCE", "") + for identity in (platform, source): + identity = str(identity or "").strip().lower() + if identity and identity not in NON_MESSAGING_SESSION_SURFACES: + return True + return False + + def declare_stateless_channel() -> None: """Declare that this session cannot receive an async background completion. diff --git a/tests/tools/test_flux3_video_tool.py b/tests/tools/test_flux3_video_tool.py index 29e01872bc3..8092c8976f1 100644 --- a/tests/tools/test_flux3_video_tool.py +++ b/tests/tools/test_flux3_video_tool.py @@ -372,6 +372,123 @@ class TestPollTransport: assert parsed["details"]["saved_path"] == str(tmp_path / "flux3-clip-2.mp4") assert (tmp_path / "flux3-clip.mp4").read_bytes() == b"an earlier clip" + def test_on_messaging_the_clip_lands_where_the_gateway_may_send_it(self, monkeypatch): + # A chat user has no filesystem: the attachment is the only way they + # ever see the clip. Downloads is not a delivery root on a strict + # gateway, so a clip saved there is dropped on the way out and the + # reply arrives with nothing attached. + monkeypatch.setenv("HERMES_SESSION_PLATFORM", "telegram") + monkeypatch.setenv("HERMES_MEDIA_DELIVERY_STRICT", "1") + # Strict mode also trusts anything written in the last 10 minutes, and + # a clip we just downloaded is always inside that window. Left on, the + # assertion below passes from any directory on earth and stops being a + # statement about where the clip was saved. + monkeypatch.setenv("HERMES_MEDIA_TRUST_RECENT_FILES", "0") + response = _FakeResponse(200, { + "id": "bfl_job_1", + "status": "Ready", + "result": {"sample": "https://cdn.example/x/flux3-clip.mp4?sig=a"}, + "guidance": "Deliver the saved file.", + }) + + with _fake_download(b"x" * (128 * 1024)): + parsed, _requests = _call(flux3._handle_get_result, {"id": "bfl_job_1"}, response) + + from gateway.platforms.base import validate_media_delivery_path + + saved = parsed["details"]["saved_path"] + assert validate_media_delivery_path(saved), "the gateway must be allowed to send it" + # The exact line to copy, so the path is never retyped from memory. + assert f"\nMEDIA:{saved}\n" in parsed["result"] + + def test_the_offered_tag_is_one_the_gateway_actually_delivers(self, monkeypatch): + # The whole point of spelling the line out is that the model pastes it + # verbatim, so the line has to survive the real extractor. A tag that + # parses but fails validation is the worst outcome: it is stripped from + # the reply either way, so the user is shown a message that looks like + # it simply forgot the attachment. + monkeypatch.setenv("HERMES_SESSION_PLATFORM", "telegram") + response = _FakeResponse(200, { + "id": "bfl_job_1", + "status": "Ready", + "result": {"sample": "https://cdn.example/x/flux3-clip.mp4?sig=a"}, + "guidance": "Deliver the saved file.", + }) + + with _fake_download(b"x" * (128 * 1024)): + parsed, _requests = _call(flux3._handle_get_result, {"id": "bfl_job_1"}, response) + + from gateway.platforms.base import BasePlatformAdapter + + offered = [ln for ln in parsed["result"].splitlines() if ln.startswith("MEDIA:")] + assert len(offered) == 1, "exactly one line to copy" + + reply = f"Here's the clip.\n\n{offered[0]}\n" + media, cleaned = BasePlatformAdapter.extract_media(reply) + assert BasePlatformAdapter.filter_media_delivery_paths(media), "must survive validation" + assert "MEDIA:" not in cleaned, "the tag is consumed, not shown to the user" + + @pytest.mark.parametrize("platform", ["", "cli", "tui", "desktop"]) + def test_off_messaging_the_clip_stays_a_file_and_no_tag_is_offered(self, tmp_path, monkeypatch, platform): + # The CLI has no attachment channel and its prompt forbids the tag — + # emitting one there just prints literal text at the user. + monkeypatch.setenv("HERMES_SESSION_PLATFORM", platform) + response = _FakeResponse(200, { + "id": "bfl_job_1", + "status": "Ready", + "result": {"sample": "https://cdn.example/x/flux3-clip.mp4?sig=a"}, + "guidance": "Deliver the saved file.", + }) + + with _fake_download(b"x" * (128 * 1024)): + parsed, _requests = _call( + flux3._handle_get_result, {"id": "bfl_job_1", "save_to": str(tmp_path)}, response, + ) + + assert parsed["result"].startswith(f"Saved to {tmp_path / 'flux3-clip.mp4'}.") + assert "MEDIA:" not in parsed["result"] + + @pytest.mark.parametrize("platform", ["api_server", "webhook", "msgraph_webhook", "local"]) + def test_platforms_without_an_attachment_channel_are_offered_no_tag(self, tmp_path, monkeypatch, platform): + # These carry a real platform value but no way to attach a file. The + # API server in particular only inlines *images* as data URLs and + # leaves every other MEDIA: tag untouched, so offering one here puts + # the literal text in front of an OpenAI-compatible caller. + monkeypatch.setenv("HERMES_SESSION_PLATFORM", platform) + response = _FakeResponse(200, { + "id": "bfl_job_1", + "status": "Ready", + "result": {"sample": "https://cdn.example/x/flux3-clip.mp4?sig=a"}, + "guidance": "Deliver the saved file.", + }) + + with _fake_download(b"x" * (128 * 1024)): + parsed, _requests = _call( + flux3._handle_get_result, {"id": "bfl_job_1", "save_to": str(tmp_path)}, response, + ) + + assert "MEDIA:" not in parsed["result"] + + def test_a_cli_session_is_recognised_by_its_source(self, tmp_path, monkeypatch): + # The CLI, TUI, and desktop leave HERMES_SESSION_PLATFORM empty and + # identify themselves on HERMES_SESSION_SOURCE instead, so keying only + # on the platform would miss them. + monkeypatch.delenv("HERMES_SESSION_PLATFORM", raising=False) + monkeypatch.setenv("HERMES_SESSION_SOURCE", "tui") + response = _FakeResponse(200, { + "id": "bfl_job_1", + "status": "Ready", + "result": {"sample": "https://cdn.example/x/flux3-clip.mp4?sig=a"}, + "guidance": "Deliver the saved file.", + }) + + with _fake_download(b"x" * (128 * 1024)): + parsed, _requests = _call( + flux3._handle_get_result, {"id": "bfl_job_1", "save_to": str(tmp_path)}, response, + ) + + assert "MEDIA:" not in parsed["result"] + def test_a_rejected_download_fails_loudly_and_leaves_no_file(self, tmp_path): # The original bug: a bad signature returns an XML error body, curl # writes it to the .mp4 and exits 0, and it reads as success. A short diff --git a/tools/flux3_video_tool.py b/tools/flux3_video_tool.py index b6415d703a0..e80f766746f 100644 --- a/tools/flux3_video_tool.py +++ b/tools/flux3_video_tool.py @@ -377,10 +377,29 @@ async def _save_if_ready(raw: str, save_to) -> str: details["saved_path"] = str(target) details["saved_bytes"] = size - payload["result"] = f"Saved to {target}. " + str(payload.get("result") or "") + payload["result"] = _delivery_lead_in(target) + str(payload.get("result") or "") return json.dumps(payload, ensure_ascii=False) +def _delivery_lead_in(target) -> str: + """Opens the result text, ahead of the gateway's own delivery guidance. + + On a messaging platform the tag is spelled out rather than described. The + model has to reproduce this path exactly or the file is not sent, and the + reply is published either way — the tag is stripped from the text whether + or not it named a real file, so a wrong path reads to the user as a + message that simply forgot the attachment. Handing over the finished line + removes the step where that goes wrong; only this side knows the path. + """ + if _delivers_as_an_attachment(): + return ( + f"Saved to {target}. To deliver it, copy the next line into your reply exactly as " + f"written, alone on its own line, with nothing added around it:\n" + f"MEDIA:{target}\n" + ) + return f"Saved to {target}. " + + async def _download_video(url: str, save_to) -> tuple: """Stream the clip to disk, returning (path, bytes). @@ -428,6 +447,47 @@ def _filename_from_url(url: str) -> str: return name or "flux3-video.mp4" +def _delivers_as_an_attachment() -> bool: + """True on a surface where the clip is received rather than opened off disk. + + Deferred to the shared classifier so this tool cannot drift from the rest + of the codebase about what counts as a chat channel. It matters here that + the API server and webhooks are *not* one: they carry a platform value but + no attachment channel, and neither strips an unfulfilled MEDIA: tag out of + the reply, so treating them as messaging puts the literal tag in front of + the caller. + """ + try: + from gateway.session_context import session_is_messaging_surface + + return session_is_messaging_surface() + except Exception: + return False + + +def _default_directory(): + """Where a clip lands when the caller named no location. + + On a messaging platform the user has no filesystem — the only way they + ever see the clip is as an attachment — so it goes to the gateway's own + video cache, which is an unconditionally allowed delivery root. Downloads + is not: an operator running HERMES_MEDIA_DELIVERY_STRICT=1 delivers only + from the cache roots, so a clip saved to Downloads there is dropped on the + way out and the user is shown a reply with nothing attached. + """ + from pathlib import Path + + if _delivers_as_an_attachment(): + try: + from hermes_constants import get_hermes_dir + + return get_hermes_dir("cache/videos", "video_cache") + except Exception: + logger.debug("Could not resolve the video cache dir; using Downloads", exc_info=True) + downloads = Path.home() / "Downloads" + return downloads if downloads.is_dir() else Path.cwd() + + def _resolve_destination(save_to, filename: str): """Where to write, honouring an explicit request and never overwriting.""" from pathlib import Path @@ -439,8 +499,7 @@ def _resolve_destination(save_to, filename: str): else: directory, name = requested.parent, requested.name else: - downloads = Path.home() / "Downloads" - directory, name = (downloads if downloads.is_dir() else Path.cwd()), filename + directory, name = _default_directory(), filename directory.mkdir(parents=True, exist_ok=True) return _free_path(directory / name)