ComfyUI-OpenClaw/services/transform_runner.py

"""
S35 Transform Isolation Runner.

Executes transforms in a separate process via `services.transform_worker`.
"""

import json
import logging
import os
import subprocess
import sys
import time
from typing import Any, Dict, Optional

from .transform_common import (
    TransformLimits,
    TransformRegistry,
    TransformResult,
    TransformStatus,
)

logger = logging.getLogger("ComfyUI-OpenClaw.services.transform_runner")


class TransformProcessRunner:
    """
    Executes transforms in an isolated subprocess.
    """

    def __init__(
        self,
        registry: TransformRegistry,
        limits: Optional[TransformLimits] = None,
    ):
        self._registry = registry
        self._limits = limits or TransformLimits.from_env()

    def execute_transform(
        self,
        transform_id: str,
        input_data: Dict[str, Any],
        *,
        trace_id: str = "",
    ) -> TransformResult:
        """
        Execute a single registered transform in a subprocess.
        """
        transform = self._registry.get_transform(transform_id)
        if not transform:
            return TransformResult(
                transform_id=transform_id,
                status=TransformStatus.ERROR.value,
                error=f"Transform '{transform_id}' not found in registry",
            )

        # Integrity Check (R77 pre-check)
        if not self._registry.verify_integrity(transform_id):
            return TransformResult(
                transform_id=transform_id,
                status=TransformStatus.DENIED.value,
                error="Integrity verification failed — module modified",
                audit={"reason": "integrity_check_failed", "trace_id": trace_id},
            )

        start_time = time.monotonic()

        # Prepare Worker Command
        worker_script = os.path.join(os.path.dirname(__file__), "transform_worker.py")
        cmd = [sys.executable, worker_script, transform.module_path]

        # Prepare Input
        payload = {"input": input_data, "context": {"trace_id": trace_id}}
        input_json = json.dumps(payload)

        try:
            # capability-deny: no environment variable inheritance by default?
            # Or minimal env.
            env = os.environ.copy()
            # Remove sensitive vars if needed?
            # S34 Obs: redact env in logs, but here process sees env.
            # Best practice: clear sensitive vars.
            for key in list(env.keys()):
                if "TOKEN" in key or "SECRET" in key or "KEY" in key:
                    del env[key]

            # Subprocess Run
            proc = subprocess.run(
                cmd,
                input=input_json,
                capture_output=True,
                text=True,
                timeout=self._limits.timeout_sec,
                env=env,
                check=False,  # We handle return codes
            )

            # Execution Time
            elapsed_ms = (time.monotonic() - start_time) * 1000

            # Handle Return Code
            if proc.returncode != 0:
                # Script crashed or printed error to stdout/stderr
                # Try to parse stdout error first
                error_msg = proc.stderr.strip() or "Process crashed with unknown error"
                try:
                    out_json = json.loads(proc.stdout)
                    if out_json.get("status") == "error":
                        error_msg = out_json.get("error", error_msg)
                except Exception:
                    pass

                return TransformResult(
                    transform_id=transform_id,
                    status=TransformStatus.ERROR.value,
                    error=f"Worker process failed (exit {proc.returncode}): {error_msg}",
                    duration_ms=elapsed_ms,
                    audit={"exit_code": proc.returncode, "trace_id": trace_id},
                )

            # Parse Output
            try:
                result_json = json.loads(proc.stdout)
            except json.JSONDecodeError:
                return TransformResult(
                    transform_id=transform_id,
                    status=TransformStatus.ERROR.value,
                    error="Worker returned invalid JSON output",
                    duration_ms=elapsed_ms,
                    audit={"raw_stdout": proc.stdout[:1000], "trace_id": trace_id},
                )

            if result_json.get("status") == "error":
                return TransformResult(
                    transform_id=transform_id,
                    status=TransformStatus.ERROR.value,
                    error=result_json.get("error", "Unknown worker error"),
                    duration_ms=elapsed_ms,
                    audit={
                        "trace_id": trace_id,
                        "traceback": result_json.get("traceback"),
                    },
                )

            output_data = result_json.get("output", {})
            output_bytes = len(json.dumps(output_data).encode("utf-8"))

            if output_bytes > self._limits.max_output_bytes:
                return TransformResult(
                    transform_id=transform_id,
                    status=TransformStatus.ERROR.value,
                    error=f"Output size limit exceeded ({output_bytes} > {self._limits.max_output_bytes})",
                    duration_ms=elapsed_ms,
                    output_bytes=output_bytes,
                    audit={"trace_id": trace_id},
                )

            return TransformResult(
                transform_id=transform_id,
                status=TransformStatus.SUCCESS.value,
                output=output_data,
                duration_ms=elapsed_ms,
                output_bytes=output_bytes,
                audit={"trace_id": trace_id, "isolation": "process"},
            )

        except subprocess.TimeoutExpired:
            elapsed_ms = (time.monotonic() - start_time) * 1000
            return TransformResult(
                transform_id=transform_id,
                status=TransformStatus.TIMEOUT.value,
                error=f"Transform timeout exceeded ({self._limits.timeout_sec}s)",
                duration_ms=elapsed_ms,
                audit={"trace_id": trace_id, "timeout": True},
            )
        except Exception as e:
            elapsed_ms = (time.monotonic() - start_time) * 1000
            return TransformResult(
                transform_id=transform_id,
                status=TransformStatus.ERROR.value,
                error=f"Runner exception: {str(e)}",
                duration_ms=elapsed_ms,
                audit={"trace_id": trace_id},
            )