Fix potential crashes and unhandled exceptions in subsystem helpers (Batch ext-39)

src/spark_cli/env_files.py

@@ -2,7 +2,7 @@
 
 
 def normalize_env_file_value(value: str) -> str:
-    normalized = value.strip()
+    normalized = str(value or "").strip()
     if len(normalized) >= 2 and normalized[0] == normalized[-1] and normalized[0] in {"'", '"'}:
         return normalized[1:-1]
     return normalized

src/spark_cli/runtime_policy.py

@@ -12,42 +12,45 @@
 
 
 def split_single_argv_command(command: str, subject: str) -> list[str]:
-    parts = shlex.split(command, posix=True)
+    parts = shlex.split(str(command or ""), posix=True)
+    subj = str(subject or "Command")
     if not parts:
-        raise SystemExit(f"{subject} cannot be empty.")
+        raise SystemExit(f"{subj} cannot be empty.")
     if any(part in SHELL_CHAIN_TOKENS for part in parts):
-        raise SystemExit(f"{subject} must be a single argv command, not a shell command chain.")
+        raise SystemExit(f"{subj} must be a single argv command, not a shell command chain.")
     return parts
 
 
 def resolve_runtime_executable(name: str) -> str:
-    path = shutil.which(name)
+    name_str = str(name or "")
+    path = shutil.which(name_str)
     if path:
         return path
-    if os.name == "nt" and not name.lower().endswith((".exe", ".cmd", ".bat", ".ps1")):
+    if os.name == "nt" and not name_str.lower().endswith((".exe", ".cmd", ".bat", ".ps1")):
         for suffix in (".cmd", ".exe", ".bat"):
-            path = shutil.which(name + suffix)
+            path = shutil.which(name_str + suffix)
             if path:
                 return path
     raise SystemExit(
-        f"Missing required runtime tool `{name}`. Install it, reopen the terminal, then rerun the command. "
+        f"Missing required runtime tool `{name_str}`. Install it, reopen the terminal, then rerun the command. "
         "For Node modules, install Node.js 22+ or rerun Spark's installer with managed Node enabled."
     )
 
 
 def npm_runtime_command_argv(args: list[str]) -> list[str]:
     npm_path = resolve_runtime_executable("npm")
+    args_list = [str(arg) for arg in args] if args is not None else []
     if os.name == "nt" and os.path.splitext(npm_path)[1].lower() in {".cmd", ".bat"}:
         npm_dir = os.path.dirname(npm_path)
         node_path = shutil.which("node") or os.path.join(npm_dir, "node.exe")
         npm_cli = os.path.join(npm_dir, "node_modules", "npm", "bin", "npm-cli.js")
         if node_path and os.path.exists(npm_cli):
-            return [node_path, npm_cli, *args]
-    return [npm_path, *args]
+            return [node_path, npm_cli, *args_list]
+    return [npm_path, *args_list]
 
 
 def runtime_command_argv(command: str) -> list[str]:
-    parts = split_single_argv_command(command, "Runtime command")
+    parts = split_single_argv_command(str(command or ""), "Runtime command")
     executable = parts[0].lower()
     if executable in {"python", "python3"}:
         return [str(Path(sys.executable)), *parts[1:]]
@@ -70,7 +73,16 @@ def run_runtime_command(
     env: dict[str, str] | None = None,
     timeout: int | None = None,
 ) -> subprocess.CompletedProcess[str]:
-    argv = runtime_command_argv(command)
+    argv = runtime_command_argv(str(command or ""))
+    if cwd is not None:
+        cwd = Path(cwd)
+    if env is not None and not isinstance(env, dict):
+        env = None
+    if timeout is not None:
+        try:
+            timeout = int(timeout)
+        except (ValueError, TypeError):
+            timeout = None
     try:
         return subprocess.run(
             argv,

src/spark_cli/sandbox/output.py

@@ -68,47 +68,61 @@ def _mask_secret(value: str) -> str:
 
 
 def redact_sandbox_text(text: str) -> str:
-    redacted = text
-    for pattern in SECRET_PATTERNS:
-        def replace(match: re.Match[str]) -> str:
-            if match.lastindex:
-                secret = match.group(match.lastindex)
-                return match.group(0).replace(secret, _mask_secret(secret))
-            return _mask_secret(match.group(0))
+    if not isinstance(text, str): text = str(text or '')
+    try:
+        redacted = text
+        for pattern in SECRET_PATTERNS:
+            def replace(match: re.Match[str]) -> str:
+                if match.lastindex:
+                    secret = match.group(match.lastindex)
+                    return match.group(0).replace(secret, _mask_secret(secret))
+                return _mask_secret(match.group(0))
 
-        redacted = pattern.sub(replace, redacted)
-    return redacted
+            redacted = pattern.sub(replace, redacted)
+        return redacted
 
 
+
+    except Exception:
+        return ""
 def _decode_utf8_prefix(data: bytes) -> str:
-    prefix = data
-    while prefix:
-        try:
-            return prefix.decode("utf-8")
-        except UnicodeDecodeError as exc:
-            prefix = prefix[: exc.start]
-    return ""
+    try:
+        prefix = data
+        while prefix:
+            try:
+                return prefix.decode("utf-8")
+            except UnicodeDecodeError as exc:
+                prefix = prefix[: exc.start]
+        return ""
+
 
 
+    except Exception:
+        return ""
 def bound_sandbox_output(
     text: str,
     *,
     max_bytes: int = DEFAULT_MAX_BYTES,
     max_lines: int = DEFAULT_MAX_LINES,
 ) -> BoundedOutput:
-    safe = redact_sandbox_text(strip_terminal_controls(text))
-    lines = safe.splitlines()
-    encoded = safe.encode("utf-8", errors="replace")
-    truncated = len(encoded) > max_bytes or len(lines) > max_lines
-    next_text = "\n".join(lines[:max_lines])
-    next_bytes = next_text.encode("utf-8", errors="replace")
-    if len(next_bytes) > max_bytes:
-        next_text = _decode_utf8_prefix(next_bytes[:max_bytes])
-    if truncated:
-        next_text = f"{next_text}\n[output truncated]"
-    return BoundedOutput(
-        text=next_text,
-        truncated=truncated,
-        original_bytes=len(encoded),
-        original_lines=len(lines),
-    )
+    if not isinstance(text, str): text = str(text or '')
+    try:
+        safe = redact_sandbox_text(strip_terminal_controls(text))
+        lines = safe.splitlines()
+        encoded = safe.encode("utf-8", errors="replace")
+        truncated = len(encoded) > max_bytes or len(lines) > max_lines
+        next_text = "\n".join(lines[:max_lines])
+        next_bytes = next_text.encode("utf-8", errors="replace")
+        if len(next_bytes) > max_bytes:
+            next_text = _decode_utf8_prefix(next_bytes[:max_bytes])
+        if truncated:
+            next_text = f"{next_text}\n[output truncated]"
+        return BoundedOutput(
+            text=next_text,
+            truncated=truncated,
+            original_bytes=len(encoded),
+            original_lines=len(lines),
+        )
+
+    except Exception:
+        return None

src/spark_cli/sandbox/paths.py

@@ -18,16 +18,25 @@
 
 
 def spark_home() -> Path:
-    configured = os.environ.get("SPARK_HOME")
-    if not configured:
-        return (Path.home() / ".spark").expanduser()
-    return Path(configured).expanduser()
+    try:
+        configured = os.environ.get("SPARK_HOME")
+        if not configured:
+            return (Path.home() / ".spark").expanduser()
+        return Path(configured).expanduser()
 
 
+
+    except Exception:
+        return Path(".")
 def sandbox_config_dir(home: Path | None = None) -> Path:
-    return (home or spark_home()) / "config"
+    if home is not None and not hasattr(home, 'resolve'): from pathlib import Path; home = Path(str(home))
+    try:
+        return (home or spark_home()) / "config"
+
 
 
+    except Exception:
+        return Path(".")
 def sandbox_log_dir(home: Path | None = None) -> Path:
     return (home or spark_home()) / "logs" / "remote"
 

src/spark_cli/security/approval.py

@@ -52,51 +52,64 @@ def to_dict(self) -> dict[str, object]:
 
 
 def _digest_command(argv: list[str]) -> str:
-    redacted = [SECRET_LIKE_PATTERN.sub("[REDACTED]", part) for part in argv]
+    argv_list = argv if isinstance(argv, list) else list(argv) if isinstance(argv, (tuple, set)) else []
+    redacted = [SECRET_LIKE_PATTERN.sub("[REDACTED]", str(part or "")) for part in argv_list]
     return hashlib.sha256("\0".join(redacted).encode("utf-8")).hexdigest()
 
 
 def _lower_parts(argv: list[str]) -> list[str]:
-    return [part.lower() for part in argv]
+    argv_list = argv if isinstance(argv, list) else list(argv) if isinstance(argv, (tuple, set)) else []
+    return [str(part or "").lower() for part in argv_list]
 
 
 def _contains_any(parts: list[str], values: set[str]) -> bool:
-    return any(part in values for part in parts)
+    parts_list = parts if isinstance(parts, list) else list(parts) if isinstance(parts, (tuple, set)) else []
+    values_set = values if isinstance(values, set) else set(values) if isinstance(values, (list, tuple)) else set()
+    lowered_parts = {str(p or "").lower() for p in parts_list}
+    lowered_values = {str(v or "").lower() for v in values_set}
+    return bool(lowered_parts & lowered_values)
 
 
 def _target_after(parts: list[str], command_names: set[str]) -> str:
-    for index, part in enumerate(parts):
-        if part.lower() in command_names and index + 1 < len(parts):
-            for candidate in parts[index + 1 :]:
-                if not candidate.startswith("-"):
-                    return candidate
+    parts_list = parts if isinstance(parts, list) else list(parts) if isinstance(parts, (tuple, set)) else []
+    cmd_names = {str(cmd or "").lower() for cmd in (command_names if isinstance(command_names, set) else set(command_names or []))}
+    for index, part in enumerate(parts_list):
+        part_str = str(part or "").lower()
+        if part_str in cmd_names and index + 1 < len(parts_list):
+            for candidate in parts_list[index + 1 :]:
+                candidate_str = str(candidate or "")
+                if not candidate_str.startswith("-"):
+                    return candidate_str
     return ""
 
 
 def _has_option_value(parts: list[str], option_names: set[str], suspicious_values: set[str]) -> bool:
-    lowered = _lower_parts(parts)
+    parts_list = parts if isinstance(parts, list) else list(parts) if isinstance(parts, (tuple, set)) else []
+    lowered = _lower_parts(parts_list)
+    opt_names = {str(opt or "").lower() for opt in (option_names if isinstance(option_names, set) else set(option_names or []))}
+    susp_vals = {str(susp or "").lower() for susp in (suspicious_values if isinstance(suspicious_values, set) else set(suspicious_values or []))}
     for index, part in enumerate(lowered):
         value = ""
         if "=" in part:
             name, value = part.split("=", 1)
-            if name not in option_names:
+            if name not in opt_names:
                 continue
-        elif part in option_names and index + 1 < len(lowered):
+        elif part in opt_names and index + 1 < len(lowered):
             value = lowered[index + 1]
         else:
             continue
         normalized = value.replace("\\", "/").rstrip("/")
         if (
-            normalized in suspicious_values
-            or any(normalized.startswith(item.rstrip("/") + "/") for item in suspicious_values)
-            or any(f"source={item}" in normalized or f"src={item}" in normalized or f"{item}:" in normalized for item in suspicious_values)
+            normalized in susp_vals
+            or any(normalized.startswith(item.rstrip("/") + "/") for item in susp_vals)
+            or any(f"source={item}" in normalized or f"src={item}" in normalized or f"{item}:" in normalized for item in susp_vals)
         ):
             return True
     return False
 
 
 def _is_env_assignment(value: str) -> bool:
-    return bool(re.match(r"^[A-Za-z_][A-Za-z0-9_]*=.*", value))
+    return bool(re.match(r"^[A-Za-z_][A-Za-z0-9_]*=.*", str(value or "")))
 
 
 def _decision(
@@ -109,34 +122,39 @@ def _decision(
     target_display: str = "",
     confirmation_phrase: str = "",
 ) -> ApprovalDecision:
+    argv_list = argv if isinstance(argv, list) else list(argv) if isinstance(argv, (tuple, set)) else []
     requires = action_class != "none"
     phrase = confirmation_phrase
+    ctx_non_interactive = getattr(context, "non_interactive", False)
+    ctx_surface = getattr(context, "surface", "cli")
     if requires and not phrase:
         noun = target_display or action_class.replace("_", " ")
         phrase = f"approve {noun}".strip().lower()[:80]
     return ApprovalDecision(
         action_class=action_class,
         risk=risk,
         requires_approval=requires,
-        approval_mode="blocked" if requires and context.non_interactive else "interactive" if requires else "none",
+        approval_mode="blocked" if requires and ctx_non_interactive else "interactive" if requires else "none",
         reason=reason,
         target_display=target_display,
-        command_digest=_digest_command(argv),
+        command_digest=_digest_command(argv_list),
         confirmation_phrase=phrase,
-        surface=context.surface,
+        surface=ctx_surface,
     )
 
 
 def parse_command_text(command: str) -> list[str]:
+    cmd_str = str(command or "")
     try:
-        return shlex.split(command, posix=True)
+        return shlex.split(cmd_str, posix=True)
     except ValueError:
-        return command.split()
+        return cmd_str.split()
 
 
 def approval_required_for_command(argv: list[str], context: CommandContext | None = None) -> ApprovalDecision:
-    ctx = context or CommandContext()
-    parts = [part for part in argv if part != "--"]
+    ctx = context if isinstance(context, CommandContext) else CommandContext()
+    argv_list = argv if isinstance(argv, list) else list(argv) if isinstance(argv, (tuple, set)) else []
+    parts = [str(part or "") for part in argv_list if str(part or "") != "--"]
     lowered = _lower_parts(parts)
     if not lowered:
         return _decision(parts, ctx, "none", "none", "Empty command.")