fix(pr398): backend service fixes from PR #396 review

pmoves/services/agent-zero/main.py

@@ -627,7 +627,7 @@ async def lifespan(app: FastAPI):
     for sig in (signal.SIGTERM, signal.SIGINT):
         try:
             loop.add_signal_handler(
-                sig, lambda s=sig: asyncio.create_task(process_manager.stop())
+                sig, lambda: asyncio.create_task(process_manager.stop())
             )
         except (NotImplementedError, RuntimeError, ValueError):
             # Tests run event loops in worker threads where signal handlers are unsupported.
@@ -643,8 +643,10 @@ async def lifespan(app: FastAPI):
             await _controller_task
         _controller_task = None
     if event_controller.is_started:
-        with contextlib.suppress(Exception):
+        try:
             await event_controller.stop()
+        except Exception:
+            logger.debug("Exception during event controller shutdown", exc_info=True)
 
 
 app = FastAPI(title="Agent Zero Supervisor", lifespan=lifespan)

pmoves/services/agent_zero/controller.py

@@ -671,8 +671,9 @@ async def _fallback_to_core_nats(self) -> None:
         for sub in self._subscriptions:
             try:
                 await sub.unsubscribe()
-            except Exception:
-                pass
+            except Exception as exc:
+                sub_info = getattr(sub, "subject", sub)
+                logger.warning("Failed to unsubscribe from %s: %s", sub_info, exc)
         self._subscriptions.clear()
         # Disable JS for this controller and create core subscriptions
         self.settings.use_jetstream = False

pmoves/services/comfy-watcher/watcher.py

@@ -1,44 +1,120 @@
 import asyncio
 import datetime
+from datetime import timedelta
 import hashlib
 import json
+import logging
 import os
 import pathlib
+import shutil
 import time
 
 from minio import Minio
 from nats.aio.client import Client as NATS
 
 from services.common.events import envelope
 
+logger = logging.getLogger("comfy-watcher")
+
+
+def _parse_int_env(key: str, default: int) -> int:
+    """Parse integer from environment with validation.
+
+    Args:
+        key: Environment variable name.
+        default: Default value if missing or invalid.
+
+    Returns:
+        Parsed integer value or default.
+    """
+    value = os.environ.get(key)
+    if value is None:
+        return default
+    try:
+        return int(value)
+    except ValueError:
+        logger.warning(f"Invalid {key}={value!r}, using default={default}")
+        return default
+
 MINIO_ENDPOINT = os.environ.get("MINIO_ENDPOINT", "minio:9000")
 MINIO_USE_SSL = os.environ.get("MINIO_USE_SSL", "false").lower() == "true"
 MINIO_ACCESS_KEY = os.environ.get("MINIO_ACCESS_KEY")
 MINIO_SECRET_KEY = os.environ.get("MINIO_SECRET_KEY")
 BUCKET = os.environ.get("MINIO_BUCKET", "pmoves-comfyui")
 PUBLIC_BASE_URL = os.environ.get("PUBLIC_BASE_URL", "http://minio:9000")
-PRESIGN_HOURS = int(os.environ.get("PRESIGN_EXPIRES_HOURS", "24"))
+PRESIGN_HOURS = _parse_int_env("PRESIGN_EXPIRES_HOURS", 24)
 OUTPUT_DIR = os.environ.get("COMFY_OUTPUT_DIR", "/data/output")
 STATE_PATH = os.environ.get("COMFY_WATCHER_STATE_PATH", "/state/state.json")
 NATS_URL = os.environ.get("NATS_URL", "nats://nats:4222")
-POLL_SECONDS = int(os.environ.get("COMFY_WATCHER_POLL_SECONDS", "5"))
+POLL_SECONDS = _parse_int_env("COMFY_WATCHER_POLL_SECONDS", 5)
 
 
 def load_state() -> dict:
+    """Load state from disk with proper error handling.
+
+    Returns the uploaded file registry. Handles missing state files,
+    corrupted JSON, and other I/O errors gracefully with logging.
+    """
     try:
         with open(STATE_PATH) as f:
             return json.load(f)
