diff --git a/agent/model_metadata.py b/agent/model_metadata.py index b8ec0d6509e4b..956bb816b9385 100644 --- a/agent/model_metadata.py +++ b/agent/model_metadata.py @@ -16,7 +16,7 @@ import requests import yaml -from utils import base_url_host_matches, base_url_hostname +from utils import base_url_host_matches, base_url_hostname, atomic_yaml_write from hermes_constants import OPENROUTER_MODELS_URL @@ -844,9 +844,7 @@ def save_context_length(model: str, base_url: str, length: int) -> None: cache[key] = length path = _get_context_cache_path() try: - path.parent.mkdir(parents=True, exist_ok=True) - with open(path, "w", encoding="utf-8") as f: - yaml.dump({"context_lengths": cache}, f, default_flow_style=False) + atomic_yaml_write(path, {"context_lengths": cache}) logger.info("Cached context length %s -> %s tokens", key, f"{length:,}") except Exception as e: logger.debug("Failed to save context length cache: %s", e) @@ -868,9 +866,7 @@ def _invalidate_cached_context_length(model: str, base_url: str) -> None: del cache[key] path = _get_context_cache_path() try: - path.parent.mkdir(parents=True, exist_ok=True) - with open(path, "w", encoding="utf-8") as f: - yaml.dump({"context_lengths": cache}, f, default_flow_style=False) + atomic_yaml_write(path, {"context_lengths": cache}) except Exception as e: logger.debug("Failed to invalidate context length cache entry %s: %s", key, e) diff --git a/gateway/delivery.py b/gateway/delivery.py index 41a25c56de03e..3f25235801c29 100644 --- a/gateway/delivery.py +++ b/gateway/delivery.py @@ -9,6 +9,8 @@ """ import logging +import os +import tempfile from pathlib import Path from datetime import datetime from dataclasses import dataclass @@ -207,7 +209,23 @@ def _deliver_local( lines.append("") lines.append(content) - output_path.write_text("\n".join(lines)) + output_path.parent.mkdir(parents=True, exist_ok=True) + content_str = "\n".join(lines) + fd, tmp_path = tempfile.mkstemp( + dir=str(output_path.parent), prefix=".delivery_", suffix=".tmp" + ) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(content_str) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp_path, output_path) + except BaseException: + try: + os.unlink(tmp_path) + except OSError: + pass + raise return { "path": str(output_path), diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 0668896e170f7..eae48a8d4bec7 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -374,7 +374,10 @@ def get(self, response_id: str) -> Optional[Dict[str, Any]]: (time.time(), response_id), ) self._conn.commit() - return json.loads(row[0]) + try: + return json.loads(row[0]) + except (json.JSONDecodeError, ValueError): + return None def put(self, response_id: str, data: Dict[str, Any]) -> None: """Store a response, evicting the oldest if at capacity.""" diff --git a/gateway/platforms/weixin.py b/gateway/platforms/weixin.py index 1c9fec0af7fb2..c71466188da96 100644 --- a/gateway/platforms/weixin.py +++ b/gateway/platforms/weixin.py @@ -383,7 +383,10 @@ async def _api_post( raw = await response.text() if not response.ok: raise RuntimeError(f"iLink POST {endpoint} HTTP {response.status}: {raw[:200]}") - return json.loads(raw) + try: + return json.loads(raw) + except (json.JSONDecodeError, ValueError) as exc: + raise RuntimeError(f"iLink POST {endpoint} returned non-JSON: {raw[:200]}") from exc async def _api_get( @@ -403,7 +406,10 @@ async def _api_get( raw = await response.text() if not response.ok: raise RuntimeError(f"iLink GET {endpoint} HTTP {response.status}: {raw[:200]}") - return json.loads(raw) + try: + return json.loads(raw) + except (json.JSONDecodeError, ValueError) as exc: + raise RuntimeError(f"iLink GET {endpoint} returned non-JSON: {raw[:200]}") from exc async def _get_updates( diff --git a/gateway/run.py b/gateway/run.py index cca9901cb4263..46990cc9e5128 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -1740,9 +1740,7 @@ def _load_voice_modes(self) -> Dict[str, str]: def _save_voice_modes(self) -> None: try: self._VOICE_MODE_PATH.parent.mkdir(parents=True, exist_ok=True) - self._VOICE_MODE_PATH.write_text( - json.dumps(self._voice_mode, indent=2) - ) + atomic_json_write(self._VOICE_MODE_PATH, self._voice_mode) except OSError as e: logger.warning("Failed to save voice modes: %s", e) diff --git a/tools/browser_cdp_tool.py b/tools/browser_cdp_tool.py index f10a15419233b..4ee247b3986d2 100644 --- a/tools/browser_cdp_tool.py +++ b/tools/browser_cdp_tool.py @@ -140,7 +140,10 @@ async def _cdp_call( f"Timed out attaching to target {target_id}" ) raw = await asyncio.wait_for(ws.recv(), timeout=remaining) - msg = json.loads(raw) + try: + msg = json.loads(raw) + except (json.JSONDecodeError, ValueError): + continue # skip malformed CDP frames if msg.get("id") == attach_id: if "error" in msg: raise RuntimeError( @@ -174,7 +177,10 @@ async def _cdp_call( f"Timed out waiting for response to {method}" ) raw = await asyncio.wait_for(ws.recv(), timeout=remaining) - msg = json.loads(raw) + try: + msg = json.loads(raw) + except (json.JSONDecodeError, ValueError): + continue # skip malformed CDP frames if msg.get("id") == call_id: if "error" in msg: raise RuntimeError(f"CDP error: {msg['error']}") diff --git a/tools/checkpoint_manager.py b/tools/checkpoint_manager.py index 16ce12fc60ef7..4d3ad267f4f0d 100644 --- a/tools/checkpoint_manager.py +++ b/tools/checkpoint_manager.py @@ -58,6 +58,7 @@ import time from pathlib import Path from hermes_constants import get_hermes_home +from utils import atomic_json_write from typing import Dict, List, Optional, Set, Tuple logger = logging.getLogger(__name__) @@ -466,7 +467,7 @@ def _register_project(store: Path, working_dir: str) -> None: pass try: meta_path.parent.mkdir(parents=True, exist_ok=True) - meta_path.write_text(json.dumps(meta), encoding="utf-8") + atomic_json_write(meta_path, meta) except OSError as exc: logger.debug("Could not write project metadata %s: %s", meta_path, exc) @@ -488,7 +489,7 @@ def _touch_project(store: Path, working_dir: str) -> None: meta["last_touch"] = time.time() meta.setdefault("created_at", meta["last_touch"]) try: - meta_path.write_text(json.dumps(meta), encoding="utf-8") + atomic_json_write(meta_path, meta) except OSError as exc: logger.debug("Could not update project metadata %s: %s", meta_path, exc) diff --git a/tools/environments/base.py b/tools/environments/base.py index 8a53cefb5bf79..41bdfc804c860 100644 --- a/tools/environments/base.py +++ b/tools/environments/base.py @@ -18,6 +18,8 @@ import uuid from abc import ABC, abstractmethod from pathlib import Path + +from utils import atomic_json_write from typing import IO, Callable, Protocol from hermes_constants import get_hermes_home @@ -166,8 +168,7 @@ def _load_json_store(path: Path) -> dict: def _save_json_store(path: Path, data: dict) -> None: """Write *data* as pretty-printed JSON to *path*.""" - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(json.dumps(data, indent=2), encoding="utf-8") + atomic_json_write(path, data) def _file_mtime_key(host_path: str) -> tuple[float, int] | None: diff --git a/tools/osv_check.py b/tools/osv_check.py index e094b2721045b..8b232ed194bc5 100644 --- a/tools/osv_check.py +++ b/tools/osv_check.py @@ -148,7 +148,10 @@ def _query_osv( ) with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp: - result = json.loads(resp.read()) + try: + result = json.loads(resp.read()) + except (json.JSONDecodeError, ValueError): + return [] vulns = result.get("vulns", []) # Only malware advisories — ignore regular CVEs diff --git a/tools/skills_hub.py b/tools/skills_hub.py index 7725c745de45f..cd65488aec717 100644 --- a/tools/skills_hub.py +++ b/tools/skills_hub.py @@ -32,6 +32,7 @@ import httpx import yaml +from utils import atomic_json_write from tools.skills_guard import ( ScanResult, content_hash, TRUSTED_REPOS, ) @@ -2770,7 +2771,7 @@ def load(self) -> dict: def save(self, data: dict) -> None: self.path.parent.mkdir(parents=True, exist_ok=True) - self.path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n") + atomic_json_write(self.path, data) def record_install( self, @@ -2837,7 +2838,7 @@ def load(self) -> List[dict]: def save(self, taps: List[dict]) -> None: self.path.parent.mkdir(parents=True, exist_ok=True) - self.path.write_text(json.dumps({"taps": taps}, indent=2) + "\n") + atomic_json_write(self.path, {"taps": taps}) def add(self, repo: str, path: str = "skills/") -> bool: """Add a tap. Returns False if already exists.""" @@ -2891,11 +2892,11 @@ def ensure_hub_dirs() -> None: QUARANTINE_DIR.mkdir(exist_ok=True) INDEX_CACHE_DIR.mkdir(exist_ok=True) if not LOCK_FILE.exists(): - LOCK_FILE.write_text('{"version": 1, "installed": {}}\n') + atomic_json_write(LOCK_FILE, {"version": 1, "installed": {}}) if not AUDIT_LOG.exists(): AUDIT_LOG.touch() if not TAPS_FILE.exists(): - TAPS_FILE.write_text('{"taps": []}\n') + atomic_json_write(TAPS_FILE, {"taps": []}) def quarantine_bundle(bundle: SkillBundle) -> Path: diff --git a/tools/vision_tools.py b/tools/vision_tools.py index 912777e2e255d..c8d07406f569a 100644 --- a/tools/vision_tools.py +++ b/tools/vision_tools.py @@ -346,62 +346,65 @@ def _resize_image_for_vision(image_path: Path, mime_type: Optional[str] = None, if data_url is None: data_url = _image_to_base64_data_url(image_path, mime_type=mime_type) return data_url # fall through to size-check in caller - # Convert RGBA to RGB for JPEG output - if pil_format == "JPEG" and img.mode in {"RGBA", "P"}: - img = img.convert("RGB") - - # Strategy: halve dimensions until base64 fits, up to 4 rounds. - # For JPEG, also try reducing quality at each size step. - # For PNG, quality is irrelevant — only dimension reduction helps. - quality_steps = (85, 70, 50) if pil_format == "JPEG" else (None,) - prev_dims = (img.width, img.height) - candidate = None # will be set on first loop iteration - - for attempt in range(5): - if attempt > 0: - # Proportional scaling: halve the longer side and scale the - # shorter side to preserve aspect ratio (min dimension 64). - scale = 0.5 - new_w = max(int(img.width * scale), 64) - new_h = max(int(img.height * scale), 64) - # Re-derive the scale from whichever dimension hit the floor - # so both axes shrink by the same factor. - if new_w == 64 and img.width > 0: - effective_scale = 64 / img.width - new_h = max(int(img.height * effective_scale), 64) - elif new_h == 64 and img.height > 0: - effective_scale = 64 / img.height - new_w = max(int(img.width * effective_scale), 64) - # Stop if dimensions can't shrink further - if (new_w, new_h) == prev_dims: - break - img = img.resize((new_w, new_h), Image.LANCZOS) - prev_dims = (new_w, new_h) - logger.info("Resized to %dx%d (attempt %d)", new_w, new_h, attempt) - - for q in quality_steps: - buf = _io.BytesIO() - save_kwargs = {"format": pil_format} - if q is not None: - save_kwargs["quality"] = q - img.save(buf, **save_kwargs) - encoded = base64.b64encode(buf.getvalue()).decode("ascii") - candidate = f"data:{out_mime};base64,{encoded}" - if len(candidate) <= max_base64_bytes: - logger.info("Auto-resized image fits: %.1f MB (quality=%s, %dx%d)", - len(candidate) / (1024 * 1024), q, - img.width, img.height) - return candidate - - # If we still can't get it small enough, return the best attempt - # and let the caller decide - if candidate is not None: - logger.warning("Auto-resize could not fit image under %.1f MB (best: %.1f MB)", - max_base64_bytes / (1024 * 1024), len(candidate) / (1024 * 1024)) - return candidate - - # Shouldn't reach here, but fall back to full encode - return data_url or _image_to_base64_data_url(image_path, mime_type=mime_type) + try: + # Convert RGBA to RGB for JPEG output + if pil_format == "JPEG" and img.mode in {"RGBA", "P"}: + img = img.convert("RGB") + + # Strategy: halve dimensions until base64 fits, up to 4 rounds. + # For JPEG, also try reducing quality at each size step. + # For PNG, quality is irrelevant — only dimension reduction helps. + quality_steps = (85, 70, 50) if pil_format == "JPEG" else (None,) + prev_dims = (img.width, img.height) + candidate = None # will be set on first loop iteration + + for attempt in range(5): + if attempt > 0: + # Proportional scaling: halve the longer side and scale the + # shorter side to preserve aspect ratio (min dimension 64). + scale = 0.5 + new_w = max(int(img.width * scale), 64) + new_h = max(int(img.height * scale), 64) + # Re-derive the scale from whichever dimension hit the floor + # so both axes shrink by the same factor. + if new_w == 64 and img.width > 0: + effective_scale = 64 / img.width + new_h = max(int(img.height * effective_scale), 64) + elif new_h == 64 and img.height > 0: + effective_scale = 64 / img.height + new_w = max(int(img.width * effective_scale), 64) + # Stop if dimensions can't shrink further + if (new_w, new_h) == prev_dims: + break + img = img.resize((new_w, new_h), Image.LANCZOS) + prev_dims = (new_w, new_h) + logger.info("Resized to %dx%d (attempt %d)", new_w, new_h, attempt) + + for q in quality_steps: + buf = _io.BytesIO() + save_kwargs = {"format": pil_format} + if q is not None: + save_kwargs["quality"] = q + img.save(buf, **save_kwargs) + encoded = base64.b64encode(buf.getvalue()).decode("ascii") + candidate = f"data:{out_mime};base64,{encoded}" + if len(candidate) <= max_base64_bytes: + logger.info("Auto-resized image fits: %.1f MB (quality=%s, %dx%d)", + len(candidate) / (1024 * 1024), q, + img.width, img.height) + return candidate + + # If we still can't get it small enough, return the best attempt + # and let the caller decide + if candidate is not None: + logger.warning("Auto-resize could not fit image under %.1f MB (best: %.1f MB)", + max_base64_bytes / (1024 * 1024), len(candidate) / (1024 * 1024)) + return candidate + + # Shouldn't reach here, but fall back to full encode + return data_url or _image_to_base64_data_url(image_path, mime_type=mime_type) + finally: + img.close() # ---------------------------------------------------------------------------