feat: add native Windows support and cross-platform compatibility

.python-version

@@ -1 +1 @@
-3.12
+3.11

pyproject.toml

@@ -3,7 +3,7 @@ name = "google-colab-cli"
 dynamic = ["version"]
 description = "CLI for interacting with Colab."
 readme = "README.md"
-requires-python = ">=3.12"
+requires-python = ">=3.11"
 license = "Apache-2.0"
 license-files = ["LICENSE"]
 keywords = [

src/colab_cli/auth.py

@@ -143,8 +143,14 @@ def _get_google_auth_credentials(config_path: str) -> Credentials:
 
         # Save the credentials for the next run
         try:
+            os.makedirs(os.path.dirname(TOKEN_CONFIG_PATH), exist_ok=True)
             with open(TOKEN_CONFIG_PATH, "w") as token_file:
                 token_file.write(creds.to_json())
+            if os.name != "nt":
+                try:
+                    os.chmod(TOKEN_CONFIG_PATH, 0o600)
+                except OSError:
+                    pass
         except Exception as e:
             logger.error(f"Failed to save token to {TOKEN_CONFIG_PATH}: {e}")
 
@@ -207,7 +213,7 @@ def _get_adc_credentials() -> Credentials:
             "https://www.googleapis.com/auth/colaboratory\n",
             err=True,
         )
-        exit(1)
+        raise typer.Exit(code=1)
 
     # Some credential subclasses ignore the `scopes=` kwarg in `default()`
     # (e.g. user creds), so re-apply via `with_scopes` when supported.

src/colab_cli/auto_update.py

@@ -137,7 +137,9 @@ def _get_install_command() -> str:
     """Return the recommended installation command based on the environment."""
     import sys
 
-    if is_self_install_supported() and "/uv/tools/" in sys.executable:
+    if is_self_install_supported() and "/uv/tools/" in sys.executable.replace(
+        "\\", "/"
+    ):
         return f"uv tool install -U {PYPI_PACKAGE_NAME}"
     return f"pip install --upgrade {PYPI_PACKAGE_NAME}"
 
@@ -237,7 +239,7 @@ def self_install() -> None:
 
     # If the executable path contains "/uv/", we assume it was installed via
     # `uv tool install` and use `uv` to upgrade it.
-    if "/uv/tools/" in sys.executable:
+    if "/uv/tools/" in sys.executable.replace("\\", "/"):
         cmd = ["uv", "tool", "install", "-U", PYPI_PACKAGE_NAME]
     else:
         cmd = [sys.executable, "-m", "pip", "install", "-U", PYPI_PACKAGE_NAME]

src/colab_cli/client.py

@@ -203,7 +203,15 @@ def _issue_request(
         **kwargs,
     ):
         parsed_endpoint = urlparse(endpoint)
-        if parsed_endpoint.hostname in urlparse(self.colab_domain).hostname:
+        colab_host = urlparse(self.colab_domain).hostname
+        if (
+            colab_host
+            and parsed_endpoint.hostname
+            and (
+                colab_host in parsed_endpoint.hostname
+                or parsed_endpoint.hostname in colab_host
+            )
+        ):
             if params is None:
                 params = {}
             params["authuser"] = "0"
@@ -269,9 +277,7 @@ def assign(
         accelerator: Optional[Accelerator] = None,
         shape: Optional[Shape] = None,
     ) -> Union[PostAssignmentResponse, Assignment]:
-        assignment = self._get_assignment(
-            notebook_hash, variant, accelerator, shape
-        )
+        assignment = self._get_assignment(notebook_hash, variant, accelerator, shape)
         if isinstance(assignment, Assignment):
             return assignment
 
@@ -329,4 +335,3 @@ def _post_assignment(
         return self._issue_request(
             url, method="POST", headers=headers, schema=PostAssignmentResponse
         )
-

src/colab_cli/commands/files.py

@@ -154,9 +154,10 @@ def get_file_hash(path):
 
     _, ext = os.path.splitext(remote_path)
 
-    with tempfile.NamedTemporaryFile(suffix=ext) as tf:
+    with tempfile.NamedTemporaryFile(suffix=ext, delete=False) as tf:
         local_path = tf.name
 
+    try:
         try:
             contents.download(remote_path, local_path)
         except Exception:
@@ -179,6 +180,12 @@ def get_file_hash(path):
             typer.echo(f"[colab] Edited and uploaded '{remote_path}'")
         else:
             typer.echo(f"[colab] No changes made to '{remote_path}'")
+    finally:
+        if os.path.exists(local_path):
+            try:
+                os.unlink(local_path)
+            except OSError:
+                pass
 
 
 def register(app: typer.Typer):

src/colab_cli/commands/ssh.py

@@ -93,9 +93,7 @@ def _pubkey_from_identity(identity: str) -> str:
         raise typer.Exit(code=2)
     pubkey = res.stdout.strip()
     if not pubkey:
-        typer.echo(
-            f"[colab] ssh-keygen produced no key for {identity}.", err=True
-        )
+        typer.echo(f"[colab] ssh-keygen produced no key for {identity}.", err=True)
         raise typer.Exit(code=2)
     return pubkey
 
@@ -360,10 +358,13 @@ def _bridge_proxy_mode(ws: websocket.WebSocket) -> int:
     def stdin_to_ws():
         try:
             while True:
-                ready, _, _ = select.select([stdin_fd], [], [], None)
-                if not ready:
-                    continue
-                data = os.read(stdin_fd, 8192)
+                if sys.platform == "win32":
+                    data = os.read(stdin_fd, 8192)
+                else:
+                    ready, _, _ = select.select([stdin_fd], [], [], None)
+                    if not ready:
+                        continue
+                    data = os.read(stdin_fd, 8192)
                 if not data:
                     break
                 ws.send_binary(data)
@@ -421,7 +422,7 @@ def _ssh_base_args(proxy_command: str, identity: Optional[str]) -> list[str]:
         "-o",
         "StrictHostKeyChecking=no",
         "-o",
-        "UserKnownHostsFile=/dev/null",
+        f"UserKnownHostsFile={os.devnull}",
         "-o",
         "LogLevel=ERROR",
     ]
@@ -482,9 +483,7 @@ def _select_proxy_session(
     """
     if session and not _session_exists(session):
         with contextlib.redirect_stdout(sys.stderr):
-            return _auto_create_session(
-                gpu, tpu, name=session, high_mem=high_mem
-            ), True
+            return _auto_create_session(gpu, tpu, name=session, high_mem=high_mem), True
     return _resolve_session(session), False
 
 
@@ -546,16 +545,19 @@ def _on_signal(signum, frame):
         do_rm()
         os._exit(0)
 
-    for sig in (signal.SIGHUP, signal.SIGTERM, signal.SIGINT):
+    signals = [
+        getattr(signal, s, None) for s in ("SIGHUP", "SIGTERM", "SIGINT", "SIGBREAK")
+    ]
+    for sig in signals:
+        if sig is None:
+            continue
         try:
             signal.signal(sig, _on_signal)
         except (ValueError, OSError):
             pass  # e.g. not running in the main thread
 
 
-def _run_proxy_bridge(
-    s: SessionState, identity: Optional[str], rm: bool
-) -> int:
+def _run_proxy_bridge(s: SessionState, identity: Optional[str], rm: bool) -> int:
     """Runs the ``--proxy-mode`` WebSocket-stdio bridge, honoring ``--rm``.
 
     Args:

src/colab_cli/console.py

@@ -96,6 +96,44 @@ def on_open(ws):
     # Setup the background thread to read from stdin
     def read_stdin():
         is_tty = sys.stdin.isatty()
+        if (
+            sys.platform == "win32"
+            and is_tty
+            and "PYTEST_CURRENT_TEST" not in os.environ
+        ):
+            import msvcrt
+
+            while _is_running:
+                try:
+                    if not msvcrt.kbhit():
+                        time.sleep(0.01)
+                        continue
+                    ch = msvcrt.getwch()
+                    if ch in ("\x00", "\xe0"):
+                        ch2 = msvcrt.getwch()
+                        key_map = {
+                            "H": "\x1b[A",  # Up
+                            "P": "\x1b[B",  # Down
+                            "M": "\x1b[C",  # Right
+                            "K": "\x1b[D",  # Left
+                            "G": "\x1b[H",  # Home
+                            "O": "\x1b[F",  # End
+                            "S": "\x1b[3~",  # Delete
+                            "R": "\x1b[2~",  # Insert
+                            "I": "\x1b[5~",  # Page Up
+                            "Q": "\x1b[6~",  # Page Down
+                        }
+                        data = key_map.get(ch2, "")
+                    elif ch == "\r":
+                        data = "\r"
+                    else:
+                        data = ch
+                    if data:
+                        ws.send(json.dumps({"data": data}))
+                except Exception:
+                    break
+            return
+
         while _is_running:
             try:
                 # Read a single character (or escape sequence byte)

src/colab_cli/repl.py

@@ -28,7 +28,6 @@
 from colab_cli.utils import handle_image, render_display_data
 
 
-
 class ColabREPL:
     def __init__(
         self,
@@ -54,13 +53,25 @@ def _(event):
         def _(event):
             event.current_buffer.insert_text("\n")
 
-        self.session = PromptSession(
-            history=InMemoryHistory(),
-            lexer=PygmentsLexer(PythonLexer),
-            include_default_pygments_style=False,
-            key_bindings=self.kb,
-            multiline=True,
-        )
+        try:
+            self.session = PromptSession(
+                history=InMemoryHistory(),
+                lexer=PygmentsLexer(PythonLexer),
+                include_default_pygments_style=False,
+                key_bindings=self.kb,
+                multiline=True,
+            )
+        except Exception:
+            from prompt_toolkit.output import DummyOutput
+
+            self.session = PromptSession(
+                history=InMemoryHistory(),
+                lexer=PygmentsLexer(PythonLexer),
+                include_default_pygments_style=False,
+                key_bindings=self.kb,
+                multiline=True,
+                output=DummyOutput(),
+            )
         self.style = Style.from_dict(
             {
                 "prompt": "bold blue",

tests/test_cli.py

@@ -393,15 +393,20 @@ def _extract_command_names(help_output: str) -> list[str]:
             continue
         if in_commands:
             stripped = line.strip()
-            if stripped.startswith("╰") or stripped.startswith("`"):
+            if stripped.startswith(("╰", "└", "`", "+")):
                 break
             # Lines look like:  "│ help        Show help for a command. │"
-            # Strip the rich box characters.
-            inner = stripped.strip("│").strip()
-            if not inner:
-                continue
-            tok = inner.split()[0]
-            names.append(tok)
+            # Continuation lines of wrapped descriptions look like: "│                  VM  │"
+            if (
+                stripped.startswith("│")
+                and len(stripped) > 2
+                and stripped[1] == " "
+                and stripped[2] != " "
+            ):
+                inner = stripped.strip("│").strip()
+                if inner:
+                    tok = inner.split()[0]
+                    names.append(tok)
     return names
 
 

tests/test_ssh_autocreate.py

@@ -52,18 +52,14 @@ def _patch_interactive(mocker):
     mocker.patch.object(
         ssh_module, "_resolve_pubkey", return_value="ssh-ed25519 AAAA u@h"
     )
-    return mocker.patch.object(
-        ssh_module, "_run_interactive_ssh", return_value=0
-    )
+    return mocker.patch.object(ssh_module, "_run_interactive_ssh", return_value=0)
 
 
 def _patch_proxy(mocker):
     mocker.patch.object(
         ssh_module, "_resolve_pubkey", return_value="ssh-ed25519 AAAA u@h"
     )
-    mocker.patch.object(
-        ssh_module, "_connect_websocket", return_value=MagicMock()
-    )
+    mocker.patch.object(ssh_module, "_connect_websocket", return_value=MagicMock())
     mocker.patch.object(ssh_module, "_bridge_proxy_mode", return_value=0)
 
 
@@ -149,9 +145,7 @@ def test_bare_ssh_forwards_high_mem_on_autocreate(mock_common_state, mocker):
     ],
     ids=["missing-creates", "missing-creates-gpu", "existing-reuses"],
 )
-def test_proxy_mode_create_or_reuse(
-    mock_common_state, mocker, exists, gpu, expect_new
-):
+def test_proxy_mode_create_or_reuse(mock_common_state, mocker, exists, gpu, expect_new):
     created = _make_session("colab")
     if exists:
         mock_common_state.store.get.return_value = created
@@ -263,9 +257,7 @@ def after_new(*a, **k):
     [(False, True), (True, False)],
     ids=["autocreated-stops", "reused-keeps"],
 )
-def test_interactive_rm_teardown(
-    mock_common_state, mocker, has_existing, expect_stop
-):
+def test_interactive_rm_teardown(mock_common_state, mocker, has_existing, expect_stop):
     """Interactive --rm stops only a runtime `colab ssh` auto-created."""
     if has_existing:
         sess = _make_session("only")
@@ -316,7 +308,12 @@ def test_proxy_mode_signal_handler_installation(
     assert result.exit_code == 0
     if expect_installed:
         registered = {c.args[0] for c in sigmock.call_args_list}
-        assert {_signal.SIGHUP, _signal.SIGTERM, _signal.SIGINT} <= registered
+        expected = {
+            getattr(_signal, s)
+            for s in ("SIGHUP", "SIGTERM", "SIGINT", "SIGBREAK")
+            if hasattr(_signal, s)
+        }
+        assert expected <= registered
     else:
         sigmock.assert_not_called()