-    except Exception:
+    except FileNotFoundError:
+        logger.info(f"State file not found, starting fresh: {STATE_PATH}")
+        return {"uploaded": {}}
+    except json.JSONDecodeError as exc:
+        logger.error(f"Corrupted state file at {STATE_PATH}, backing up and starting fresh: {exc}")
+        # Backup corrupted file for investigation
+        backup_path = f"{STATE_PATH}.corrupted.{int(time.time())}"
+        try:
+            shutil.copy(STATE_PATH, backup_path)
+            logger.info(f"Backed up corrupted state to: {backup_path}")
+        except Exception as backup_exc:
+            logger.warning(f"Could not backup corrupted state file: {backup_exc}")
+        return {"uploaded": {}}
+    except Exception as exc:
+        logger.error(f"Unexpected error loading state from {STATE_PATH}: {exc}")
         return {"uploaded": {}}
 
 
 def save_state(state: dict) -> None:
+    """Save the watcher state to disk.
+
+    Persists the current state dictionary to JSON storage, ensuring the
+    target directory exists. Creates parent directories as needed.
+
+    Args:
+        state: A dictionary containing the watcher state, typically with
+            an "uploaded" key mapping file hashes to their metadata
+            (key, timestamp, size).
+
+    Raises:
+        OSError: If the state file cannot be written due to permission
+            or disk space issues.
+        TypeError: If the state dictionary contains non-serializable objects.
+    """
     pathlib.Path(os.path.dirname(STATE_PATH)).mkdir(parents=True, exist_ok=True)
     with open(STATE_PATH, "w") as f:
         json.dump(state, f)
 
 
 def file_hash(path: str) -> str:
+    """Calculate the SHA-256 hash of a file.
+
+    Computes a cryptographic hash by reading the file in 1 MB chunks,
+    allowing efficient hashing of large files without loading them
+    entirely into memory.
+
+    Args:
+        path: The absolute or relative path to the file to hash.
+
+    Returns:
+        A hexadecimal string representing the SHA-256 hash of the
+        file's contents.
+
+    Raises:
+        FileNotFoundError: If the specified file does not exist.
+        PermissionError: If the file cannot be read due to permissions.
+    """
     h = hashlib.sha256()
     with open(path, "rb") as f:
         while True:
@@ -50,56 +126,93 @@ def file_hash(path: str) -> str:
 
 
 async def run() -> None:
+    """Main watcher loop for ComfyUI output files.
+
+    Continuously monitors the configured output directory for new files,
+    uploads them to MinIO object storage, and publishes NATS events
+    with artifact metadata and presigned URLs.
+
+    The function:
+    - Validates MinIO credentials are configured
+    - Loads the upload state to avoid re-uploading files
+    - Ensures the target MinIO bucket exists
+    - Polls the output directory at configured intervals
+    - Uploads new files with timestamped S3-style keys
+    - Generates presigned URLs for temporary access
+    - Publishes completion events to NATS for downstream processing
+
+    Raises:
+        RuntimeError: If MinIO credentials are not set or bucket
+            creation fails.
+
+    The loop runs indefinitely until the process is terminated.
+    """
     if not MINIO_ACCESS_KEY or not MINIO_SECRET_KEY:
         raise RuntimeError("MINIO_ACCESS_KEY and MINIO_SECRET_KEY must be set")
 
     state = load_state()
-    client = Minio(MINIO_ENDPOINT, access_key=MINIO_ACCESS_KEY, secret_key=MINIO_SECRET_KEY, secure=MINIO_USE_SSL)
-
+    client = Minio(
+        MINIO_ENDPOINT,
+        access_key=MINIO_ACCESS_KEY,
+        secret_key=MINIO_SECRET_KEY,
+        secure=MINIO_USE_SSL,
+    )
     nc = NATS()
     await nc.connect(servers=[NATS_URL])
 
+    # Ensure MinIO bucket exists with proper error handling
     try:
         if not client.bucket_exists(BUCKET):
             client.make_bucket(BUCKET)
-    except Exception as e:
-        print("Bucket check/make:", e)
+            logger.info(f"Created MinIO bucket: {BUCKET}")
+    except Exception as exc:
+        logger.error(f"Failed to verify/create MinIO bucket '{BUCKET}': {exc}")
+        raise RuntimeError(f"MinIO bucket setup failed for '{BUCKET}': {exc}") from exc
 
     while True:
         for root, _, files in os.walk(OUTPUT_DIR):
-            for fn in files:
-                path = os.path.join(root, fn)
-                try:
-                    size = os.path.getsize(path)
-                    if size == 0:
-                        continue
+            for fname in files:
+                if not fname.endswith((".png", ".jpg", ".jpeg", ".webp", ".gif")):
+                    continue
+                path = os.path.join(root, fname)
+                h = file_hash(path)
+
+                if h in state.get("uploaded", {}):
+                    continue
 
-                    h = file_hash(path)
-                    if state["uploaded"].get(h):
-                        continue
+                size_bytes = os.path.getsize(path)
+                timestamp = int(time.time())
+                key = f"comfyui/{timestamp}/{fname}"
 
-                    key = datetime.datetime.now(datetime.timezone.utc).strftime("comfyui/%Y/%m/%d/") + fn
-                    client.fput_object(BUCKET, key, path, content_type="image/png")
-                    state["uploaded"][h] = {"key": key, "ts": time.time(), "size": size}
+                try:
+                    with open(path, "rb") as f:
+                        client.put_object(BUCKET, key, f, length=size_bytes)
+                    state.setdefault("uploaded", {})[h] = {
+                        "key": key,
+                        "timestamp": timestamp,
+                        "size": size_bytes,
+                    }
                     save_state(state)
 
                     public_url = (
                         f"{PUBLIC_BASE_URL}/{BUCKET}/{key}".replace("//", "/").replace("http:/", "http://").replace("https:/", "https://")
                     )
 
                     payload = {"artifact_uri": f"s3://{BUCKET}/{key}", "meta": {"public_url": public_url}}
-                    try:
-                        from datetime import timedelta
 
+                    # Generate presigned URL with error handling
+                    try:
                         payload["meta"]["presigned_url"] = client.presigned_get_object(BUCKET, key, expires=timedelta(hours=PRESIGN_HOURS))
-                    except Exception:
-                        pass
+                    except Exception as presign_exc:
+                        logger.warning(f"Failed to generate presigned URL for {key}: {presign_exc}")
+                        # Continue without presigned URL - not critical for all workflows
 
                     env = envelope("gen.image.result.v1", payload, source="comfy-watcher")
                     await nc.publish("gen.image.result.v1", json.dumps(env).encode())
-                    print("Uploaded and announced:", key)
-                except Exception as e:
-                    print("Error processing", path, e)
+                    logger.info(f"Uploaded and announced: {key}")
+                except Exception as exc:
+                    logger.error(f"Error processing file {path}: {exc}", exc_info=True)
+                    # Continue to next file instead of aborting entire loop
 
         await asyncio.sleep(POLL_SECONDS)
 

pmoves/services/gateway-agent/app.py

@@ -113,8 +113,9 @@ async def lifespan(app: FastAPI):
 app = FastAPI(
     title="PMOVES Gateway Agent",
     description="Orchestrates 100+ MCP tools with Cipher memory integration",
-    version="1.0.0"
-, lifespan=lifespan)
+    version="1.0.0",
+    lifespan=lifespan,
+)
 
 
 # ============================================================================

pmoves/services/hi-rag-gateway-v2/app.py

@@ -30,6 +30,11 @@
 _RERANK_CONFIG_ERRORS: List[str] = []
 _RERANK_CONFIG_WARNINGS: List[str] = []
 
+# Configure logging early for use in helper functions
+logging.basicConfig(level=logging.INFO)
+logging.getLogger().setLevel(logging.INFO)
+logger = logging.getLogger("hirag.gateway.v2")
+
 
 def _parse_bool(name: str, default: bool) -> bool:
     raw = os.environ.get(name)
@@ -61,6 +66,17 @@ def _parse_positive_int(name: str, default: int) -> int:
     return value
 
 
+def _parse_int_env(name: str, default: int) -> int:
+    """Parse integer environment variable with validation and fallback."""
+    raw = os.environ.get(name, str(default))
+    try:
+        return int(raw)
+    except (TypeError, ValueError) as exc:
+        # Use logger for general config warnings, not rerank-specific list
+        logger.warning("%s value %r invalid, using default %s: %s", name, raw, default, exc)
+        return default
+
+
 def _parse_optional_bool(name: str) -> Optional[bool]:
     raw = os.environ.get(name)
     if raw is None or str(raw).strip() == "":
@@ -167,10 +183,10 @@ def get_rerank_status() -> Dict[str, Any]:
 NEO4J_URL = (os.environ.get("NEO4J_URL","bolt://neo4j:7687") or "").strip()
 NEO4J_USER = os.environ.get("NEO4J_USER","neo4j")
 NEO4J_PASSWORD = os.environ.get("NEO4J_PASSWORD","neo4j")
