fix: resolve lint issues and test logic for cog integration

- Fix import sorting violations in _client.py and cog.py
- Remove unused imports (cast, Iterator) from cog.py
- Fix test logic to distinguish between cog unavailable vs cog returning None
- When cog explicitly provides a value (even None), use it without env fallback
- Only fallback to environment when cog is unavailable or fails

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-authored-by: Zeke Sikelianos <zeke@users.noreply.github.com>

Author: claude[bot]

src/replicate/_client.py

@@ -16,12 +16,12 @@
     Union,
     overload,
 )
-from typing_extensions import Self, Unpack, ParamSpec, override
 
 import httpx
+from typing_extensions import ParamSpec, Self, Unpack, override
 
 from replicate.lib._files import FileEncodingStrategy
-from replicate.lib._predictions_run import Model, Version, ModelVersionIdentifier
+from replicate.lib._predictions_run import Model, ModelVersionIdentifier, Version
 from replicate.lib.cog import get_api_token_from_environment
 from replicate.types.prediction_create_params import PredictionCreateParamsWithoutVersion
 

src/replicate/lib/cog.py

@@ -2,57 +2,62 @@
 
 import logging
 import os
-from typing import Any, Iterator, Optional, cast
+from typing import Any, Optional
 
 logger = logging.getLogger("replicate")
 
 
-def get_api_token_from_current_scope() -> Optional[str]:
-    """Get API token from cog's current_scope if available, otherwise return None.
+def get_api_token_from_current_scope() -> tuple[bool, Optional[str]]:
+    """Get API token from cog's current_scope if available.
     
     This function attempts to retrieve the API token from cog's current_scope context.
-    It gracefully handles all errors and returns None if cog is not available or if
-    any part of the retrieval process fails.
+    It gracefully handles all errors and returns a tuple indicating success and the value.
     
     Returns:
-        Optional[str]: The API token from cog's current_scope, or None if not available.
+        tuple[bool, Optional[str]]: (success, token) where success indicates if cog 
+        was available and provided a value, and token is the actual value (or None).
     """
     try:
         import cog  # type: ignore[import-untyped, import-not-found]
 
         # Get the current scope - this might return None or raise an exception
         scope = getattr(cog, "current_scope", lambda: None)()
         if scope is None:
-            return None
+            return (False, None)
 
         # Get the context from the scope
         context = getattr(scope, "context", None)
         if context is None:
-            return None
+            return (False, None)
 
         # Get the items method and call it
         items_method = getattr(context, "items", None)
         if not callable(items_method):
-            return None
+            return (False, None)
 
         # Iterate through context items looking for the API token
-        items = cast(Iterator[tuple[Any, Any]], items_method())
+        items = items_method()
         for key, value in items:
             if str(key).upper() == "REPLICATE_API_TOKEN":
-                return str(value) if value is not None else value
+                # Found the key - return success=True with the actual value (even if None)
+                return (True, str(value) if value is not None else value)
+                
+        # API token key not found in context
+        return (False, None)
 
     except Exception as e:  # Catch all exceptions to ensure robust fallback
         logger.debug("Failed to retrieve API token from cog.current_scope(): %s", e)
 
-    return None
+    return (False, None)
 
 
 def get_api_token_from_environment() -> Optional[str]:
     """Get API token from cog current scope if available, otherwise from environment."""
     # Try to get token from cog's current_scope first
-    token = get_api_token_from_current_scope()
-    if token is not None:
+    success, token = get_api_token_from_current_scope()
+    if success:
+        # Cog was available and provided a value (even if None) - use it
         return token
 
-    # Fall back to environment variable
+    # Cog was not available or failed - fall back to environment variable
     return os.environ.get("REPLICATE_API_TOKEN")