Method _local_run_subprocess

ms_agent/skill/container.py:548–591

Run subprocess locally with security restrictions. Cross-platform support with timeout and resource limits. Args: cmd: Command list to execute. env: Environment variables. cwd: Working directory. stdin_input: Input to pass to

(self,
                              cmd: List[str],
                              env: Dict[str, str] = None,
                              cwd: Path = None,
                              stdin_input: str = None)

    # -------------------------------------------------------------------------

    def _local_run_subprocess(self,
                              cmd: List[str],
                              env: Dict[str, str] = None,
                              cwd: Path = None,
                              stdin_input: str = None) -> tuple[str, str, int]:
        """
        Run subprocess locally with security restrictions.

        Cross-platform support with timeout and resource limits.

        Args:
            cmd: Command list to execute.
            env: Environment variables.
            cwd: Working directory.
            stdin_input: Input to pass to stdin.

        Returns:
            Tuple of (stdout, stderr, exit_code).
        """
        # Setup environment
        run_env = os.environ.copy()
        run_env['SKILL_OUTPUT_DIR'] = str(self.output_dir)
        if env:
            run_env.update(env)

        # Use workspace as default cwd
        work_dir = cwd or self.workspace_dir

        try:
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=self.timeout,
                cwd=str(work_dir),
                env=run_env,
                stdin=subprocess.PIPE if stdin_input else None,
                input=stdin_input,
            )
            return result.stdout, result.stderr, result.returncode
        except subprocess.TimeoutExpired:
            return '', f'Execution timed out after {self.timeout}s', -1
        except Exception as e:
            return '', str(e), -1

    def _get_python_executable(self) -> str:
        """Get the Python executable for the current platform."""