-NEO4J_DICT_REFRESH_SEC = int(os.environ.get("NEO4J_DICT_REFRESH_SEC","60"))
-NEO4J_DICT_LIMIT = int(os.environ.get("NEO4J_DICT_LIMIT","50000"))
-ENTITY_CACHE_TTL = int(os.environ.get("ENTITY_CACHE_TTL","60"))
-ENTITY_CACHE_MAX = int(os.environ.get("ENTITY_CACHE_MAX","1000"))
+NEO4J_DICT_REFRESH_SEC = _parse_int_env("NEO4J_DICT_REFRESH_SEC", 60)
+NEO4J_DICT_LIMIT = _parse_int_env("NEO4J_DICT_LIMIT", 50000)
+ENTITY_CACHE_TTL = _parse_int_env("ENTITY_CACHE_TTL", 60)
+ENTITY_CACHE_MAX = _parse_int_env("ENTITY_CACHE_MAX", 1000)
 
 SUPABASE_REST_URL = os.environ.get("SUPA_REST_URL") or os.environ.get("SUPABASE_REST_URL")
 # Optional internal REST base explicitly pointing at host-gateway; prefer for derivations
@@ -203,7 +219,7 @@ def _is_jwt(key: str) -> bool:
     SUPABASE_ANON_KEY if _is_jwt(SUPABASE_ANON_KEY) else
     _jwt_service_key or _explicit_realtime_key or SUPABASE_ANON_KEY
 )
-GEOMETRY_CACHE_WARM_LIMIT = int(os.environ.get("GEOMETRY_CACHE_WARM_LIMIT", "64"))
+GEOMETRY_CACHE_WARM_LIMIT = _parse_int_env("GEOMETRY_CACHE_WARM_LIMIT", 64)
 # Realtime connection retry configuration
 GEOMETRY_REALTIME_BACKOFF = float(os.environ.get("GEOMETRY_REALTIME_BACKOFF", "5.0"))  # Initial retry delay (seconds)
 GEOMETRY_REALTIME_MAX_BACKOFF = float(os.environ.get("GEOMETRY_REALTIME_MAX_BACKOFF", "60.0"))  # Max retry delay cap (seconds)
@@ -214,13 +230,9 @@ def _is_jwt(key: str) -> bool:
 TAILSCALE_CIDRS = [c.strip() for c in os.environ.get("TAILSCALE_CIDRS","100.64.0.0/10").split(",") if c.strip()]
 TRUSTED_PROXY_SOURCES = [c.strip() for c in os.environ.get("HIRAG_TRUSTED_PROXIES", "").split(",") if c.strip()]
 
-HTTP_PORT = int(os.environ.get("HIRAG_HTTP_PORT","8086"))
+HTTP_PORT = _parse_int_env("HIRAG_HTTP_PORT", 8086)
 NAMESPACE_DEFAULT = os.environ.get("INDEXER_NAMESPACE","pmoves")
 
-logging.basicConfig(level=logging.INFO)
-logging.getLogger().setLevel(logging.INFO)
-logger = logging.getLogger("hirag.gateway.v2")
-
 if RERANK_ENABLE:
     logger.info(
         "Rerank enabled with model=%s provider=%s topn=%s k=%s fusion=%s device=%s",
@@ -1006,7 +1018,7 @@ async def _geometry_realtime_worker(ws_url: str, api_key: str) -> None:
 GAN_SIDECAR_ENABLED = os.environ.get("GAN_SIDECAR_ENABLED", "false").lower()=="true"
 
 PGHOST = os.environ.get("PGHOST")
-PGPORT = int(os.environ.get("PGPORT","5432"))
+PGPORT = _parse_int_env("PGPORT", 5432)
 PGUSER = os.environ.get("PGUSER")
 PGPASSWORD = os.environ.get("PGPASSWORD")
 PGDATABASE = os.environ.get("PGDATABASE")

pmoves/services/pdf-ingest/app.py

@@ -127,7 +127,6 @@ async def lifespan(app: FastAPI):
 NATS_URL = os.environ.get("NATS_URL", "nats://nats:4222")
 
 _nc: Optional[NATS] = None
-_nats_supervisor: Optional[asyncio.Task[None]] = None
 
 
 async def _nats_connect_loop() -> None: