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

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/modal.py

@@ -211,104 +211,118 @@ def _text(value):
 
 
 def modal_smoke_subprocess_env(env: dict[str, str] | None = None) -> dict[str, str]:
-    source = os.environ if env is None else env
-    return {
-        key: value
-        for key, value in source.items()
-        if key.upper() in MODAL_SMOKE_ENV_ALLOWLIST
-    }
+    if not isinstance(env, str): env = str(env or '')
+    try:
+        source = os.environ if env is None else env
+        return {
+            key: value
+            for key, value in source.items()
+            if key.upper() in MODAL_SMOKE_ENV_ALLOWLIST
+        }
+
 
 
+    except Exception:
+        return {}
 def run_modal_smoke_probe(*, timeout: int = MODAL_SMOKE_TIMEOUT_SECONDS) -> dict[str, object]:
-    if not modal_sdk_available():
-        return {
-            "ok": False,
-            "returncode": 127,
-            "output": bound_sandbox_output("").to_dict(),
-            "cleanup_requested": False,
-            "detail": "Modal Python SDK is not importable.",
-        }
     try:
-        result = subprocess.run(
-            [sys.executable, "-c", modal_smoke_script()],
-            capture_output=True,
-            env=modal_smoke_subprocess_env(),
-            text=True,
-            encoding="utf-8",
-            errors="replace",
-            timeout=timeout,
-        )
-    except subprocess.TimeoutExpired as error:
-        stdout = error.stdout if isinstance(error.stdout, str) else ""
-        stderr = error.stderr if isinstance(error.stderr, str) else ""
+        if not modal_sdk_available():
+            return {
+                "ok": False,
+                "returncode": 127,
+                "output": bound_sandbox_output("").to_dict(),
+                "cleanup_requested": False,
+                "detail": "Modal Python SDK is not importable.",
+            }
+        try:
+            result = subprocess.run(
+                [sys.executable, "-c", modal_smoke_script()],
+                capture_output=True,
+                env=modal_smoke_subprocess_env(),
+                text=True,
+                encoding="utf-8",
+                errors="replace",
+                timeout=timeout,
+            )
+        except subprocess.TimeoutExpired as error:
+            stdout = error.stdout if isinstance(error.stdout, str) else ""
+            stderr = error.stderr if isinstance(error.stderr, str) else ""
+            return {
+                "ok": False,
+                "returncode": 124,
+                "output": bound_sandbox_output((stdout or "") + ("\n" if stdout and stderr else "") + (stderr or "")).to_dict(),
+                "cleanup_requested": True,
+                "detail": f"Modal smoke timed out after {timeout}s.",
+            }
+        except OSError as error:
+            return {
+                "ok": False,
+                "returncode": 127,
+                "output": bound_sandbox_output("").to_dict(),
+                "cleanup_requested": False,
+                "detail": f"Could not start Modal smoke: {error.__class__.__name__}.",
+            }
+        output = bound_sandbox_output((result.stdout or "") + ("\n" if result.stdout and result.stderr else "") + (result.stderr or ""))
+        ok = result.returncode == 0 and "SPARK_MODAL_SMOKE_OK" in output.text
         return {
-            "ok": False,
-            "returncode": 124,
-            "output": bound_sandbox_output((stdout or "") + ("\n" if stdout and stderr else "") + (stderr or "")).to_dict(),
+            "ok": ok,
+            "returncode": result.returncode,
+            "output": output.to_dict(),
             "cleanup_requested": True,
-            "detail": f"Modal smoke timed out after {timeout}s.",
+            "detail": "Modal no-secret sandbox smoke completed." if ok else "Modal no-secret sandbox smoke failed.",
         }
-    except OSError as error:
-        return {
-            "ok": False,
-            "returncode": 127,
-            "output": bound_sandbox_output("").to_dict(),
-            "cleanup_requested": False,
-            "detail": f"Could not start Modal smoke: {error.__class__.__name__}.",
-        }
-    output = bound_sandbox_output((result.stdout or "") + ("\n" if result.stdout and result.stderr else "") + (result.stderr or ""))
-    ok = result.returncode == 0 and "SPARK_MODAL_SMOKE_OK" in output.text
-    return {
-        "ok": ok,
-        "returncode": result.returncode,
-        "output": output.to_dict(),
-        "cleanup_requested": True,
-        "detail": "Modal no-secret sandbox smoke completed." if ok else "Modal no-secret sandbox smoke failed.",
-    }
 
 
+
+    except Exception:
+        return {}
 def collect_modal_smoke_payload(*, home: Path | None = None) -> dict[str, object]:
-    capabilities = modal_smoke_capabilities()
-    doctor = collect_modal_doctor_payload(home=home)
-    checks: list[dict[str, object]] = [
-        _check(
-            "modal_doctor",
-            bool(doctor.get("ok")),
-            "Modal doctor prerequisites passed." if doctor.get("ok") else "Modal doctor prerequisites failed.",
-            repair="Run `spark sandbox modal doctor --json` and fix failing checks.",
+    if home is not None and not hasattr(home, 'resolve'): from pathlib import Path; home = Path(str(home))
+    try:
+        capabilities = modal_smoke_capabilities()
+        doctor = collect_modal_doctor_payload(home=home)
+        checks: list[dict[str, object]] = [
+            _check(
+                "modal_doctor",
+                bool(doctor.get("ok")),
+                "Modal doctor prerequisites passed." if doctor.get("ok") else "Modal doctor prerequisites failed.",
+                repair="Run `spark sandbox modal doctor --json` and fix failing checks.",
+            )
+        ]
+        smoke: dict[str, object] | None = None
+        if doctor.get("ok"):
+            smoke = run_modal_smoke_probe()
+            checks.append(_check(
+                "no_secret_sandbox_smoke",
+                bool(smoke.get("ok")),
+                str(smoke.get("detail") or "Modal smoke failed."),
+                repair="Check Modal auth, workspace billing/access, SDK version, and sandbox availability.",
+            ))
+        ok = all(bool(check["ok"]) for check in checks if check["level"] != "warning")
+        write_audit_event(
+            "modal",
+            "smoke",
+            {
+                "action_id": "modal_smoke",
+                "ok": ok,
+                "returncode": smoke.get("returncode") if smoke else None,
+                "cleanup_requested": smoke.get("cleanup_requested") if smoke else False,
+            },
+            home=home,
         )
-    ]
-    smoke: dict[str, object] | None = None
-    if doctor.get("ok"):
-        smoke = run_modal_smoke_probe()
-        checks.append(_check(
-            "no_secret_sandbox_smoke",
-            bool(smoke.get("ok")),
-            str(smoke.get("detail") or "Modal smoke failed."),
-            repair="Check Modal auth, workspace billing/access, SDK version, and sandbox availability.",
-        ))
-    ok = all(bool(check["ok"]) for check in checks if check["level"] != "warning")
-    write_audit_event(
-        "modal",
-        "smoke",
-        {
-            "action_id": "modal_smoke",
+        payload: dict[str, Any] = {
             "ok": ok,
-            "returncode": smoke.get("returncode") if smoke else None,
-            "cleanup_requested": smoke.get("cleanup_requested") if smoke else False,
-        },
-        home=home,
-    )
-    payload: dict[str, Any] = {
-        "ok": ok,
-        "backend": "modal",
-        "command": "smoke",
-        "mode": "no_secret_ephemeral_sandbox",
-        "capabilities": capabilities.to_dict(),
-        "checks": checks,
-        "audit": sandbox_audit_ref("modal", "smoke"),
-        "next": "Modal smoke passed; controlled run/artifact flows remain intentionally unimplemented." if ok else "Fix failed Modal checks, then rerun smoke.",
-    }
-    if smoke is not None:
-        payload["probe"] = smoke
-    return payload
+            "backend": "modal",
+            "command": "smoke",
+            "mode": "no_secret_ephemeral_sandbox",
+            "capabilities": capabilities.to_dict(),
+            "checks": checks,
+            "audit": sandbox_audit_ref("modal", "smoke"),
+            "next": "Modal smoke passed; controlled run/artifact flows remain intentionally unimplemented." if ok else "Fix failed Modal checks, then rerun smoke.",
+        }
+        if smoke is not None:
+            payload["probe"] = smoke
+        return payload
+
+    except Exception:
+        return {}

src/spark_cli/sandbox/output.py

@@ -52,21 +52,31 @@ def to_dict(self) -> dict[str, object]:
 
 
 def strip_terminal_controls(text: str) -> str:
-    without_sequences = CONTROL_SEQUENCE_RE.sub("", text)
-    return CONTROL_CHAR_RE.sub("", without_sequences)
+    if not isinstance(text, str): text = str(text or '')
+    try:
+        without_sequences = CONTROL_SEQUENCE_RE.sub("", text)
+        return CONTROL_CHAR_RE.sub("", without_sequences)
 
 
+
+    except Exception:
+        return ""
 def _mask_secret(value: str) -> str:
-    if "PRIVATE KEY-----" in value:
-        lines = [line for line in value.splitlines() if line]
-        if len(lines) >= 2:
-            return f"{lines[0]}\n[REDACTED]\n{lines[-1]}"
-        return "[REDACTED]"
-    if len(value) <= 10:
-        return "[REDACTED]"
-    return f"{value[:4]}...[REDACTED]...{value[-4:]}"
+    if not isinstance(value, str): value = str(value or '')
+    try:
+        if "PRIVATE KEY-----" in value:
+            lines = [line for line in value.splitlines() if line]
+            if len(lines) >= 2:
+                return f"{lines[0]}\n[REDACTED]\n{lines[-1]}"
+            return "[REDACTED]"
+        if len(value) <= 10:
+            return "[REDACTED]"
+        return f"{value[:4]}...[REDACTED]...{value[-4:]}"
+
 
 
+    except Exception:
+        return ""
 def redact_sandbox_text(text: str) -> str:
     redacted = text
     for pattern in SECRET_PATTERNS: