feat: carry compression transport upgrades from upstream #98205

agent/trajectory.py

@@ -1,6 +1,8 @@
 """Trajectory saving + scratchpad helpers (``_convert_to_trajectory_format`` stays an AIAgent method — batch_runner.py calls it)."""
 
 import json
+import gzip
+import io
 import logging
 import os
 from datetime import datetime
@@ -28,29 +30,63 @@ def _lock_append_handle(f, acquire: bool) -> None:
         import msvcrt
         f.seek(0)
         msvcrt.locking(f.fileno(), msvcrt.LK_LOCK if acquire else msvcrt.LK_UNLCK, 1)
-        f.seek(0, os.SEEK_END)
     else:
         import fcntl
         fcntl.flock(f.fileno(), fcntl.LOCK_EX if acquire else fcntl.LOCK_UN)
 
 
+def _build_gzip_member(line: str) -> bytes:
+    """Return one complete gzip member without touching the destination file."""
+    member = io.BytesIO()
+    with gzip.GzipFile(fileobj=member, mode="wb") as compressed:
+        compressed.write(line.encode("utf-8"))
+    return member.getvalue()
+
+
 def save_trajectory(trajectory: List[Dict[str, Any]], model: str, completed: bool, filename: str = None):
-    """Append a ShareGPT-format entry to a JSONL file (default trajectory_samples.jsonl / failed_trajectories.jsonl by ``completed``)."""
+    """Append a ShareGPT-format entry, gzip-compressed by default."""
     if filename is None:
-        filename = "trajectory_samples.jsonl" if completed else "failed_trajectories.jsonl"
+        filename = "trajectory_samples.jsonl.gz" if completed else "failed_trajectories.jsonl.gz"
     entry = {"conversations": trajectory, "timestamp": datetime.now().isoformat(), "model": model, "completed": completed}
     try:
         line = json.dumps(entry, ensure_ascii=False) + "\n"  # serialize before taking the lock
-        with open(filename, "a", encoding="utf-8") as f:
-            # Gateway sessions and batch workers append to the SAME default file; without an
-            # exclusive lock around write+flush, entries larger than one write() interleave and the
-            # JSONL stops parsing (#12684).
-            _lock_append_handle(f, True)
-            try:
-                f.write(line)
-                f.flush()
-            finally:
-                _lock_append_handle(f, False)
+        is_gzip = str(filename).endswith(".gz")
+        if is_gzip:
+            # Finalize the complete gzip member away from the destination. A
+            # killed worker can then leave no half-written member for the next
+            # append to consume.
+            payload = _build_gzip_member(line)
+            with open(filename, "ab") as raw:
+                locked = False
+                try:
+                    # Lock the stable raw descriptor for the one-shot append;
+                    # Windows locking must never use a moving gzip wrapper.
+                    _lock_append_handle(raw, True)
+                    locked = True
+                    raw.write(payload)
+                    raw.flush()
+                    locked = False
+                finally:
+                    if locked:
+                        try:
+                            _lock_append_handle(raw, False)
+                        except (OSError, ValueError):
+                            pass
+        else:
+            with open(filename, "a", encoding="utf-8") as text_file:
+                locked = False
+                try:
+                    _lock_append_handle(text_file, True)
+                    locked = True
+                    text_file.write(line)
+                    text_file.flush()
+                    locked = False
+                finally:
+                    if locked:
+                        try:
+                            _lock_append_handle(text_file, False)
+                        except (OSError, ValueError):
+                            pass
         logger.info("Trajectory saved to %s", filename)
     except Exception as e:
         logger.warning("Failed to save trajectory: %s", e)

hermes_cli/response_compression.py

@@ -0,0 +1,187 @@
+"""Selective lossless compression for HTTP responses.
+
+Only data formats that are normally text-heavy and safe to compress are
+included. HTML is intentionally excluded because the dashboard HTML carries
+session material and may contain attacker-influenced values; JSON API
+responses and exports are the primary bandwidth target.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from starlette.datastructures import Headers
+from starlette.middleware.gzip import GZipMiddleware, GZipResponder
+
+
+_COMPRESSIBLE_CONTENT_TYPES = {
+    "application/json",
+    "application/javascript",
+    "application/xml",
+    "image/svg+xml",
+    "text/css",
+    "text/plain",
+}
+_MEDIA_TYPE_RE = re.compile(
+    r"^[!#$%&'*+.^_`|~0-9A-Za-z-]+/[!#$%&'*+.^_`|~0-9A-Za-z-]+$"
+)
+_QVALUE_RE = re.compile(r"^(?:0(?:\.[0-9]{0,3})?|1(?:\.0{0,3})?)$")
+
+# Exact responses that can contain unredacted credentials or arbitrary local
+# file/log content. Query strings are not part of ASGI ``scope['path']``.
+_EXCLUDED_EXACT_PATHS = {
+    "/api/env/reveal",
+    "/api/files/download",
+    "/api/files/read",
+    "/api/fs/download",
+    "/api/fs/read-data-url",
+    "/api/fs/read-text",
+    "/api/logs",
+    "/api/media",
+    "/api/ops/backup/download",
+}
+# Route families that contain one-time tokens, secrets, raw configuration, or
+# authentication/session material. Match only the root or a slash descendant.
+_EXCLUDED_PATH_TREES = (
+    "/api/actions",
+    "/api/auth",
+    "/api/audio/voice-config",
+    "/api/config",
+    "/api/mcp/oauth",
+    "/api/mcp/servers",
+    "/api/messaging/telegram/onboarding",
+    "/api/messaging/whatsapp/onboarding",
+    "/api/pairing",
+    "/api/providers/oauth",
+    "/api/webhooks",
+    "/auth",
+)
+
+
+def _is_compressible_content_type(content_types: list[str]) -> bool:
+    if len(content_types) != 1:
+        return False
+    content_type = content_types[0]
+    media_type = content_type.split(";", 1)[0].strip().lower()
+    if not _MEDIA_TYPE_RE.fullmatch(media_type):
+        return False
+    if media_type in _COMPRESSIBLE_CONTENT_TYPES:
+        return True
+    top_level, subtype = media_type.split("/", 1)
+    return top_level == "application" and subtype.endswith("+json")
+
+
+def _quality(parameters: list[str]) -> float:
+    quality = 1.0
+    seen = False
+    for parameter in parameters:
+        name, separator, raw_value = parameter.partition("=")
+        if name.lower() != "q" or seen or not separator:
+            return 0.0
+        seen = True
+        if not _QVALUE_RE.fullmatch(raw_value):
+            return 0.0
+        quality = float(raw_value)
+    return quality
+
+
+def _accepts_gzip(values: list[str]) -> bool:
+    """Return whether ``Accept-Encoding`` permits gzip compression."""
+    gzip_qualities: list[float] = []
+    wildcard_qualities: list[float] = []
+    for value in values:
+        for item in value.split(","):
+            parts = [part.strip() for part in item.split(";")]
+            encoding = parts[0].lower()
+            if not encoding:
+                continue
+            quality = _quality(parts[1:])
+            if encoding == "gzip":
+                gzip_qualities.append(quality)
+            elif encoding == "*":
+                wildcard_qualities.append(quality)
+    if gzip_qualities:
+        return all(quality > 0 for quality in gzip_qualities)
+    return bool(wildcard_qualities) and all(
+        quality > 0 for quality in wildcard_qualities
+    )
+
+
+def _is_excluded_path(path: str) -> bool:
+    if path in _EXCLUDED_EXACT_PATHS:
+        return True
+    return any(
+        path == root or path.startswith(f"{root}/")
+        for root in _EXCLUDED_PATH_TREES
+    )
+
+
+class _SelectiveGZipResponder(GZipResponder):
+    """Starlette's streaming gzip responder with a content-type allowlist."""
+
+    async def _send_with_selective_compression(self, message: dict[str, Any]) -> None:
+        if message["type"] == "http.response.start":
+            # Delay response.start until the first body chunk, as the parent
+            # responder does, but make the exclusion decision from our safer
+            # allowlist rather than Starlette's event-stream-only default.
+            self.initial_message = message
+            headers = Headers(raw=self.initial_message["headers"])
+            self.content_encoding_set = "content-encoding" in headers
+            self.content_type_is_excluded = (
+                message["status"] == 206
+                or "content-range" in headers
+                or "etag" in headers
+                or not _is_compressible_content_type(
+                    headers.getlist("content-type")
+                )
+            )
+            return
+
+        # Older Starlette responders do not consult ``content_type_is_excluded``
+        # themselves. Bypass the parent compressor explicitly so excluded
+        # responses (including 206/range and ETagged bodies) retain their
+        # original headers and bytes on every supported API generation.
+        if self.content_type_is_excluded:
+            if not self.started:
+                self.started = True
+                await self.send(self.initial_message)
+            await self.send(message)
+            return
+
+        # Starlette renamed this hook from ``send_with_gzip`` to
+        # ``send_with_compression``. Resolve the parent hook dynamically so the
+        # middleware keeps its allowlist on both supported API generations.
+        parent = super()
+        send_hook = getattr(parent, "send_with_compression", None)
+        if send_hook is None:
+            send_hook = parent.send_with_gzip
+        await send_hook(message)
+
+    async def send_with_compression(self, message: dict[str, Any]) -> None:
+        await self._send_with_selective_compression(message)
+
+    async def send_with_gzip(self, message: dict[str, Any]) -> None:
+        await self._send_with_selective_compression(message)
+
+
+class SelectiveGZipMiddleware(GZipMiddleware):
+    """Compress eligible JSON/text responses while preserving streaming."""
+
+    async def __call__(self, scope, receive, send):
+        if (
+            scope["type"] != "http"
+            or _is_excluded_path(scope.get("path", ""))
+            or not _accepts_gzip(
+                Headers(scope=scope).getlist("Accept-Encoding")
+            )
+        ):
+            await self.app(scope, receive, send)
+            return
+
+        responder = _SelectiveGZipResponder(
+            self.app,
+            self.minimum_size,
+            compresslevel=self.compresslevel,
+        )
+        await responder(scope, receive, send)

hermes_cli/web_server.py

@@ -39,6 +39,7 @@
     from fastapi import FastAPI, HTTPException, Request
     from fastapi.middleware.cors import CORSMiddleware
     from fastapi.responses import JSONResponse
+    from hermes_cli.response_compression import SelectiveGZipMiddleware
 except ImportError:
     # First try lazy-installing the dashboard extras. Only the user actually
     # running `hermes dashboard` needs fastapi+uvicorn; lazy install keeps
@@ -49,6 +50,7 @@
         from fastapi import FastAPI, HTTPException, Request
         from fastapi.middleware.cors import CORSMiddleware
         from fastapi.responses import JSONResponse
+        from hermes_cli.response_compression import SelectiveGZipMiddleware
     except Exception:
         raise SystemExit(
             "Web UI requires fastapi and uvicorn.\n"
@@ -395,6 +397,7 @@ def _ssh_runtime_intact() -> bool:
     allow_methods=["*"],
     allow_headers=["*"],
 )
+app.add_middleware(SelectiveGZipMiddleware, minimum_size=1024, compresslevel=6)
 
 # Endpoints that do NOT require the session token; everything else under /api/
 # is gated below. Shared with the OAuth gate so the two allowlists cannot

run_agent.py

@@ -1451,8 +1451,8 @@ def main(
                               Multiple toolsets can be combined: "web,vision"
         disabled_toolsets (str): Comma-separated list of toolsets to disable (e.g., "terminal")
         list_tools (bool): Just list available tools and exit
-        save_trajectories (bool): Save conversation trajectories to JSONL files (appends to
-        trajectory_samples.jsonl). Defaults to False.
+        save_trajectories (bool): Save conversation trajectories to gzip-compressed JSONL files
+        (appends to trajectory_samples.jsonl.gz). Defaults to False.
         save_sample (bool): Save a single trajectory sample to a UUID-named JSONL file for inspection.
         Defaults to False.
         verbose (bool): Enable verbose logging for debugging. Defaults to False.
@@ -1471,8 +1471,8 @@ def main(
     disabled_toolsets_list = _parse_toolset_arg(disabled_toolsets, "🚫 Disabled toolsets")
     if save_trajectories:
         print("💾 Trajectory saving: ENABLED")
-        print("   - Successful conversations → trajectory_samples.jsonl")
-        print("   - Failed conversations → failed_trajectories.jsonl")
+        print("   - Successful conversations → trajectory_samples.jsonl.gz")
+        print("   - Failed conversations → failed_trajectories.jsonl.gz")
 
     try:
         agent = AIAgent(