Add files using upload-large-folder tool

lib/python3.12/site-packages/wandb/agents/pyagent.py

@@ -0,0 +1,357 @@
+"""Agent - Agent object.
+
+Manage wandb agent.
+
+"""
+
+import ctypes
+import logging
+import os
+import queue
+import socket
+import threading
+import time
+import traceback
+
+import wandb
+from wandb.apis import InternalApi
+from wandb.sdk.launch.sweeps import utils as sweep_utils
+
+logger = logging.getLogger(__name__)
+
+
+def _terminate_thread(thread):
+    if not thread.is_alive():
+        return
+    if hasattr(thread, "_terminated"):
+        return
+    thread._terminated = True
+    tid = getattr(thread, "_thread_id", None)
+    if tid is None:
+        for k, v in threading._active.items():
+            if v is thread:
+                tid = k
+    if tid is None:
+        # This should never happen
+        return
+    logger.debug(f"Terminating thread: {tid}")
+    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(
+        ctypes.c_long(tid), ctypes.py_object(Exception)
+    )
+    if res == 0:
+        # This should never happen
+        return
+    elif res != 1:
+        # Revert
+        logger.debug(f"Termination failed for thread {tid}")
+        ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid), None)
+
+
+class Job:
+    def __init__(self, command):
+        self.command = command
+        job_type = command.get("type")
+        self.type = job_type
+        self.run_id = command.get("run_id")
+        self.config = command.get("args")
+
+    def __repr__(self):
+        if self.type == "run":
+            return f"Job({self.run_id},{self.config})"
+        elif self.type == "stop":
+            return f"stop({self.run_id})"
+        else:
+            return "exit"
+
+
+class RunStatus:
+    QUEUED = "QUEUED"
+    RUNNING = "RUNNING"
+    STOPPED = "STOPPED"
+    ERRORED = "ERRORED"
+    DONE = "DONE"
+
+
+class Agent:
+    FLAPPING_MAX_SECONDS = 60
+    FLAPPING_MAX_FAILURES = 3
+    MAX_INITIAL_FAILURES = 5
+
+    def __init__(
+        self, sweep_id=None, project=None, entity=None, function=None, count=None
+    ):
+        self._sweep_path = sweep_id
+        self._sweep_id = None
+        self._project = project
+        self._entity = entity
+        self._function = function
+        self._count = count
+        # glob_config = os.path.expanduser('~/.config/wandb/settings')
+        # loc_config = 'wandb/settings'
+        # files = (glob_config, loc_config)
+        self._api = InternalApi()
+        self._agent_id = None
+        self._max_initial_failures = wandb.env.get_agent_max_initial_failures(
+            self.MAX_INITIAL_FAILURES
+        )
+        # if the directory to log to is not set, set it
+        if os.environ.get(wandb.env.DIR) is None:
+            os.environ[wandb.env.DIR] = os.path.abspath(os.getcwd())
+
+    def _init(self):
+        # These are not in constructor so that Agent instance can be rerun
+        self._run_threads = {}
+        self._run_status = {}
+        self._queue = queue.Queue()
+        self._exit_flag = False
+        self._exceptions = {}
+        self._start_time = time.time()
+
+    def _register(self):
+        logger.debug("Agent._register()")
+        agent = self._api.register_agent(socket.gethostname(), sweep_id=self._sweep_id)
+        self._agent_id = agent["id"]
+        logger.debug(f"agent_id = {self._agent_id}")
+
+    def _setup(self):
+        logger.debug("Agent._setup()")
+        self._init()
+        parts = dict(entity=self._entity, project=self._project, name=self._sweep_path)
+        err = sweep_utils.parse_sweep_id(parts)
+        if err:
+            wandb.termerror(err)
+            return
+        entity = parts.get("entity") or self._entity
+        project = parts.get("project") or self._project
+        sweep_id = parts.get("name") or self._sweep_id
+        if sweep_id:
+            os.environ[wandb.env.SWEEP_ID] = sweep_id
+        if entity:
+            wandb.env.set_entity(entity)
+        if project:
+            wandb.env.set_project(project)
+        if sweep_id:
+            self._sweep_id = sweep_id
+        self._register()
+
+    def _stop_run(self, run_id):
+        logger.debug(f"Stopping run {run_id}.")
+        self._run_status[run_id] = RunStatus.STOPPED
+        thread = self._run_threads.get(run_id)
+        if thread:
+            _terminate_thread(thread)
+
+    def _stop_all_runs(self):
+        logger.debug("Stopping all runs.")
+        for run in list(self._run_threads.keys()):
+            self._stop_run(run)
+
+    def _exit(self):
+        self._stop_all_runs()
+        self._exit_flag = True
+        # _terminate_thread(self._main_thread)
+
+    def _heartbeat(self):
+        while True:
+            if self._exit_flag:
+                return
+            # if not self._main_thread.is_alive():
+            #     return
+            run_status = {
+                run: True
+                for run, status in self._run_status.items()
+                if status in (RunStatus.QUEUED, RunStatus.RUNNING)
+            }
+            commands = self._api.agent_heartbeat(self._agent_id, {}, run_status)
+            if commands:
+                job = Job(commands[0])
+                logger.debug(f"Job received: {job}")
+                if job.type in ["run", "resume"]:
+                    self._queue.put(job)
+                    self._run_status[job.run_id] = RunStatus.QUEUED
+                elif job.type == "stop":
+                    self._stop_run(job.run_id)
+                elif job.type == "exit":
+                    self._exit()
+                    return
+            time.sleep(5)
+
+    def _run_jobs_from_queue(self):
+        global _INSTANCES
+        _INSTANCES += 1
+        try:
+            waiting = False
+            count = 0
+            while True:
+                if self._exit_flag:
+                    return
+                try:
+                    try:
+                        job = self._queue.get(timeout=5)
+                        if self._exit_flag:
+                            logger.debug("Exiting main loop due to exit flag.")
+                            wandb.termlog("Sweep Agent: Exiting.")
+                            return
+                    except queue.Empty:
+                        if not waiting:
+                            logger.debug("Paused.")
+                            wandb.termlog("Sweep Agent: Waiting for job.")
+                            waiting = True
+                        time.sleep(5)
+                        if self._exit_flag:
+                            logger.debug("Exiting main loop due to exit flag.")
+                            wandb.termlog("Sweep Agent: Exiting.")
+                            return
+                        continue
+                    if waiting:
+                        logger.debug("Resumed.")
+                        wandb.termlog("Job received.")
+                        waiting = False
+                    count += 1
+                    run_id = job.run_id
+                    if self._run_status[run_id] == RunStatus.STOPPED:
+                        continue
+                    logger.debug(f"Spawning new thread for run {run_id}.")
+                    thread = threading.Thread(target=self._run_job, args=(job,))
+                    self._run_threads[run_id] = thread
+                    thread.start()
+                    self._run_status[run_id] = RunStatus.RUNNING
+                    thread.join()
+                    logger.debug(f"Thread joined for run {run_id}.")
+                    if self._run_status[run_id] == RunStatus.RUNNING:
+                        self._run_status[run_id] = RunStatus.DONE
+                    elif self._run_status[run_id] == RunStatus.ERRORED:
+                        exc = self._exceptions[run_id]
+                        exc_type, exc_value, exc_traceback = (
+                            exc.__class__,
+                            exc,
+                            exc.__traceback__,
+                        )
+                        exc_traceback_formatted = traceback.format_exception(
+                            exc_type, exc_value, exc_traceback
+                        )
+                        exc_repr = "".join(exc_traceback_formatted)
+                        logger.error(f"Run {run_id} errored:\n{exc_repr}")
+                        wandb.termerror(f"Run {run_id} errored:\n{exc_repr}")
+                        if os.getenv(wandb.env.AGENT_DISABLE_FLAPPING) == "true":
+                            self._exit_flag = True
+                            return
+                        elif (
+                            time.time() - self._start_time < self.FLAPPING_MAX_SECONDS
+                        ) and (len(self._exceptions) >= self.FLAPPING_MAX_FAILURES):
+                            msg = f"Detected {self.FLAPPING_MAX_FAILURES} failed runs in the first {self.FLAPPING_MAX_SECONDS} seconds, killing sweep."
+                            logger.error(msg)
+                            wandb.termerror(msg)
+                            wandb.termlog(
+                                "To disable this check set WANDB_AGENT_DISABLE_FLAPPING=true"
+                            )
+                            self._exit_flag = True
+                            return
+                        if (
+                            self._max_initial_failures < len(self._exceptions)
+                            and len(self._exceptions) >= count
+                        ):
+                            msg = f"Detected {self._max_initial_failures} failed runs in a row at start, killing sweep."
+                            logger.error(msg)
+                            wandb.termerror(msg)
+                            wandb.termlog(
+                                "To change this value set WANDB_AGENT_MAX_INITIAL_FAILURES=val"
+                            )
+                            self._exit_flag = True
+                            return
+                    if self._count and self._count == count:
+                        logger.debug("Exiting main loop because max count reached.")
+                        self._exit_flag = True
+                        return
+                except KeyboardInterrupt:
+                    logger.debug("Ctrl + C detected. Stopping sweep.")
+                    wandb.termlog("Ctrl + C detected. Stopping sweep.")
+                    self._exit()
+                    return
+                except Exception:
+                    if self._exit_flag:
+                        logger.debug("Exiting main loop due to exit flag.")
+                        wandb.termlog("Sweep Agent: Killed.")
+                        return
+                    else:
+                        raise
+        finally:
+            _INSTANCES -= 1
+
+    def _run_job(self, job):
+        try:
+            run_id = job.run_id
+
+            config_file = os.path.join(
+                "wandb", "sweep-" + self._sweep_id, "config-" + run_id + ".yaml"
+            )
+            os.environ[wandb.env.RUN_ID] = run_id
+            base_dir = os.environ.get(wandb.env.DIR, "")
+            sweep_param_path = os.path.join(base_dir, config_file)
+            os.environ[wandb.env.SWEEP_PARAM_PATH] = sweep_param_path
+            wandb.wandb_lib.config_util.save_config_file_from_dict(
+                sweep_param_path, job.config
+            )
+            os.environ[wandb.env.SWEEP_ID] = self._sweep_id
+            wandb.teardown()
+
+            wandb.termlog(f"Agent Starting Run: {run_id} with config:")
+            for k, v in job.config.items():
+                wandb.termlog("\t{}: {}".format(k, v["value"]))
+
+            self._function()
+            wandb.finish()
+        except KeyboardInterrupt:
+            raise
+        except Exception as e:
+            wandb.finish(exit_code=1)
+            if self._run_status[run_id] == RunStatus.RUNNING:
+                self._run_status[run_id] = RunStatus.ERRORED
+                self._exceptions[run_id] = e
+        finally:
+            # clean up the environment changes made
+            os.environ.pop(wandb.env.RUN_ID, None)
+            os.environ.pop(wandb.env.SWEEP_ID, None)
+            os.environ.pop(wandb.env.SWEEP_PARAM_PATH, None)
+
+    def run(self):
+        logger.info(
+            f"Starting sweep agent: entity={self._entity}, project={self._project}, count={self._count}"
+        )
+        self._setup()
+        # self._main_thread = threading.Thread(target=self._run_jobs_from_queue)
+        self._heartbeat_thread = threading.Thread(target=self._heartbeat)
+        self._heartbeat_thread.daemon = True
+        # self._main_thread.start()
+        self._heartbeat_thread.start()
+        # self._main_thread.join()
+        self._run_jobs_from_queue()
+
+
+def pyagent(sweep_id, function, entity=None, project=None, count=None):
+    """Generic agent entrypoint, used for CLI or jupyter.
+
+    Args:
+        sweep_id (dict): Sweep ID generated by CLI or sweep API
+        function (func, optional): A function to call instead of the "program"
+        entity (str, optional): W&B Entity
+        project (str, optional): W&B Project
+        count (int, optional): the number of trials to run.
+    """
+    if not callable(function):
+        raise TypeError("function parameter must be callable!")
+    agent = Agent(
+        sweep_id,
+        function=function,
+        entity=entity,
+        project=project,
+        count=count,
+    )
+    agent.run()
+
+
+_INSTANCES = 0
+
+
+def is_running():
+    return bool(_INSTANCES)

lib/python3.12/site-packages/wandb/analytics/__init__.py

@@ -0,0 +1,3 @@
+__all__ = ("Sentry",)
+
+from .sentry import Sentry

lib/python3.12/site-packages/wandb/analytics/sentry.py

@@ -0,0 +1,262 @@
+from __future__ import annotations
+
+__all__ = ("Sentry",)
+
+
+import atexit
+import functools
+import os
+import pathlib
+import sys
+from types import TracebackType
+from typing import TYPE_CHECKING, Any, Callable, Literal
+from urllib.parse import quote
+
+import sentry_sdk  # type: ignore
+import sentry_sdk.scope  # type: ignore
+import sentry_sdk.utils  # type: ignore
+from typing_extensions import Never
+
+import wandb
+import wandb.env
+import wandb.util
+
+if TYPE_CHECKING:
+    import wandb.sdk.internal.settings_static
+
+SENTRY_DEFAULT_DSN = (
+    "https://2592b1968ea94cca9b5ef5e348e094a7@o151352.ingest.sentry.io/4504800232407040"
+)
+
+SessionStatus = Literal["ok", "exited", "crashed", "abnormal"]
+
+
+def _safe_noop(func: Callable) -> Callable:
+    """Decorator to ensure that Sentry methods do nothing if disabled and don't raise."""
+
+    @functools.wraps(func)
+    def wrapper(self: type[Sentry], *args: Any, **kwargs: Any) -> Any:
+        if self._disabled:
+            return None
+        try:
+            return func(self, *args, **kwargs)
+        except Exception as e:
+            # do not call self.exception here to avoid infinite recursion
+            if func.__name__ != "exception":
+                self.exception(f"Error in {func.__name__}: {e}")
+            return None
+
+    return wrapper
+
+
+class Sentry:
+    _disabled: bool
+
+    def __init__(self) -> None:
+        self._disabled = not wandb.env.error_reporting_enabled()
+        self._sent_messages: set = set()
+
+        self.dsn = os.environ.get(wandb.env.SENTRY_DSN, SENTRY_DEFAULT_DSN)
+
+        self.scope: sentry_sdk.scope.Scope | None = None
+
+        # ensure we always end the Sentry session
+        atexit.register(self.end_session)
+
+    @property
+    def environment(self) -> str:
+        """Return the environment we're running in."""
+        # check if we're in a git repo
+        is_git = pathlib.Path(__file__).parent.parent.parent.joinpath(".git").exists()
+
+        # these match the environments for gorilla
+        return "development" if is_git else "production"
+
+    @_safe_noop
+    def setup(self) -> None:
+        """Setup Sentry SDK.
+
+        We use lower-level APIs (i.e., not sentry_sdk.init) here
+        to avoid the possibility of interfering with the user's
+        own Sentry SDK setup.
+        """
+        client = sentry_sdk.Client(
+            dsn=self.dsn,
+            default_integrations=False,
+            environment=self.environment,
+            release=wandb.__version__,
+        )
+        self.scope = sentry_sdk.get_global_scope().fork()
+        self.scope.clear()
+        self.scope.set_client(client)
+
+    @_safe_noop
+    def message(self, message: str, repeat: bool = True) -> str | None:
+        """Send a message to Sentry."""
+        if not repeat and message in self._sent_messages:
+            return None
+        self._sent_messages.add(message)
+        with sentry_sdk.scope.use_isolation_scope(self.scope):  # type: ignore
+            return sentry_sdk.capture_message(message)  # type: ignore
+
+    @_safe_noop
+    def exception(
+        self,
+        exc: str
+        | BaseException
+        | tuple[
+            type[BaseException] | None,
+            BaseException | None,
+            TracebackType | None,
+        ]
+        | None,
+        handled: bool = False,
+        status: SessionStatus | None = None,
+    ) -> str | None:
+        """Log an exception to Sentry."""
+        if isinstance(exc, str):
+            exc_info = sentry_sdk.utils.exc_info_from_error(Exception(exc))
+        elif isinstance(exc, BaseException):
+            exc_info = sentry_sdk.utils.exc_info_from_error(exc)
+        else:
+            exc_info = sys.exc_info()
+
+        event, _ = sentry_sdk.utils.event_from_exception(
+            exc_info,
+            client_options=self.scope.get_client().options,  # type: ignore
+            mechanism={"type": "generic", "handled": handled},
+        )
+        event_id = None
+        try:
+            with sentry_sdk.scope.use_isolation_scope(self.scope):  # type: ignore
+                event_id = sentry_sdk.capture_event(event)  # type: ignore
+        except Exception:
+            pass
+
+        # if the status is not explicitly set, we'll set it to "crashed" if the exception
+        # was unhandled, or "errored" if it was handled
+        status = status or ("crashed" if not handled else "errored")  # type: ignore
+        self.mark_session(status=status)
+
+        client = self.scope.get_client()  # type: ignore
+        if client is not None:
+            client.flush()
+
+        return event_id
+
+    def reraise(self, exc: Any) -> Never:
+        """Re-raise an exception after logging it to Sentry.
+
+        Use this for top-level exceptions when you want the user to see the traceback.
+
+        Must be called from within an exception handler.
+        """
+        self.exception(exc)
+        # this will messily add this "reraise" function to the stack trace,
+        # but hopefully it's not too bad
+        raise exc.with_traceback(sys.exc_info()[2])
+
+    @_safe_noop
+    def start_session(self) -> None:
+        """Start a new session."""
+        assert self.scope is not None
+        # get the current client and scope
+        session = self.scope._session
+
+        # if there's no session, start one
+        if session is None:
+            self.scope.start_session()
+
+    @_safe_noop
+    def end_session(self) -> None:
+        """End the current session."""
+        assert self.scope is not None
+        # get the current client and scope
+        client = self.scope.get_client()
+        session = self.scope._session
+
+        if session is not None and client is not None:
+            self.scope.end_session()
+            client.flush()
+
+    @_safe_noop
+    def mark_session(self, status: SessionStatus | None = None) -> None:
+        """Mark the current session with a status."""
+        assert self.scope is not None
+        session = self.scope._session
+
+        if session is not None:
+            session.update(status=status)
+
+    @_safe_noop
+    def configure_scope(
+        self,
+        tags: dict[str, Any] | None = None,
+        process_context: str | None = None,
+    ) -> None:
+        """Configure the Sentry scope for the current thread.
+
+        This function should be called at the beginning of every thread that
+        will send events to Sentry. It sets the tags that will be applied to
+        all events sent from this thread. It also tries to start a session
+        if one doesn't already exist for this thread.
+        """
+        assert self.scope is not None
+        settings_tags = (
+            "entity",
+            "project",
+            "run_id",
+            "run_url",
+            "sweep_url",
+            "sweep_id",
+            "deployment",
+            "launch",
+            "_platform",
+        )
+
+        # set context
+        if process_context:
+            self.scope.set_tag("process_context", process_context)
+
+        # apply settings tags
+        if tags is None:
+            return None
+
+        for tag in settings_tags:
+            val = tags.get(tag, None)
+            if val not in (None, ""):
+                self.scope.set_tag(tag, val)
+
+        if tags.get("_colab", None):
+            python_runtime = "colab"
+        elif tags.get("_jupyter", None):
+            python_runtime = "jupyter"
+        elif tags.get("_ipython", None):
+            python_runtime = "ipython"
+        else:
+            python_runtime = "python"
+        self.scope.set_tag("python_runtime", python_runtime)
+
+        # Construct run_url and sweep_url given run_id and sweep_id
+        for obj in ("run", "sweep"):
+            obj_id, obj_url = f"{obj}_id", f"{obj}_url"
+            if tags.get(obj_url, None):
+                continue
+
+            try:
+                app_url = wandb.util.app_url(tags["base_url"])  # type: ignore
+                entity, project = (quote(tags[k]) for k in ("entity", "project"))  # type: ignore
+                self.scope.set_tag(
+                    obj_url,
+                    f"{app_url}/{entity}/{project}/{obj}s/{tags[obj_id]}",
+                )
+            except Exception:
+                pass
+
+        email = tags.get("email")
+        if email:
+            self.scope.user = {"email": email}
+
+        # todo: add back the option to pass general tags see: c645f625d1c1a3db4a6b0e2aa8e924fee101904c (wandb/util.py)
+
+        self.start_session()

lib/python3.12/site-packages/wandb/apis/__init__.py

@@ -0,0 +1,48 @@
+"""api."""
+
+from typing import Callable
+
+import requests
+from urllib3.exceptions import InsecureRequestWarning
+
+import wandb
+from wandb import env, util
+
+
+def _disable_ssl() -> Callable[[], None]:
+    # Because third party libraries may also use requests, we monkey patch it globally
+    # and turn off urllib3 warnings instead printing a global warning to the user.
+    wandb.termwarn(
+        "Disabling SSL verification.  Connections to this server are not verified and may be insecure!"
+    )
+
+    requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)
+    old_merge_environment_settings = requests.Session.merge_environment_settings
+
+    def merge_environment_settings(self, url, proxies, stream, verify, cert):
+        settings = old_merge_environment_settings(
+            self, url, proxies, stream, verify, cert
+        )
+        settings["verify"] = False
+        return settings
+
+    requests.Session.merge_environment_settings = merge_environment_settings
+
+    def reset():
+        requests.Session.merge_environment_settings = old_merge_environment_settings
+
+    return reset
+
+
+if env.ssl_disabled():
+    _disable_ssl()
+
+
+reset_path = util.vendor_setup()
+
+from .internal import Api as InternalApi  # noqa
+from .public import Api as PublicApi  # noqa
+
+reset_path()
+
+__all__ = ["InternalApi", "PublicApi"]

lib/python3.12/site-packages/wandb/apis/attrs.py

@@ -0,0 +1,50 @@
+from typing import Any, MutableMapping
+
+import wandb
+
+from ..sdk.lib import ipython
+
+
+class Attrs:
+    def __init__(self, attrs: MutableMapping[str, Any]):
+        self._attrs = attrs
+
+    def snake_to_camel(self, string):
+        camel = "".join([i.title() for i in string.split("_")])
+        return camel[0].lower() + camel[1:]
+
+    def display(self, height=420, hidden=False) -> bool:
+        """Display this object in jupyter."""
+        if wandb.run and wandb.run._settings.silent:
+            return False
+
+        if not ipython.in_jupyter():
+            return False
+
+        html = self.to_html(height, hidden)
+        if html is None:
+            wandb.termwarn("This object does not support `.display()`")
+            return False
+
+        try:
+            from IPython import display
+        except ImportError:
+            wandb.termwarn(".display() only works in jupyter environments")
+            return False
+
+        display.display(display.HTML(html))
+        return True
+
+    def to_html(self, *args, **kwargs):
+        return None
+
+    def __getattr__(self, name):
+        key = self.snake_to_camel(name)
+        if key == "user":
+            raise AttributeError
+        if key in self._attrs.keys():
+            return self._attrs[key]
+        elif name in self._attrs.keys():
+            return self._attrs[name]
+        else:
+            raise AttributeError(f"{repr(self)!r} object has no attribute {name!r}")

lib/python3.12/site-packages/wandb/apis/importers/__init__.py

@@ -0,0 +1 @@
+from .internals.util import Namespace

lib/python3.12/site-packages/wandb/apis/importers/internals/internal.py

@@ -0,0 +1,382 @@
+import json
+import logging
+import math
+import os
+import queue
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Dict, Iterable, Optional
+
+import numpy as np
+from google.protobuf.json_format import ParseDict
+from tenacity import retry, stop_after_attempt, wait_random_exponential
+
+from wandb import Artifact
+from wandb.proto import wandb_internal_pb2 as pb
+from wandb.proto import wandb_settings_pb2
+from wandb.proto import wandb_telemetry_pb2 as telem_pb
+from wandb.sdk.interface.interface import file_policy_to_enum
+from wandb.sdk.interface.interface_queue import InterfaceQueue
+from wandb.sdk.internal import context
+from wandb.sdk.internal.sender import SendManager
+from wandb.sdk.internal.settings_static import SettingsStatic
+from wandb.util import coalesce, recursive_cast_dictlike_to_dict
+
+from .protocols import ImporterRun
+
+ROOT_DIR = "./wandb-importer"
+
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+
+if os.getenv("WANDB_IMPORTER_ENABLE_RICH_LOGGING"):
+    from rich.logging import RichHandler
+
+    logger.addHandler(RichHandler(rich_tracebacks=True, tracebacks_show_locals=True))
+else:
+    console_handler = logging.StreamHandler()
+    console_handler.setLevel(logging.INFO)
+
+    formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
+    console_handler.setFormatter(formatter)
+
+    logger.addHandler(console_handler)
+
+
+exp_retry = retry(
+    wait=wait_random_exponential(multiplier=1, max=10), stop=stop_after_attempt(3)
+)
+
+
+class AlternateSendManager(SendManager):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._send_artifact = exp_retry(self._send_artifact)
+
+
+@dataclass(frozen=True)
+class SendManagerConfig:
+    """Configure which parts of SendManager tooling to use."""
+
+    use_artifacts: bool = False
+    log_artifacts: bool = False
+    metadata: bool = False
+    files: bool = False
+    media: bool = False
+    code: bool = False
+    history: bool = False
+    summary: bool = False
+    terminal_output: bool = False
+
+
+@dataclass
+class RecordMaker:
+    run: ImporterRun
+    interface: InterfaceQueue = InterfaceQueue()
+
+    @property
+    def run_dir(self) -> str:
+        p = Path(f"{ROOT_DIR}/{self.run.run_id()}/wandb")
+        p.mkdir(parents=True, exist_ok=True)
+        return f"{ROOT_DIR}/{self.run.run_id()}"
+
+    def make_artifacts_only_records(
+        self,
+        artifacts: Optional[Iterable[Artifact]] = None,
+        used_artifacts: Optional[Iterable[Artifact]] = None,
+    ) -> Iterable[pb.Record]:
+        """Only make records required to upload artifacts.
+
+        Escape hatch for adding extra artifacts to a run.
+        """
+        yield self._make_run_record()
+
+        if used_artifacts:
+            for art in used_artifacts:
+                yield self._make_artifact_record(art, use_artifact=True)
+
+        if artifacts:
+            for art in artifacts:
+                yield self._make_artifact_record(art)
+
+    def make_records(
+        self,
+        config: SendManagerConfig,
+    ) -> Iterable[pb.Record]:
+        """Make all the records that constitute a run."""
+        yield self._make_run_record()
+        yield self._make_telem_record()
+
+        include_artifacts = config.log_artifacts or config.use_artifacts
+        yield self._make_files_record(
+            include_artifacts, config.files, config.media, config.code
+        )
+
+        if config.use_artifacts:
+            if (used_artifacts := self.run.used_artifacts()) is not None:
+                for artifact in used_artifacts:
+                    yield self._make_artifact_record(artifact, use_artifact=True)
+
+        if config.log_artifacts:
+            if (artifacts := self.run.artifacts()) is not None:
+                for artifact in artifacts:
+                    yield self._make_artifact_record(artifact)
+
+        if config.history:
+            yield from self._make_history_records()
+
+        if config.summary:
+            yield self._make_summary_record()
+
+        if config.terminal_output:
+            if (lines := self.run.logs()) is not None:
+                for line in lines:
+                    yield self._make_output_record(line)
+
+    def _make_run_record(self) -> pb.Record:
+        run = pb.RunRecord()
+        run.run_id = self.run.run_id()
+        run.entity = self.run.entity()
+        run.project = self.run.project()
+        run.display_name = coalesce(self.run.display_name())
+        run.notes = coalesce(self.run.notes(), "")
+        run.tags.extend(coalesce(self.run.tags(), []))
+        run.start_time.FromMilliseconds(self.run.start_time())
+
+        host = self.run.host()
+        if host is not None:
+            run.host = host
+
+        runtime = self.run.runtime()
+        if runtime is not None:
+            run.runtime = runtime
+
+        run_group = self.run.run_group()
+        if run_group is not None:
+            run.run_group = run_group
+
+        config = self.run.config()
+        if "_wandb" not in config:
+            config["_wandb"] = {}
+
+        # how do I get this automatically?
+        config["_wandb"]["code_path"] = self.run.code_path()
+        config["_wandb"]["python_version"] = self.run.python_version()
+        config["_wandb"]["cli_version"] = self.run.cli_version()
+
+        self.interface._make_config(
+            data=config,
+            obj=run.config,
+        )  # is there a better way?
+        return self.interface._make_record(run=run)
+
+    def _make_output_record(self, line) -> pb.Record:
+        output_raw = pb.OutputRawRecord()
+        output_raw.output_type = pb.OutputRawRecord.OutputType.STDOUT
+        output_raw.line = line
+        return self.interface._make_record(output_raw=output_raw)
+
+    def _make_summary_record(self) -> pb.Record:
+        d: dict = {
+            **self.run.summary(),
+            "_runtime": self.run.runtime(),  # quirk of runtime -- it has to be here!
+            # '_timestamp': self.run.start_time()/1000,
+        }
+        d = recursive_cast_dictlike_to_dict(d)
+        summary = self.interface._make_summary_from_dict(d)
+        return self.interface._make_record(summary=summary)
+
+    def _make_history_records(self) -> Iterable[pb.Record]:
+        for metrics in self.run.metrics():
+            history = pb.HistoryRecord()
+            for k, v in metrics.items():
+                item = history.item.add()
+                item.key = k
+                # There seems to be some conversion issue to breaks when we try to re-upload.
+                # np.NaN gets converted to float("nan"), which is not expected by our system.
+                # If this cast to string (!) is not done, the row will be dropped.
+                if (isinstance(v, float) and math.isnan(v)) or v == "NaN":
+                    v = np.NaN
+
+                if isinstance(v, bytes):
+                    # it's a json string encoded as bytes
+                    v = v.decode("utf-8")
+                else:
+                    v = json.dumps(v)
+
+                item.value_json = v
+            rec = self.interface._make_record(history=history)
+            yield rec
+
+    def _make_files_record(
+        self, artifacts: bool, files: bool, media: bool, code: bool
+    ) -> pb.Record:
+        run_files = self.run.files()
+        metadata_fname = f"{self.run_dir}/files/wandb-metadata.json"
+        if not files or run_files is None:
+            # We'll always need a metadata file even if there are no other files to upload
+            metadata_fname = self._make_metadata_file()
+            run_files = [(metadata_fname, "end")]
+        files_record = pb.FilesRecord()
+        for path, policy in run_files:
+            if not artifacts and path.startswith("artifact/"):
+                continue
+            if not media and path.startswith("media/"):
+                continue
+            if not code and path.startswith("code/"):
+                continue
+
+            # DirWatcher requires the path to start with media/ instead of the full path
+            if "media" in path:
+                p = Path(path)
+                path = str(p.relative_to(f"{self.run_dir}/files"))
+            f = files_record.files.add()
+            f.path = path
+            f.policy = file_policy_to_enum(policy)
+
+        return self.interface._make_record(files=files_record)
+
+    def _make_artifact_record(
+        self, artifact: Artifact, use_artifact=False
+    ) -> pb.Record:
+        proto = self.interface._make_artifact(artifact)
+        proto.run_id = str(self.run.run_id())
+        proto.project = str(self.run.project())
+        proto.entity = str(self.run.entity())
+        proto.user_created = use_artifact
+        proto.use_after_commit = use_artifact
+        proto.finalize = True
+
+        aliases = artifact._aliases
+        aliases += ["latest", "imported"]
+
+        for alias in aliases:
+            proto.aliases.append(alias)
+        return self.interface._make_record(artifact=proto)
+
+    def _make_telem_record(self) -> pb.Record:
+        telem = telem_pb.TelemetryRecord()
+
+        feature = telem_pb.Feature()
+        feature.importer_mlflow = True
+        telem.feature.CopyFrom(feature)
+
+        cli_version = self.run.cli_version()
+        if cli_version:
+            telem.cli_version = cli_version
+
+        python_version = self.run.python_version()
+        if python_version:
+            telem.python_version = python_version
+
+        return self.interface._make_record(telemetry=telem)
+
+    def _make_metadata_file(self) -> str:
+        missing_text = "This data was not captured"
+        files_dir = f"{self.run_dir}/files"
+        os.makedirs(files_dir, exist_ok=True)
+
+        d = {}
+        d["os"] = coalesce(self.run.os_version(), missing_text)
+        d["python"] = coalesce(self.run.python_version(), missing_text)
+        d["program"] = coalesce(self.run.program(), missing_text)
+        d["cuda"] = coalesce(self.run.cuda_version(), missing_text)
+        d["host"] = coalesce(self.run.host(), missing_text)
+        d["username"] = coalesce(self.run.username(), missing_text)
+        d["executable"] = coalesce(self.run.executable(), missing_text)
+
+        gpus_used = self.run.gpus_used()
+        if gpus_used is not None:
+            d["gpu_devices"] = json.dumps(gpus_used)
+            d["gpu_count"] = json.dumps(len(gpus_used))
+
+        cpus_used = self.run.cpus_used()
+        if cpus_used is not None:
+            d["cpu_count"] = json.dumps(self.run.cpus_used())
+
+        mem_used = self.run.memory_used()
+        if mem_used is not None:
+            d["memory"] = json.dumps({"total": self.run.memory_used()})
+
+        fname = f"{files_dir}/wandb-metadata.json"
+        with open(fname, "w") as f:
+            f.write(json.dumps(d))
+        return fname
+
+
+def _make_settings(
+    root_dir: str, settings_override: Optional[Dict[str, Any]] = None
+) -> SettingsStatic:
+    _settings_override = coalesce(settings_override, {})
+
+    default_settings: Dict[str, Any] = {
+        "x_files_dir": os.path.join(root_dir, "files"),
+        "root_dir": root_dir,
+        "sync_file": os.path.join(root_dir, "txlog.wandb"),
+        "resume": "false",
+        "program": None,
+        "ignore_globs": [],
+        "disable_job_creation": True,
+        "x_start_time": 0,
+        "_offline": None,
+        "x_sync": True,
+        "x_live_policy_rate_limit": 15,  # matches dir_watcher
+        "x_live_policy_wait_time": 600,  # matches dir_watcher
+        "x_file_stream_timeout_seconds": 60,
+    }
+
+    combined_settings = {**default_settings, **_settings_override}
+    settings_message = wandb_settings_pb2.Settings()
+    ParseDict(combined_settings, settings_message)
+
+    return SettingsStatic(settings_message)
+
+
+def send_run(
+    run: ImporterRun,
+    *,
+    extra_arts: Optional[Iterable[Artifact]] = None,
+    extra_used_arts: Optional[Iterable[Artifact]] = None,
+    config: Optional[SendManagerConfig] = None,
+    overrides: Optional[Dict[str, Any]] = None,
+    settings_override: Optional[Dict[str, Any]] = None,
+) -> None:
+    if config is None:
+        config = SendManagerConfig()
+
+    # does this need to be here for pmap?
+    if overrides:
+        for k, v in overrides.items():
+            # `lambda: v` won't work!
+            # https://stackoverflow.com/questions/10802002/why-deepcopy-doesnt-create-new-references-to-lambda-function
+            setattr(run, k, lambda v=v: v)
+
+    rm = RecordMaker(run)
+    root_dir = rm.run_dir
+
+    settings = _make_settings(root_dir, settings_override)
+    sm_record_q = queue.Queue()
+    # wm_record_q = queue.Queue()
+    result_q = queue.Queue()
+    interface = InterfaceQueue(record_q=sm_record_q)
+    context_keeper = context.ContextKeeper()
+    sm = AlternateSendManager(
+        settings, sm_record_q, result_q, interface, context_keeper
+    )
+
+    if extra_arts or extra_used_arts:
+        records = rm.make_artifacts_only_records(extra_arts, extra_used_arts)
+    else:
+        records = rm.make_records(config)
+
+    for r in records:
+        logger.debug(f"Sending {r=}")
+        # In a future update, it might be good to write to a transaction log and have
+        # incremental uploads only send the missing records.
+        # wm.write(r)
+
+        sm.send(r)
+
+    sm.finish()
+    # wm.finish()

lib/python3.12/site-packages/wandb/apis/importers/internals/protocols.py

@@ -0,0 +1,103 @@
+import logging
+from typing import (
+    Any,
+    Dict,
+    Iterable,
+    List,
+    Literal,
+    Optional,
+    Protocol,
+    Tuple,
+    runtime_checkable,
+)
+
+from wandb.sdk.artifacts.artifact import Artifact
+
+logger = logging.getLogger("import_logger")
+
+PathStr = str
+Policy = Literal["now", "end", "live"]
+
+
+@runtime_checkable
+class ImporterRun(Protocol):
+    def run_id(self) -> str: ...  # pragma: no cover
+
+    def entity(self) -> str: ...  # pragma: no cover
+
+    def project(self) -> str: ...  # pragma: no cover
+
+    def config(self) -> Dict[str, Any]: ...  # pragma: no cover
+
+    def summary(self) -> Dict[str, float]: ...  # pragma: no cover
+
+    def metrics(self) -> Iterable[Dict[str, float]]:
+        """Metrics for the run.
+
+        We expect metrics in this shape:
+
+        [
+            {'metric1': 1, 'metric2': 1, '_step': 0},
+            {'metric1': 2, 'metric2': 4, '_step': 1},
+            {'metric1': 3, 'metric2': 9, '_step': 2},
+            ...
+        ]
+
+        You can also submit metrics in this shape:
+        [
+            {'metric1': 1, '_step': 0},
+            {'metric2': 1, '_step': 0},
+            {'metric1': 2, '_step': 1},
+            {'metric2': 4, '_step': 1},
+            ...
+        ]
+        """
+        ...  # pragma: no cover
+
+    def run_group(self) -> Optional[str]: ...  # pragma: no cover
+
+    def job_type(self) -> Optional[str]: ...  # pragma: no cover
+
+    def display_name(self) -> str: ...  # pragma: no cover
+
+    def notes(self) -> Optional[str]: ...  # pragma: no cover
+
+    def tags(self) -> Optional[List[str]]: ...  # pragma: no cover
+
+    def artifacts(self) -> Optional[Iterable[Artifact]]: ...  # pragma: no cover
+
+    def used_artifacts(self) -> Optional[Iterable[Artifact]]: ...  # pragma: no cover
+
+    def os_version(self) -> Optional[str]: ...  # pragma: no cover
+
+    def python_version(self) -> Optional[str]: ...  # pragma: no cover
+
+    def cuda_version(self) -> Optional[str]: ...  # pragma: no cover
+
+    def program(self) -> Optional[str]: ...  # pragma: no cover
+
+    def host(self) -> Optional[str]: ...  # pragma: no cover
+
+    def username(self) -> Optional[str]: ...  # pragma: no cover
+
+    def executable(self) -> Optional[str]: ...  # pragma: no cover
+
+    def gpus_used(self) -> Optional[str]: ...  # pragma: no cover
+
+    def cpus_used(self) -> Optional[int]: ...  # pragma: no cover
+
+    def memory_used(self) -> Optional[int]: ...  # pragma: no cover
+
+    def runtime(self) -> Optional[int]: ...  # pragma: no cover
+
+    def start_time(self) -> Optional[int]: ...  # pragma: no cover
+
+    def code_path(self) -> Optional[str]: ...  # pragma: no cover
+
+    def cli_version(self) -> Optional[str]: ...  # pragma: no cover
+
+    def files(
+        self,
+    ) -> Optional[Iterable[Tuple[PathStr, Policy]]]: ...  # pragma: no cover
+
+    def logs(self) -> Optional[Iterable[str]]: ...  # pragma: no cover

lib/python3.12/site-packages/wandb/apis/importers/internals/util.py

@@ -0,0 +1,78 @@
+import logging
+import sys
+import traceback
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from dataclasses import dataclass
+from typing import Iterable, Optional
+
+
+@dataclass(frozen=True)
+class Namespace:
+    """Configure an alternate entity/project at the dst server your data will end up in."""
+
+    entity: str
+    project: str
+
+    @classmethod
+    def from_path(cls, path: str):
+        entity, project = path.split("/")
+        return cls(entity, project)
+
+    @property
+    def path(self):
+        return f"{self.entity}/{self.project}"
+
+    @property
+    def send_manager_overrides(self):
+        overrides = {}
+        if self.entity:
+            overrides["entity"] = self.entity
+        if self.project:
+            overrides["project"] = self.project
+        return overrides
+
+
+logger = logging.getLogger("import_logger")
+
+
+def parallelize(
+    func,
+    iterable: Iterable,
+    *args,
+    max_workers: Optional[int] = None,
+    raise_on_error: bool = False,
+    **kwargs,
+):
+    def safe_func(*args, **kwargs):
+        try:
+            return func(*args, **kwargs)
+        except Exception as e:
+            _, _, exc_traceback = sys.exc_info()
+            traceback_details = traceback.extract_tb(exc_traceback)
+            filename = traceback_details[-1].filename
+            lineno = traceback_details[-1].lineno
+            logger.debug(
+                f"Exception: {func=} {args=} {kwargs=} {e=} {filename=} {lineno=}. {traceback_details=}"
+            )
+            if raise_on_error:
+                raise
+
+    results = []
+    with ThreadPoolExecutor(max_workers) as exc:
+        futures = {exc.submit(safe_func, x, *args, **kwargs): x for x in iterable}
+        for future in as_completed(futures):
+            results.append(future.result())
+    return results
+
+
+def for_each(
+    func, iterable: Iterable, parallel: bool = True, max_workers: Optional[int] = None
+):
+    if parallel:
+        return parallelize(
+            func,
+            iterable,
+            max_workers=max_workers,
+        )
+
+    return [func(x) for x in iterable]

lib/python3.12/site-packages/wandb/apis/importers/mlflow.py

@@ -0,0 +1,254 @@
+import itertools
+import logging
+import re
+from collections import defaultdict
+from typing import Any, Dict, Iterable, List, Optional, Tuple
+
+import mlflow
+from packaging.version import Version  # type: ignore
+
+import wandb
+from wandb import Artifact
+
+from .internals import internal
+from .internals.util import Namespace, for_each
+
+mlflow_version = Version(mlflow.__version__)
+
+logger = logging.getLogger("import_logger")
+
+
+class MlflowRun:
+    def __init__(self, run, mlflow_client):
+        self.run = run
+        self.mlflow_client: mlflow.MlflowClient = mlflow_client
+
+    def run_id(self) -> str:
+        return self.run.info.run_id
+
+    def entity(self) -> str:
+        return self.run.info.user_id
+
+    def project(self) -> str:
+        return "imported-from-mlflow"
+
+    def config(self) -> Dict[str, Any]:
+        conf = self.run.data.params
+
+        # Add tags here since mlflow supports very long tag names but we only support up to 64 chars
+        tags = {
+            k: v for k, v in self.run.data.tags.items() if not k.startswith("mlflow.")
+        }
+        return {**conf, "imported_mlflow_tags": tags}
+
+    def summary(self) -> Dict[str, float]:
+        return self.run.data.metrics
+
+    def metrics(self) -> Iterable[Dict[str, float]]:
+        d: Dict[int, Dict[str, float]] = defaultdict(dict)
+        for k in self.run.data.metrics.keys():
+            metric = self.mlflow_client.get_metric_history(self.run.info.run_id, k)
+            for item in metric:
+                d[item.step][item.key] = item.value
+
+        for k, v in d.items():
+            yield {"_step": k, **v}
+
+    def run_group(self) -> Optional[str]:
+        # this is nesting?  Parent at `run.info.tags.get("mlflow.parentRunId")`
+        return f"Experiment {self.run.info.experiment_id}"
+
+    def job_type(self) -> Optional[str]:
+        # Is this the right approach?
+        return f"User {self.run.info.user_id}"
+
+    def display_name(self) -> str:
+        if mlflow_version < Version("1.30.0"):
+            return self.run.data.tags["mlflow.runName"]
+        return self.run.info.run_name
+
+    def notes(self) -> Optional[str]:
+        return self.run.data.tags.get("mlflow.note.content")
+
+    def tags(self) -> Optional[List[str]]:
+        ...
+
+        # W&B tags are different than mlflow tags.
+        # The full mlflow tags are added to config under key `imported_mlflow_tags` instead
+
+    def artifacts(self) -> Optional[Iterable[Artifact]]:  # type: ignore
+        if mlflow_version < Version("2.0.0"):
+            dir_path = self.mlflow_client.download_artifacts(
+                run_id=self.run.info.run_id,
+                path="",
+            )
+        else:
+            dir_path = mlflow.artifacts.download_artifacts(run_id=self.run.info.run_id)
+
+        # Since mlflow doesn't have extra metadata about the artifacts,
+        # we just lump them all together into a single wandb.Artifact
+        artifact_name = self._handle_incompatible_strings(self.display_name())
+        art = wandb.Artifact(artifact_name, "imported-artifacts")
+        art.add_dir(dir_path)
+
+        return [art]
+
+    def used_artifacts(self) -> Optional[Iterable[Artifact]]:  # type: ignore
+        ...  # pragma: no cover
+
+    def os_version(self) -> Optional[str]: ...  # pragma: no cover
+
+    def python_version(self) -> Optional[str]: ...  # pragma: no cover
+
+    def cuda_version(self) -> Optional[str]: ...  # pragma: no cover
+
+    def program(self) -> Optional[str]: ...  # pragma: no cover
+
+    def host(self) -> Optional[str]: ...  # pragma: no cover
+
+    def username(self) -> Optional[str]: ...  # pragma: no cover
+
+    def executable(self) -> Optional[str]: ...  # pragma: no cover
+
+    def gpus_used(self) -> Optional[str]: ...  # pragma: no cover
+
+    def cpus_used(self) -> Optional[int]:  # can we get the model?
+        ...  # pragma: no cover
+
+    def memory_used(self) -> Optional[int]: ...  # pragma: no cover
+
+    def runtime(self) -> Optional[int]:
+        end_time = (
+            self.run.info.end_time // 1000
+            if self.run.info.end_time is not None
+            else self.start_time()
+        )
+        return end_time - self.start_time()
+
+    def start_time(self) -> Optional[int]:
+        return self.run.info.start_time // 1000
+
+    def code_path(self) -> Optional[str]: ...  # pragma: no cover
+
+    def cli_version(self) -> Optional[str]: ...  # pragma: no cover
+
+    def files(self) -> Optional[Iterable[Tuple[str, str]]]: ...  # pragma: no cover
+
+    def logs(self) -> Optional[Iterable[str]]: ...  # pragma: no cover
+
+    @staticmethod
+    def _handle_incompatible_strings(s: str) -> str:
+        valid_chars = r"[^a-zA-Z0-9_\-\.]"
+        replacement = "__"
+
+        return re.sub(valid_chars, replacement, s)
+
+
+class MlflowImporter:
+    def __init__(
+        self,
+        dst_base_url: str,
+        dst_api_key: str,
+        mlflow_tracking_uri: str,
+        mlflow_registry_uri: Optional[str] = None,
+        *,
+        custom_api_kwargs: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        self.dst_base_url = dst_base_url
+        self.dst_api_key = dst_api_key
+
+        if custom_api_kwargs is None:
+            custom_api_kwargs = {"timeout": 600}
+
+        self.dst_api = wandb.Api(
+            api_key=dst_api_key,
+            overrides={"base_url": dst_base_url},
+            **custom_api_kwargs,
+        )
+        self.mlflow_tracking_uri = mlflow_tracking_uri
+        mlflow.set_tracking_uri(self.mlflow_tracking_uri)
+
+        if mlflow_registry_uri:
+            mlflow.set_registry_uri(mlflow_registry_uri)
+
+        self.mlflow_client = mlflow.tracking.MlflowClient(mlflow_tracking_uri)
+
+    def __repr__(self):
+        return f"<MlflowImporter src={self.mlflow_tracking_uri}>"
+
+    def collect_runs(self, *, limit: Optional[int] = None) -> Iterable[MlflowRun]:
+        if mlflow_version < Version("1.28.0"):
+            experiments = self.mlflow_client.list_experiments()
+        else:
+            experiments = self.mlflow_client.search_experiments()
+
+        def _runs():
+            for exp in experiments:
+                for run in self.mlflow_client.search_runs(exp.experiment_id):
+                    yield MlflowRun(run, self.mlflow_client)
+
+        runs = itertools.islice(_runs(), limit)
+        yield from runs
+
+    def _import_run(
+        self,
+        run: MlflowRun,
+        *,
+        artifacts: bool = True,
+        namespace: Optional[Namespace] = None,
+        config: Optional[internal.SendManagerConfig] = None,
+    ) -> None:
+        if namespace is None:
+            namespace = Namespace(run.entity(), run.project())
+
+        if config is None:
+            config = internal.SendManagerConfig(
+                metadata=True,
+                files=True,
+                media=True,
+                code=True,
+                history=True,
+                summary=True,
+                terminal_output=True,
+            )
+
+        settings_override = {
+            "api_key": self.dst_api_key,
+            "base_url": self.dst_base_url,
+            "resume": "true",
+            "resumed": True,
+        }
+
+        mlflow.set_tracking_uri(self.mlflow_tracking_uri)
+        internal.send_run(
+            run,
+            overrides=namespace.send_manager_overrides,
+            settings_override=settings_override,
+            config=config,
+        )
+
+        # in mlflow, the artifacts come with the runs, so import them together
+        if artifacts:
+            arts = list(run.artifacts())
+            logger.debug(f"Importing history artifacts, {run=}")
+            internal.send_run(
+                run,
+                extra_arts=arts,
+                overrides=namespace.send_manager_overrides,
+                settings_override=settings_override,
+                config=internal.SendManagerConfig(log_artifacts=True),
+            )
+
+    def import_runs(
+        self,
+        runs: Iterable[MlflowRun],
+        *,
+        artifacts: bool = True,
+        namespace: Optional[Namespace] = None,
+        parallel: bool = True,
+        max_workers: Optional[int] = None,
+    ) -> None:
+        def _import_run_wrapped(run):
+            self._import_run(run, namespace=namespace, artifacts=artifacts)
+
+        for_each(_import_run_wrapped, runs, parallel=parallel, max_workers=max_workers)

lib/python3.12/site-packages/wandb/apis/importers/validation.py

@@ -0,0 +1,108 @@
+import filecmp
+import logging
+import os
+
+import requests
+
+import wandb
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+
+
+def _compare_artifact_manifests(
+    src_art: wandb.Artifact, dst_art: wandb.Artifact
+) -> list:
+    problems = []
+    if isinstance(dst_art, wandb.CommError):
+        return ["commError"]
+
+    if src_art.digest != dst_art.digest:
+        problems.append(f"digest mismatch {src_art.digest=}, {dst_art.digest=}")
+
+    for name, src_entry in src_art.manifest.entries.items():
+        dst_entry = dst_art.manifest.entries.get(name)
+        if dst_entry is None:
+            problems.append(f"missing manifest entry {name=}, {src_entry=}")
+            continue
+
+        for attr in ["path", "digest", "size"]:
+            if getattr(src_entry, attr) != getattr(dst_entry, attr):
+                problems.append(
+                    f"manifest entry mismatch {attr=}, {getattr(src_entry, attr)=}, {getattr(dst_entry, attr)=}"
+                )
+
+    return problems
+
+
+def _compare_artifact_dirs(src_dir, dst_dir) -> list:
+    def compare(src_dir, dst_dir):
+        comparison = filecmp.dircmp(src_dir, dst_dir)
+        differences = {
+            "left_only": comparison.left_only,
+            "right_only": comparison.right_only,
+            "diff_files": comparison.diff_files,
+            "subdir_differences": {},
+        }
+
+        # Recursively find differences in subdirectories
+        for subdir in comparison.subdirs:
+            subdir_src = os.path.join(src_dir, subdir)
+            subdir_dst = os.path.join(dst_dir, subdir)
+            subdir_differences = compare(subdir_src, subdir_dst)
+            # If there are differences, add them to the result
+            if subdir_differences and any(subdir_differences.values()):
+                differences["subdir_differences"][subdir] = subdir_differences
+
+        if all(not diff for diff in differences.values()):
+            return None
+
+        return differences
+
+    return compare(src_dir, dst_dir)
+
+
+def _check_entries_are_downloadable(art):
+    entries = _collect_entries(art)
+    for entry in entries:
+        if not _check_entry_is_downloable(entry):
+            return False
+    return True
+
+
+def _collect_entries(art):
+    has_next_page = True
+    cursor = None
+    entries = []
+    while has_next_page:
+        attrs = art._fetch_file_urls(cursor)
+        has_next_page = attrs["pageInfo"]["hasNextPage"]
+        cursor = attrs["pageInfo"]["endCursor"]
+        for edge in attrs["edges"]:
+            name = edge["node"]["name"]
+            entry = art.get_entry(name)
+            entry._download_url = edge["node"]["directUrl"]
+            entries.append(entry)
+    return entries
+
+
+def _check_entry_is_downloable(entry):
+    url = entry._download_url
+    expected_size = entry.size
+
+    try:
+        resp = requests.head(url, allow_redirects=True)
+    except Exception:
+        logger.exception(f"Problem validating {entry=}")
+        return False
+
+    if resp.status_code != 200:
+        return False
+
+    actual_size = resp.headers.get("content-length", -1)
+    actual_size = int(actual_size)
+
+    if expected_size == actual_size:
+        return True
+
+    return False

lib/python3.12/site-packages/wandb/apis/importers/wandb.py

@@ -0,0 +1,1608 @@
+"""Tooling for the W&B Importer."""
+
+import itertools
+import json
+import logging
+import numbers
+import os
+import re
+import shutil
+from dataclasses import dataclass, field
+from datetime import datetime as dt
+from pathlib import Path
+from typing import Any, Dict, Iterable, Iterator, List, Optional, Tuple
+from unittest.mock import patch
+
+import filelock
+import polars as pl
+import requests
+import urllib3
+import wandb_workspaces.reports.v1 as wr
+import yaml
+from wandb_gql import gql
+from wandb_workspaces.reports.v1 import Report
+
+import wandb
+from wandb.apis.public import ArtifactCollection, Run
+from wandb.apis.public.files import File
+from wandb.util import coalesce, remove_keys_with_none_values
+
+from . import validation
+from .internals import internal
+from .internals.protocols import PathStr, Policy
+from .internals.util import Namespace, for_each
+
+Artifact = wandb.Artifact
+Api = wandb.Api
+Project = wandb.apis.public.Project
+
+ARTIFACT_ERRORS_FNAME = "artifact_errors.jsonl"
+ARTIFACT_SUCCESSES_FNAME = "artifact_successes.jsonl"
+RUN_ERRORS_FNAME = "run_errors.jsonl"
+RUN_SUCCESSES_FNAME = "run_successes.jsonl"
+
+ART_SEQUENCE_DUMMY_PLACEHOLDER = "__ART_SEQUENCE_DUMMY_PLACEHOLDER__"
+RUN_DUMMY_PLACEHOLDER = "__RUN_DUMMY_PLACEHOLDER__"
+ART_DUMMY_PLACEHOLDER_PATH = "__importer_temp__"
+ART_DUMMY_PLACEHOLDER_TYPE = "__temp__"
+
+SRC_ART_PATH = "./artifacts/src"
+DST_ART_PATH = "./artifacts/dst"
+
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+
+if os.getenv("WANDB_IMPORTER_ENABLE_RICH_LOGGING"):
+    from rich.logging import RichHandler
+
+    logger.addHandler(RichHandler(rich_tracebacks=True, tracebacks_show_locals=True))
+else:
+    console_handler = logging.StreamHandler()
+    console_handler.setLevel(logging.INFO)
+
+    formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
+    console_handler.setFormatter(formatter)
+
+    logger.addHandler(console_handler)
+
+
+@dataclass
+class ArtifactSequence:
+    artifacts: Iterable[wandb.Artifact]
+    entity: str
+    project: str
+    type_: str
+    name: str
+
+    def __iter__(self) -> Iterator:
+        return iter(self.artifacts)
+
+    def __repr__(self) -> str:
+        return f"ArtifactSequence({self.identifier})"
+
+    @property
+    def identifier(self) -> str:
+        return "/".join([self.entity, self.project, self.type_, self.name])
+
+    @classmethod
+    def from_collection(cls, collection: ArtifactCollection):
+        arts = collection.artifacts()
+        arts = sorted(arts, key=lambda a: int(a.version.lstrip("v")))
+        return ArtifactSequence(
+            arts,
+            collection.entity,
+            collection.project,
+            collection.type,
+            collection.name,
+        )
+
+
+class WandbRun:
+    def __init__(
+        self,
+        run: Run,
+        *,
+        src_base_url: str,
+        src_api_key: str,
+        dst_base_url: str,
+        dst_api_key: str,
+    ) -> None:
+        self.run = run
+        self.api = wandb.Api(
+            api_key=src_api_key,
+            overrides={"base_url": src_base_url},
+        )
+        self.dst_api = wandb.Api(
+            api_key=dst_api_key,
+            overrides={"base_url": dst_base_url},
+        )
+
+        # For caching
+        self._files: Optional[Iterable[Tuple[str, str]]] = None
+        self._artifacts: Optional[Iterable[Artifact]] = None
+        self._used_artifacts: Optional[Iterable[Artifact]] = None
+        self._parquet_history_paths: Optional[Iterable[str]] = None
+
+    def __repr__(self) -> str:
+        s = os.path.join(self.entity(), self.project(), self.run_id())
+        return f"WandbRun({s})"
+
+    def run_id(self) -> str:
+        return self.run.id
+
+    def entity(self) -> str:
+        return self.run.entity
+
+    def project(self) -> str:
+        return self.run.project
+
+    def config(self) -> Dict[str, Any]:
+        return self.run.config
+
+    def summary(self) -> Dict[str, float]:
+        s = self.run.summary
+        return s
+
+    def metrics(self) -> Iterable[Dict[str, float]]:
+        if self._parquet_history_paths is None:
+            self._parquet_history_paths = list(self._get_parquet_history_paths())
+
+        if self._parquet_history_paths:
+            rows = self._get_rows_from_parquet_history_paths()
+        else:
+            logger.warning(
+                "No parquet files detected; using scan history (this may not be reliable)"
+            )
+            rows = self.run.scan_history()
+
+        for row in rows:
+            row = remove_keys_with_none_values(row)
+            yield row
+
+    def run_group(self) -> Optional[str]:
+        return self.run.group
+
+    def job_type(self) -> Optional[str]:
+        return self.run.job_type
+
+    def display_name(self) -> str:
+        return self.run.display_name
+
+    def notes(self) -> Optional[str]:
+        # Notes includes the previous notes and serves as a catch-all for things we missed or can't add back
+        previous_link = f"Imported from: {self.run.url}"
+        previous_author = f"Author: {self.run.user.username}"
+
+        header = [previous_link, previous_author]
+        previous_notes = self.run.notes or ""
+
+        return "\n".join(header) + "\n---\n" + previous_notes
+
+    def tags(self) -> Optional[List[str]]:
+        return self.run.tags
+
+    def artifacts(self) -> Optional[Iterable[Artifact]]:
+        if self._artifacts is None:
+            _artifacts = []
+            for art in self.run.logged_artifacts():
+                a = _clone_art(art)
+                _artifacts.append(a)
+            self._artifacts = _artifacts
+
+        yield from self._artifacts
+
+    def used_artifacts(self) -> Optional[Iterable[Artifact]]:
+        if self._used_artifacts is None:
+            _used_artifacts = []
+            for art in self.run.used_artifacts():
+                a = _clone_art(art)
+                _used_artifacts.append(a)
+            self._used_artifacts = _used_artifacts
+
+        yield from self._used_artifacts
+
+    def os_version(self) -> Optional[str]: ...  # pragma: no cover
+
+    def python_version(self) -> Optional[str]:
+        return self._metadata_file().get("python")
+
+    def cuda_version(self) -> Optional[str]: ...  # pragma: no cover
+
+    def program(self) -> Optional[str]: ...  # pragma: no cover
+
+    def host(self) -> Optional[str]:
+        return self._metadata_file().get("host")
+
+    def username(self) -> Optional[str]: ...  # pragma: no cover
+
+    def executable(self) -> Optional[str]: ...  # pragma: no cover
+
+    def gpus_used(self) -> Optional[str]: ...  # pragma: no cover
+
+    def cpus_used(self) -> Optional[int]:  # can we get the model?
+        ...  # pragma: no cover
+
+    def memory_used(self) -> Optional[int]: ...  # pragma: no cover
+
+    def runtime(self) -> Optional[int]:
+        wandb_runtime = self.run.summary.get("_wandb", {}).get("runtime")
+        base_runtime = self.run.summary.get("_runtime")
+
+        if (t := coalesce(wandb_runtime, base_runtime)) is None:
+            return t
+        return int(t)
+
+    def start_time(self) -> Optional[int]:
+        t = dt.fromisoformat(self.run.created_at).timestamp() * 1000
+        return int(t)
+
+    def code_path(self) -> Optional[str]:
+        path = self._metadata_file().get("codePath", "")
+        return f"code/{path}"
+
+    def cli_version(self) -> Optional[str]:
+        return self._config_file().get("_wandb", {}).get("value", {}).get("cli_version")
+
+    def files(self) -> Optional[Iterable[Tuple[PathStr, Policy]]]:
+        if self._files is None:
+            files_dir = f"{internal.ROOT_DIR}/{self.run_id()}/files"
+            _files = []
+            for f in self.run.files():
+                f: File
+                # These optimizations are intended to avoid rate limiting when importing many runs in parallel
+                # Don't carry over empty files
+                if f.size == 0:
+                    continue
+                # Skip deadlist to avoid overloading S3
+                if "wandb_manifest.json.deadlist" in f.name:
+                    continue
+
+                result = f.download(files_dir, exist_ok=True, api=self.api)
+                file_and_policy = (result.name, "end")
+                _files.append(file_and_policy)
+            self._files = _files
+
+        yield from self._files
+
+    def logs(self) -> Optional[Iterable[str]]:
+        log_files = self._find_all_in_files_regex(r"^.*output\.log$")
+        for path in log_files:
+            with open(path) as f:
+                yield from f.readlines()
+
+    def _metadata_file(self) -> Dict[str, Any]:
+        if (fname := self._find_in_files("wandb-metadata.json")) is None:
+            return {}
+
+        with open(fname) as f:
+            return json.loads(f.read())
+
+    def _config_file(self) -> Dict[str, Any]:
+        if (fname := self._find_in_files("config.yaml")) is None:
+            return {}
+
+        with open(fname) as f:
+            return yaml.safe_load(f) or {}
+
+    def _get_rows_from_parquet_history_paths(self) -> Iterable[Dict[str, Any]]:
+        # Unfortunately, it's not feasible to validate non-parquet history
+        if not (paths := self._get_parquet_history_paths()):
+            yield {}
+            return
+
+        # Collect and merge parquet history
+        dfs = [
+            pl.read_parquet(p) for path in paths for p in Path(path).glob("*.parquet")
+        ]
+        if "_step" in (df := _merge_dfs(dfs)):
+            df = df.with_columns(pl.col("_step").cast(pl.Int64))
+        yield from df.iter_rows(named=True)
+
+    def _get_parquet_history_paths(self) -> Iterable[str]:
+        if self._parquet_history_paths is None:
+            paths = []
+            # self.artifacts() returns a copy of the artifacts; use this to get raw
+            for art in self.run.logged_artifacts():
+                if art.type != "wandb-history":
+                    continue
+                if (
+                    path := _download_art(art, root=f"{SRC_ART_PATH}/{art.name}")
+                ) is None:
+                    continue
+                paths.append(path)
+            self._parquet_history_paths = paths
+
+        yield from self._parquet_history_paths
+
+    def _find_in_files(self, name: str) -> Optional[str]:
+        if files := self.files():
+            for path, _ in files:
+                if name in path:
+                    return path
+        return None
+
+    def _find_all_in_files_regex(self, regex: str) -> Iterable[str]:
+        if files := self.files():
+            for path, _ in files:
+                if re.match(regex, path):
+                    yield path
+
+
+class WandbImporter:
+    """Transfers runs, reports, and artifact sequences between W&B instances."""
+
+    def __init__(
+        self,
+        src_base_url: str,
+        src_api_key: str,
+        dst_base_url: str,
+        dst_api_key: str,
+        *,
+        custom_api_kwargs: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        self.src_base_url = src_base_url
+        self.src_api_key = src_api_key
+        self.dst_base_url = dst_base_url
+        self.dst_api_key = dst_api_key
+
+        if custom_api_kwargs is None:
+            custom_api_kwargs = {"timeout": 600}
+
+        self.src_api = wandb.Api(
+            api_key=src_api_key,
+            overrides={"base_url": src_base_url},
+            **custom_api_kwargs,
+        )
+        self.dst_api = wandb.Api(
+            api_key=dst_api_key,
+            overrides={"base_url": dst_base_url},
+            **custom_api_kwargs,
+        )
+
+        self.run_api_kwargs = {
+            "src_base_url": src_base_url,
+            "src_api_key": src_api_key,
+            "dst_base_url": dst_base_url,
+            "dst_api_key": dst_api_key,
+        }
+
+    def __repr__(self):
+        return f"<WandbImporter src={self.src_base_url}, dst={self.dst_base_url}>"  # pragma: no cover
+
+    def _import_run(
+        self,
+        run: WandbRun,
+        *,
+        namespace: Optional[Namespace] = None,
+        config: Optional[internal.SendManagerConfig] = None,
+    ) -> None:
+        """Import one WandbRun.
+
+        Use `namespace` to specify alternate settings like where the run should be uploaded
+        """
+        if namespace is None:
+            namespace = Namespace(run.entity(), run.project())
+
+        if config is None:
+            config = internal.SendManagerConfig(
+                metadata=True,
+                files=True,
+                media=True,
+                code=True,
+                history=True,
+                summary=True,
+                terminal_output=True,
+            )
+
+        settings_override = {
+            "api_key": self.dst_api_key,
+            "base_url": self.dst_base_url,
+            "resume": "true",
+            "resumed": True,
+        }
+
+        # Send run with base config
+        logger.debug(f"Importing run, {run=}")
+        internal.send_run(
+            run,
+            overrides=namespace.send_manager_overrides,
+            settings_override=settings_override,
+            config=config,
+        )
+
+        if config.history:
+            # Send run again with history artifacts in case config history=True, artifacts=False
+            # The history artifact must come with the actual history data
+
+            logger.debug(f"Collecting history artifacts, {run=}")
+            history_arts = []
+            for art in run.run.logged_artifacts():
+                if art.type != "wandb-history":
+                    continue
+                logger.debug(f"Collecting history artifact {art.name=}")
+                new_art = _clone_art(art)
+                history_arts.append(new_art)
+
+            logger.debug(f"Importing history artifacts, {run=}")
+            internal.send_run(
+                run,
+                extra_arts=history_arts,
+                overrides=namespace.send_manager_overrides,
+                settings_override=settings_override,
+                config=config,
+            )
+
+    def _delete_collection_in_dst(
+        self,
+        seq: ArtifactSequence,
+        namespace: Optional[Namespace] = None,
+    ):
+        """Deletes the equivalent artifact collection in destination.
+
+        Intended to clear the destination when an uploaded artifact does not pass validation.
+        """
+        entity = coalesce(namespace.entity, seq.entity)
+        project = coalesce(namespace.project, seq.project)
+        art_type = f"{entity}/{project}/{seq.type_}"
+        art_name = seq.name
+
+        logger.info(
+            f"Deleting collection {entity=}, {project=}, {art_type=}, {art_name=}"
+        )
+        try:
+            dst_collection = self.dst_api.artifact_collection(art_type, art_name)
+        except (wandb.CommError, ValueError):
+            logger.warning(f"Collection doesn't exist {art_type=}, {art_name=}")
+            return
+
+        try:
+            dst_collection.delete()
+        except (wandb.CommError, ValueError) as e:
+            logger.warning(
+                f"Collection can't be deleted, {art_type=}, {art_name=}, {e=}"
+            )
+            return
+
+    def _import_artifact_sequence(
+        self,
+        seq: ArtifactSequence,
+        *,
+        namespace: Optional[Namespace] = None,
+    ) -> None:
+        """Import one artifact sequence.
+
+        Use `namespace` to specify alternate settings like where the artifact sequence should be uploaded
+        """
+        if not seq.artifacts:
+            # The artifact sequence has no versions.  This usually means all artifacts versions were deleted intentionally,
+            # but it can also happen if the sequence represents run history and that run was deleted.
+            logger.warning(f"Artifact {seq=} has no artifacts, skipping.")
+            return
+
+        if namespace is None:
+            namespace = Namespace(seq.entity, seq.project)
+
+        settings_override = {
+            "api_key": self.dst_api_key,
+            "base_url": self.dst_base_url,
+            "resume": "true",
+            "resumed": True,
+        }
+
+        send_manager_config = internal.SendManagerConfig(log_artifacts=True)
+
+        # Delete any existing artifact sequence, otherwise versions will be out of order
+        # Unfortunately, you can't delete only part of the sequence because versions are "remembered" even after deletion
+        self._delete_collection_in_dst(seq, namespace)
+
+        # Get a placeholder run for dummy artifacts we'll upload later
+        art = seq.artifacts[0]
+        run_or_dummy: Optional[Run] = _get_run_or_dummy_from_art(art, self.src_api)
+
+        # Each `group_of_artifacts` is either:
+        # 1. A single "real" artifact in a list; or
+        # 2. A list of dummy artifacts that are uploaded together.
+        # This guarantees the real artifacts have the correct version numbers while allowing for parallel upload of dummies.
+        groups_of_artifacts = list(_make_groups_of_artifacts(seq))
+        for i, group in enumerate(groups_of_artifacts, 1):
+            art = group[0]
+            if art.description == ART_SEQUENCE_DUMMY_PLACEHOLDER:
+                run = WandbRun(run_or_dummy, **self.run_api_kwargs)
+            else:
+                try:
+                    wandb_run = art.logged_by()
+                except ValueError:
+                    # The run used to exist but has since been deleted
+                    # wandb_run = None
+                    pass
+
+                # Could be logged by None (rare) or ValueError
+                if wandb_run is None:
+                    logger.warning(
+                        f"Run for {art.name=} does not exist (deleted?), using {run_or_dummy=}"
+                    )
+                    wandb_run = run_or_dummy
+
+                new_art = _clone_art(art)
+                group = [new_art]
+                run = WandbRun(wandb_run, **self.run_api_kwargs)
+
+            logger.info(
+                f"Uploading partial artifact {seq=}, {i}/{len(groups_of_artifacts)}"
+            )
+            internal.send_run(
+                run,
+                extra_arts=group,
+                overrides=namespace.send_manager_overrides,
+                settings_override=settings_override,
+                config=send_manager_config,
+            )
+        logger.info(f"Finished uploading {seq=}")
+
+        # query it back and remove placeholders
+        self._remove_placeholders(seq)
+
+    def _remove_placeholders(self, seq: ArtifactSequence) -> None:
+        try:
+            retry_arts_func = internal.exp_retry(self._dst_api.artifacts)
+            dst_arts = list(retry_arts_func(seq.type_, seq.name))
+        except wandb.CommError:
+            logger.warning(
+                f"{seq=} does not exist in dst.  Has it already been deleted?"
+            )
+            return
+        except TypeError:
+            logger.exception("Problem getting dst versions (try again later).")
+            return
+
+        for art in dst_arts:
+            if art.description != ART_SEQUENCE_DUMMY_PLACEHOLDER:
+                continue
+            if art.type in ("wandb-history", "job"):
+                continue
+
+            try:
+                art.delete(delete_aliases=True)
+            except wandb.CommError as e:
+                if "cannot delete system managed artifact" in str(e):
+                    logger.warning("Cannot delete system managed artifact")
+                else:
+                    raise
+
+    def _get_dst_art(
+        self, src_art: Run, entity: Optional[str] = None, project: Optional[str] = None
+    ) -> Artifact:
+        entity = coalesce(entity, src_art.entity)
+        project = coalesce(project, src_art.project)
+        name = src_art.name
+
+        return self.dst_api._artifact(f"{entity}/{project}/{name}")
+
+    def _get_run_problems(
+        self, src_run: Run, dst_run: Run, force_retry: bool = False
+    ) -> List[dict]:
+        problems = []
+
+        if force_retry:
+            problems.append("__force_retry__")
+
+        if non_matching_metadata := self._compare_run_metadata(src_run, dst_run):
+            problems.append("metadata:" + str(non_matching_metadata))
+
+        if non_matching_summary := self._compare_run_summary(src_run, dst_run):
+            problems.append("summary:" + str(non_matching_summary))
+
+        # TODO: Compare files?
+
+        return problems
+
+    def _compare_run_metadata(self, src_run: Run, dst_run: Run) -> dict:
+        fname = "wandb-metadata.json"
+        # problems = {}
+
+        src_f = src_run.file(fname)
+        if src_f.size == 0:
+            # the src was corrupted so no comparisons here will ever work
+            return {}
+
+        dst_f = dst_run.file(fname)
+        try:
+            contents = wandb.util.download_file_into_memory(
+                dst_f.url, self.dst_api.api_key
+            )
+        except urllib3.exceptions.ReadTimeoutError:
+            return {"Error checking": "Timeout"}
+        except requests.HTTPError as e:
+            if e.response.status_code == 404:
+                return {"Bad upload": f"File not found: {fname}"}
+            return {"http problem": f"{fname}: ({e})"}
+
+        dst_meta = wandb.wandb_sdk.lib.json_util.loads(contents)
+
+        non_matching = {}
+        if src_run.metadata:
+            for k, src_v in src_run.metadata.items():
+                if k not in dst_meta:
+                    non_matching[k] = {"src": src_v, "dst": "KEY NOT FOUND"}
+                    continue
+                dst_v = dst_meta[k]
+                if src_v != dst_v:
+                    non_matching[k] = {"src": src_v, "dst": dst_v}
+
+        return non_matching
+
+    def _compare_run_summary(self, src_run: Run, dst_run: Run) -> dict:
+        non_matching = {}
+        for k, src_v in src_run.summary.items():
+            # These won't match between systems and that's ok
+            if isinstance(src_v, str) and src_v.startswith("wandb-client-artifact://"):
+                continue
+            if k in ("_wandb", "_runtime"):
+                continue
+
+            src_v = _recursive_cast_to_dict(src_v)
+
+            dst_v = dst_run.summary.get(k)
+            dst_v = _recursive_cast_to_dict(dst_v)
+
+            if isinstance(src_v, dict) and isinstance(dst_v, dict):
+                for kk, sv in src_v.items():
+                    # These won't match between systems and that's ok
+                    if isinstance(sv, str) and sv.startswith(
+                        "wandb-client-artifact://"
+                    ):
+                        continue
+                    dv = dst_v.get(kk)
+                    if not _almost_equal(sv, dv):
+                        non_matching[f"{k}-{kk}"] = {"src": sv, "dst": dv}
+            else:
+                if not _almost_equal(src_v, dst_v):
+                    non_matching[k] = {"src": src_v, "dst": dst_v}
+
+        return non_matching
+
+    def _collect_failed_artifact_sequences(self) -> Iterable[ArtifactSequence]:
+        if (df := _read_ndjson(ARTIFACT_ERRORS_FNAME)) is None:
+            logger.debug(f"{ARTIFACT_ERRORS_FNAME=} is empty, returning nothing")
+            return
+
+        unique_failed_sequences = df[
+            ["src_entity", "src_project", "name", "type"]
+        ].unique()
+
+        for row in unique_failed_sequences.iter_rows(named=True):
+            entity = row["src_entity"]
+            project = row["src_project"]
+            name = row["name"]
+            _type = row["type"]
+
+            art_name = f"{entity}/{project}/{name}"
+            arts = self.src_api.artifacts(_type, art_name)
+            arts = sorted(arts, key=lambda a: int(a.version.lstrip("v")))
+            arts = sorted(arts, key=lambda a: a.type)
+
+            yield ArtifactSequence(arts, entity, project, _type, name)
+
+    def _cleanup_dummy_runs(
+        self,
+        *,
+        namespaces: Optional[Iterable[Namespace]] = None,
+        api: Optional[Api] = None,
+        remapping: Optional[Dict[Namespace, Namespace]] = None,
+    ) -> None:
+        api = coalesce(api, self.dst_api)
+        namespaces = coalesce(namespaces, self._all_namespaces())
+
+        for ns in namespaces:
+            if remapping and ns in remapping:
+                ns = remapping[ns]
+
+            logger.debug(f"Cleaning up, {ns=}")
+            try:
+                runs = list(
+                    api.runs(ns.path, filters={"displayName": RUN_DUMMY_PLACEHOLDER})
+                )
+            except ValueError as e:
+                if "Could not find project" in str(e):
+                    logger.exception("Could not find project, does it exist?")
+                    continue
+
+            for run in runs:
+                logger.debug(f"Deleting dummy {run=}")
+                run.delete(delete_artifacts=False)
+
+    def _import_report(
+        self, report: Report, *, namespace: Optional[Namespace] = None
+    ) -> None:
+        """Import one wandb.Report.
+
+        Use `namespace` to specify alternate settings like where the report should be uploaded
+        """
+        if namespace is None:
+            namespace = Namespace(report.entity, report.project)
+
+        entity = coalesce(namespace.entity, report.entity)
+        project = coalesce(namespace.project, report.project)
+        name = report.name
+        title = report.title
+        description = report.description
+
+        api = self.dst_api
+
+        # We shouldn't need to upsert the project for every report
+        logger.debug(f"Upserting {entity=}/{project=}")
+        try:
+            api.create_project(project, entity)
+        except requests.exceptions.HTTPError as e:
+            if e.response.status_code != 409:
+                logger.warning(f"Issue upserting {entity=}/{project=}, {e=}")
+
+        logger.debug(f"Upserting report {entity=}, {project=}, {name=}, {title=}")
+        api.client.execute(
+            wr.report.UPSERT_VIEW,
+            variable_values={
+                "id": None,  # Is there any benefit for this to be the same as default report?
+                "name": name,
+                "entityName": entity,
+                "projectName": project,
+                "description": description,
+                "displayName": title,
+                "type": "runs",
+                "spec": json.dumps(report.spec),
+            },
+        )
+
+    def _use_artifact_sequence(
+        self,
+        sequence: ArtifactSequence,
+        *,
+        namespace: Optional[Namespace] = None,
+    ):
+        if namespace is None:
+            namespace = Namespace(sequence.entity, sequence.project)
+
+        settings_override = {
+            "api_key": self.dst_api_key,
+            "base_url": self.dst_base_url,
+            "resume": "true",
+            "resumed": True,
+        }
+        logger.debug(f"Using artifact sequence with {settings_override=}, {namespace=}")
+
+        send_manager_config = internal.SendManagerConfig(use_artifacts=True)
+
+        for art in sequence:
+            if (used_by := art.used_by()) is None:
+                continue
+
+            for wandb_run in used_by:
+                run = WandbRun(wandb_run, **self.run_api_kwargs)
+
+                internal.send_run(
+                    run,
+                    overrides=namespace.send_manager_overrides,
+                    settings_override=settings_override,
+                    config=send_manager_config,
+                )
+
+    def import_runs(
+        self,
+        *,
+        namespaces: Optional[Iterable[Namespace]] = None,
+        remapping: Optional[Dict[Namespace, Namespace]] = None,
+        parallel: bool = True,
+        incremental: bool = True,
+        max_workers: Optional[int] = None,
+        limit: Optional[int] = None,
+        metadata: bool = True,
+        files: bool = True,
+        media: bool = True,
+        code: bool = True,
+        history: bool = True,
+        summary: bool = True,
+        terminal_output: bool = True,
+    ):
+        logger.info("START: Import runs")
+
+        logger.info("Setting up for import")
+        _create_files_if_not_exists()
+        _clear_fname(RUN_ERRORS_FNAME)
+
+        logger.info("Collecting runs")
+        runs = list(self._collect_runs(namespaces=namespaces, limit=limit))
+
+        logger.info(f"Validating runs, {len(runs)=}")
+        self._validate_runs(
+            runs,
+            skip_previously_validated=incremental,
+            remapping=remapping,
+        )
+
+        logger.info("Collecting failed runs")
+        runs = list(self._collect_failed_runs())
+
+        logger.info(f"Importing runs, {len(runs)=}")
+
+        def _import_run_wrapped(run):
+            namespace = Namespace(run.entity(), run.project())
+            if remapping is not None and namespace in remapping:
+                namespace = remapping[namespace]
+
+            config = internal.SendManagerConfig(
+                metadata=metadata,
+                files=files,
+                media=media,
+                code=code,
+                history=history,
+                summary=summary,
+                terminal_output=terminal_output,
+            )
+
+            logger.debug(f"Importing {run=}, {namespace=}, {config=}")
+            self._import_run(run, namespace=namespace, config=config)
+            logger.debug(f"Finished importing {run=}, {namespace=}, {config=}")
+
+        for_each(_import_run_wrapped, runs, max_workers=max_workers, parallel=parallel)
+        logger.info("END: Importing runs")
+
+    def import_reports(
+        self,
+        *,
+        namespaces: Optional[Iterable[Namespace]] = None,
+        limit: Optional[int] = None,
+        remapping: Optional[Dict[Namespace, Namespace]] = None,
+    ):
+        logger.info("START: Importing reports")
+
+        logger.info("Collecting reports")
+        reports = self._collect_reports(namespaces=namespaces, limit=limit)
+
+        logger.info("Importing reports")
+
+        def _import_report_wrapped(report):
+            namespace = Namespace(report.entity, report.project)
+            if remapping is not None and namespace in remapping:
+                namespace = remapping[namespace]
+
+            logger.debug(f"Importing {report=}, {namespace=}")
+            self._import_report(report, namespace=namespace)
+            logger.debug(f"Finished importing {report=}, {namespace=}")
+
+        for_each(_import_report_wrapped, reports)
+
+        logger.info("END: Importing reports")
+
+    def import_artifact_sequences(
+        self,
+        *,
+        namespaces: Optional[Iterable[Namespace]] = None,
+        incremental: bool = True,
+        max_workers: Optional[int] = None,
+        remapping: Optional[Dict[Namespace, Namespace]] = None,
+    ):
+        """Import all artifact sequences from `namespaces`.
+
+        Note: There is a known bug with the AWS backend where artifacts > 2048MB will fail to upload.  This seems to be related to multipart uploads, but we don't have a fix yet.
+        """
+        logger.info("START: Importing artifact sequences")
+        _clear_fname(ARTIFACT_ERRORS_FNAME)
+
+        logger.info("Collecting artifact sequences")
+        seqs = list(self._collect_artifact_sequences(namespaces=namespaces))
+
+        logger.info("Validating artifact sequences")
+        self._validate_artifact_sequences(
+            seqs,
+            incremental=incremental,
+            remapping=remapping,
+        )
+
+        logger.info("Collecting failed artifact sequences")
+        seqs = list(self._collect_failed_artifact_sequences())
+
+        logger.info(f"Importing artifact sequences, {len(seqs)=}")
+
+        def _import_artifact_sequence_wrapped(seq):
+            namespace = Namespace(seq.entity, seq.project)
+            if remapping is not None and namespace in remapping:
+                namespace = remapping[namespace]
+
+            logger.debug(f"Importing artifact sequence {seq=}, {namespace=}")
+            self._import_artifact_sequence(seq, namespace=namespace)
+            logger.debug(f"Finished importing artifact sequence {seq=}, {namespace=}")
+
+        for_each(_import_artifact_sequence_wrapped, seqs, max_workers=max_workers)
+
+        # it's safer to just use artifact on all seqs to make sure we don't miss anything
+        # For seqs that have already been used, this is a no-op.
+        logger.debug(f"Using artifact sequences, {len(seqs)=}")
+
+        def _use_artifact_sequence_wrapped(seq):
+            namespace = Namespace(seq.entity, seq.project)
+            if remapping is not None and namespace in remapping:
+                namespace = remapping[namespace]
+
+            logger.debug(f"Using artifact sequence {seq=}, {namespace=}")
+            self._use_artifact_sequence(seq, namespace=namespace)
+            logger.debug(f"Finished using artifact sequence {seq=}, {namespace=}")
+
+        for_each(_use_artifact_sequence_wrapped, seqs, max_workers=max_workers)
+
+        # Artifacts whose parent runs have been deleted should have that run deleted in the
+        # destination as well
+
+        logger.info("Cleaning up dummy runs")
+        self._cleanup_dummy_runs(
+            namespaces=namespaces,
+            remapping=remapping,
+        )
+
+        logger.info("END: Importing artifact sequences")
+
+    def import_all(
+        self,
+        *,
+        runs: bool = True,
+        artifacts: bool = True,
+        reports: bool = True,
+        namespaces: Optional[Iterable[Namespace]] = None,
+        incremental: bool = True,
+        remapping: Optional[Dict[Namespace, Namespace]] = None,
+    ):
+        logger.info(f"START: Importing all, {runs=}, {artifacts=}, {reports=}")
+        if runs:
+            self.import_runs(
+                namespaces=namespaces,
+                incremental=incremental,
+                remapping=remapping,
+            )
+
+        if reports:
+            self.import_reports(
+                namespaces=namespaces,
+                remapping=remapping,
+            )
+
+        if artifacts:
+            self.import_artifact_sequences(
+                namespaces=namespaces,
+                incremental=incremental,
+                remapping=remapping,
+            )
+
+        logger.info("END: Importing all")
+
+    def _validate_run(
+        self,
+        src_run: Run,
+        *,
+        remapping: Optional[Dict[Namespace, Namespace]] = None,
+    ) -> None:
+        namespace = Namespace(src_run.entity, src_run.project)
+        if remapping is not None and namespace in remapping:
+            namespace = remapping[namespace]
+
+        dst_entity = namespace.entity
+        dst_project = namespace.project
+        run_id = src_run.id
+
+        try:
+            dst_run = self.dst_api.run(f"{dst_entity}/{dst_project}/{run_id}")
+        except wandb.CommError:
+            problems = [f"run does not exist in dst at {dst_entity=}/{dst_project=}"]
+        else:
+            problems = self._get_run_problems(src_run, dst_run)
+
+        d = {
+            "src_entity": src_run.entity,
+            "src_project": src_run.project,
+            "dst_entity": dst_entity,
+            "dst_project": dst_project,
+            "run_id": run_id,
+        }
+        if problems:
+            d["problems"] = problems
+            fname = RUN_ERRORS_FNAME
+        else:
+            fname = RUN_SUCCESSES_FNAME
+
+        with filelock.FileLock("runs.lock"):
+            with open(fname, "a") as f:
+                f.write(json.dumps(d) + "\n")
+
+    def _filter_previously_checked_runs(
+        self,
+        runs: Iterable[Run],
+        *,
+        remapping: Optional[Dict[Namespace, Namespace]] = None,
+    ) -> Iterable[Run]:
+        if (df := _read_ndjson(RUN_SUCCESSES_FNAME)) is None:
+            logger.debug(f"{RUN_SUCCESSES_FNAME=} is empty, yielding all runs")
+            yield from runs
+            return
+
+        data = []
+        for r in runs:
+            namespace = Namespace(r.entity, r.project)
+            if remapping is not None and namespace in remapping:
+                namespace = remapping[namespace]
+
+            data.append(
+                {
+                    "src_entity": r.entity,
+                    "src_project": r.project,
+                    "dst_entity": namespace.entity,
+                    "dst_project": namespace.project,
+                    "run_id": r.id,
+                    "data": r,
+                }
+            )
+        df2 = pl.DataFrame(data)
+        logger.debug(f"Starting with {len(runs)=} in namespaces")
+
+        results = df2.join(
+            df,
+            how="anti",
+            on=["src_entity", "src_project", "dst_entity", "dst_project", "run_id"],
+        )
+        logger.debug(f"After filtering out already successful runs, {len(results)=}")
+
+        if not results.is_empty():
+            results = results.filter(~results["run_id"].is_null())
+            results = results.unique(
+                ["src_entity", "src_project", "dst_entity", "dst_project", "run_id"]
+            )
+
+        for r in results.iter_rows(named=True):
+            yield r["data"]
+
+    def _validate_artifact(
+        self,
+        src_art: Artifact,
+        dst_entity: str,
+        dst_project: str,
+        download_files_and_compare: bool = False,
+        check_entries_are_downloadable: bool = True,
+    ):
+        problems = []
+
+        # These patterns of artifacts are special and should not be validated
+        ignore_patterns = [
+            r"^job-(.*?)\.py(:v\d+)?$",
+            # r"^run-.*-history(?:\:v\d+)?$$",
+        ]
+        for pattern in ignore_patterns:
+            if re.search(pattern, src_art.name):
+                return (src_art, dst_entity, dst_project, problems)
+
+        try:
+            dst_art = self._get_dst_art(src_art, dst_entity, dst_project)
+        except Exception:
+            problems.append("destination artifact not found")
+            return (src_art, dst_entity, dst_project, problems)
+
+        try:
+            logger.debug("Comparing artifact manifests")
+        except Exception as e:
+            problems.append(
+                f"Problem getting problems! problem with {src_art.entity=}, {src_art.project=}, {src_art.name=} {e=}"
+            )
+        else:
+            problems += validation._compare_artifact_manifests(src_art, dst_art)
+
+        if check_entries_are_downloadable:
+            # validation._check_entries_are_downloadable(src_art)
+            validation._check_entries_are_downloadable(dst_art)
+
+        if download_files_and_compare:
+            logger.debug(f"Downloading {src_art=}")
+            try:
+                src_dir = _download_art(src_art, root=f"{SRC_ART_PATH}/{src_art.name}")
+            except requests.HTTPError as e:
+                problems.append(
+                    f"Invalid download link for src {src_art.entity=}, {src_art.project=}, {src_art.name=}, {e}"
+                )
+
+            logger.debug(f"Downloading {dst_art=}")
+            try:
+                dst_dir = _download_art(dst_art, root=f"{DST_ART_PATH}/{dst_art.name}")
+            except requests.HTTPError as e:
+                problems.append(
+                    f"Invalid download link for dst {dst_art.entity=}, {dst_art.project=}, {dst_art.name=}, {e}"
+                )
+            else:
+                logger.debug(f"Comparing artifact dirs {src_dir=}, {dst_dir=}")
+                if problem := validation._compare_artifact_dirs(src_dir, dst_dir):
+                    problems.append(problem)
+
+        return (src_art, dst_entity, dst_project, problems)
+
+    def _validate_runs(
+        self,
+        runs: Iterable[WandbRun],
+        *,
+        skip_previously_validated: bool = True,
+        remapping: Optional[Dict[Namespace, Namespace]] = None,
+    ):
+        base_runs = [r.run for r in runs]
+        if skip_previously_validated:
+            base_runs = list(
+                self._filter_previously_checked_runs(
+                    base_runs,
+                    remapping=remapping,
+                )
+            )
+
+        def _validate_run(run):
+            logger.debug(f"Validating {run=}")
+            self._validate_run(run, remapping=remapping)
+            logger.debug(f"Finished validating {run=}")
+
+        for_each(_validate_run, base_runs)
+
+    def _collect_failed_runs(self):
+        if (df := _read_ndjson(RUN_ERRORS_FNAME)) is None:
+            logger.debug(f"{RUN_ERRORS_FNAME=} is empty, returning nothing")
+            return
+
+        unique_failed_runs = df[
+            ["src_entity", "src_project", "dst_entity", "dst_project", "run_id"]
+        ].unique()
+
+        for row in unique_failed_runs.iter_rows(named=True):
+            src_entity = row["src_entity"]
+            src_project = row["src_project"]
+            # dst_entity = row["dst_entity"]
+            # dst_project = row["dst_project"]
+            run_id = row["run_id"]
+
+            run = self.src_api.run(f"{src_entity}/{src_project}/{run_id}")
+            yield WandbRun(run, **self.run_api_kwargs)
+
+    def _filter_previously_checked_artifacts(self, seqs: Iterable[ArtifactSequence]):
+        if (df := _read_ndjson(ARTIFACT_SUCCESSES_FNAME)) is None:
+            logger.info(
+                f"{ARTIFACT_SUCCESSES_FNAME=} is empty, yielding all artifact sequences"
+            )
+            for seq in seqs:
+                yield from seq.artifacts
+            return
+
+        for seq in seqs:
+            for art in seq:
+                try:
+                    logged_by = _get_run_or_dummy_from_art(art, self.src_api)
+                except requests.HTTPError:
+                    logger.exception(f"Failed to get run, skipping: {art=}")
+                    continue
+
+                if art.type == "wandb-history" and isinstance(logged_by, _DummyRun):
+                    logger.debug(f"Skipping history artifact {art=}")
+                    # We can never upload valid history for a deleted run, so skip it
+                    continue
+
+                entity = art.entity
+                project = art.project
+                _type = art.type
+                name, ver = _get_art_name_ver(art)
+
+                filtered_df = df.filter(
+                    (df["src_entity"] == entity)
+                    & (df["src_project"] == project)
+                    & (df["name"] == name)
+                    & (df["version"] == ver)
+                    & (df["type"] == _type)
+                )
+
+                # not in file, so not verified yet, don't filter out
+                if len(filtered_df) == 0:
+                    yield art
+
+    def _validate_artifact_sequences(
+        self,
+        seqs: Iterable[ArtifactSequence],
+        *,
+        incremental: bool = True,
+        download_files_and_compare: bool = False,
+        check_entries_are_downloadable: bool = True,
+        remapping: Optional[Dict[Namespace, Namespace]] = None,
+    ):
+        if incremental:
+            logger.info("Validating in incremental mode")
+
+            def filtered_sequences():
+                for seq in seqs:
+                    if not seq.artifacts:
+                        continue
+
+                    art = seq.artifacts[0]
+                    try:
+                        logged_by = _get_run_or_dummy_from_art(art, self.src_api)
+                    except requests.HTTPError:
+                        logger.exception(
+                            f"Validate Artifact http error: {art.entity=},"
+                            f" {art.project=}, {art.name=}"
+                        )
+                        continue
+
+                    if art.type == "wandb-history" and isinstance(logged_by, _DummyRun):
+                        # We can never upload valid history for a deleted run, so skip it
+                        continue
+
+                    yield seq
+
+            artifacts = self._filter_previously_checked_artifacts(filtered_sequences())
+        else:
+            logger.info("Validating in non-incremental mode")
+            artifacts = [art for seq in seqs for art in seq.artifacts]
+
+        def _validate_artifact_wrapped(args):
+            art, entity, project = args
+            if (
+                remapping is not None
+                and (namespace := Namespace(entity, project)) in remapping
+            ):
+                remapped_ns = remapping[namespace]
+                entity = remapped_ns.entity
+                project = remapped_ns.project
+
+            logger.debug(f"Validating {art=}, {entity=}, {project=}")
+            result = self._validate_artifact(
+                art,
+                entity,
+                project,
+                download_files_and_compare=download_files_and_compare,
+                check_entries_are_downloadable=check_entries_are_downloadable,
+            )
+            logger.debug(f"Finished validating {art=}, {entity=}, {project=}")
+            return result
+
+        args = ((art, art.entity, art.project) for art in artifacts)
+        art_problems = for_each(_validate_artifact_wrapped, args)
+        for art, dst_entity, dst_project, problems in art_problems:
+            name, ver = _get_art_name_ver(art)
+            d = {
+                "src_entity": art.entity,
+                "src_project": art.project,
+                "dst_entity": dst_entity,
+                "dst_project": dst_project,
+                "name": name,
+                "version": ver,
+                "type": art.type,
+            }
+
+            if problems:
+                d["problems"] = problems
+                fname = ARTIFACT_ERRORS_FNAME
+            else:
+                fname = ARTIFACT_SUCCESSES_FNAME
+
+            with open(fname, "a") as f:
+                f.write(json.dumps(d) + "\n")
+
+    def _collect_runs(
+        self,
+        *,
+        namespaces: Optional[Iterable[Namespace]] = None,
+        limit: Optional[int] = None,
+        skip_ids: Optional[List[str]] = None,
+        start_date: Optional[str] = None,
+        api: Optional[Api] = None,
+    ) -> Iterable[WandbRun]:
+        api = coalesce(api, self.src_api)
+        namespaces = coalesce(namespaces, self._all_namespaces())
+
+        filters: Dict[str, Any] = {}
+        if skip_ids is not None:
+            filters["name"] = {"$nin": skip_ids}
+        if start_date is not None:
+            filters["createdAt"] = {"$gte": start_date}
+
+        def _runs():
+            for ns in namespaces:
+                logger.debug(f"Collecting runs from {ns=}")
+                for run in api.runs(ns.path, filters=filters):
+                    yield WandbRun(run, **self.run_api_kwargs)
+
+        runs = itertools.islice(_runs(), limit)
+        yield from runs
+
+    def _all_namespaces(
+        self, *, entity: Optional[str] = None, api: Optional[Api] = None
+    ):
+        api = coalesce(api, self.src_api)
+        entity = coalesce(entity, api.default_entity)
+        projects = api.projects(entity)
+        for p in projects:
+            yield Namespace(p.entity, p.name)
+
+    def _collect_reports(
+        self,
+        *,
+        namespaces: Optional[Iterable[Namespace]] = None,
+        limit: Optional[int] = None,
+        api: Optional[Api] = None,
+    ):
+        api = coalesce(api, self.src_api)
+        namespaces = coalesce(namespaces, self._all_namespaces())
+
+        wandb.login(key=self.src_api_key, host=self.src_base_url)
+
+        def reports():
+            for ns in namespaces:
+                for r in api.reports(ns.path):
+                    yield wr.Report.from_url(r.url, api=api)
+
+        yield from itertools.islice(reports(), limit)
+
+    def _collect_artifact_sequences(
+        self,
+        *,
+        namespaces: Optional[Iterable[Namespace]] = None,
+        limit: Optional[int] = None,
+        api: Optional[Api] = None,
+    ):
+        api = coalesce(api, self.src_api)
+        namespaces = coalesce(namespaces, self._all_namespaces())
+
+        def artifact_sequences():
+            for ns in namespaces:
+                logger.debug(f"Collecting artifact sequences from {ns=}")
+                types = []
+                try:
+                    types = [t for t in api.artifact_types(ns.path)]
+                except Exception:
+                    logger.exception("Failed to get artifact types.")
+
+                for t in types:
+                    collections = []
+
+                    # Skip history because it's really for run history
+                    if t.name == "wandb-history":
+                        continue
+
+                    try:
+                        collections = t.collections()
+                    except Exception:
+                        logger.exception("Failed to get artifact collections.")
+
+                    for c in collections:
+                        if c.is_sequence():
+                            yield ArtifactSequence.from_collection(c)
+
+        seqs = itertools.islice(artifact_sequences(), limit)
+        unique_sequences = {seq.identifier: seq for seq in seqs}
+        yield from unique_sequences.values()
+
+
+def _get_art_name_ver(art: Artifact) -> Tuple[str, int]:
+    name, ver = art.name.split(":v")
+    return name, int(ver)
+
+
+def _make_dummy_art(name: str, _type: str, ver: int):
+    art = Artifact(name, ART_DUMMY_PLACEHOLDER_TYPE)
+    art._type = _type
+    art._description = ART_SEQUENCE_DUMMY_PLACEHOLDER
+
+    p = Path(ART_DUMMY_PLACEHOLDER_PATH)
+    p.mkdir(parents=True, exist_ok=True)
+
+    # dummy file with different name to prevent dedupe
+    fname = p / str(ver)
+    with open(fname, "w"):
+        pass
+    art.add_file(fname)
+
+    return art
+
+
+def _make_groups_of_artifacts(seq: ArtifactSequence, start: int = 0):
+    prev_ver = start - 1
+    for art in seq:
+        name, ver = _get_art_name_ver(art)
+
+        # If there's a gap between versions, fill with dummy artifacts
+        if ver - prev_ver > 1:
+            yield [_make_dummy_art(name, art.type, v) for v in range(prev_ver + 1, ver)]
+
+        # Then yield the actual artifact
+        # Must always be a list of one artifact to guarantee ordering
+        yield [art]
+        prev_ver = ver
+
+
+def _recursive_cast_to_dict(obj):
+    if isinstance(obj, list):
+        return [_recursive_cast_to_dict(item) for item in obj]
+    elif isinstance(obj, dict) or hasattr(obj, "items"):
+        new_dict = {}
+        for key, value in obj.items():
+            new_dict[key] = _recursive_cast_to_dict(value)
+        return new_dict
+    else:
+        return obj
+
+
+def _almost_equal(x, y, eps=1e-6):
+    if isinstance(x, dict) and isinstance(y, dict):
+        if x.keys() != y.keys():
+            return False
+        return all(_almost_equal(x[k], y[k], eps) for k in x)
+
+    if isinstance(x, numbers.Number) and isinstance(y, numbers.Number):
+        return abs(x - y) < eps
+
+    if type(x) is not type(y):
+        return False
+
+    return x == y
+
+
+@dataclass
+class _DummyUser:
+    username: str = ""
+
+
+@dataclass
+class _DummyRun:
+    entity: str = ""
+    project: str = ""
+    run_id: str = RUN_DUMMY_PLACEHOLDER
+    id: str = RUN_DUMMY_PLACEHOLDER
+    display_name: str = RUN_DUMMY_PLACEHOLDER
+    notes: str = ""
+    url: str = ""
+    group: str = ""
+    created_at: str = "2000-01-01"
+    user: _DummyUser = field(default_factory=_DummyUser)
+    tags: list = field(default_factory=list)
+    summary: dict = field(default_factory=dict)
+    config: dict = field(default_factory=dict)
+
+    def files(self):
+        return []
+
+
+def _read_ndjson(fname: str) -> Optional[pl.DataFrame]:
+    try:
+        df = pl.read_ndjson(fname)
+    except FileNotFoundError:
+        return None
+    except RuntimeError as e:
+        # No runs previously checked
+        if "empty string is not a valid JSON value" in str(e):
+            return None
+        if "error parsing ndjson" in str(e):
+            return None
+        raise
+
+    return df
+
+
+def _get_run_or_dummy_from_art(art: Artifact, api=None):
+    run = None
+
+    try:
+        run = art.logged_by()
+    except ValueError as e:
+        logger.warning(
+            f"Can't log artifact because run doesn't exist, {art=}, {run=}, {e=}"
+        )
+
+    if run is not None:
+        return run
+
+    query = gql(
+        """
+        query ArtifactCreatedBy(
+            $id: ID!
+        ) {
+            artifact(id: $id) {
+                createdBy {
+                    ... on Run {
+                        name
+                        project {
+                            name
+                            entityName
+                        }
+                    }
+                }
+            }
+        }
+    """
+    )
+    response = api.client.execute(query, variable_values={"id": art.id})
+    creator = response.get("artifact", {}).get("createdBy", {})
+    run = _DummyRun(
+        entity=art.entity,
+        project=art.project,
+        run_id=creator.get("name", RUN_DUMMY_PLACEHOLDER),
+        id=creator.get("name", RUN_DUMMY_PLACEHOLDER),
+    )
+    return run
+
+
+def _clear_fname(fname: str) -> None:
+    old_fname = f"{internal.ROOT_DIR}/{fname}"
+    new_fname = f"{internal.ROOT_DIR}/prev_{fname}"
+
+    logger.debug(f"Moving {old_fname=} to {new_fname=}")
+    try:
+        shutil.copy2(old_fname, new_fname)
+    except FileNotFoundError:
+        # this is just to make a copy of the last iteration, so its ok if the src doesn't exist
+        pass
+
+    with open(fname, "w"):
+        pass
+
+
+def _download_art(art: Artifact, root: str) -> Optional[str]:
+    try:
+        with patch("click.echo"):
+            return art.download(root=root, skip_cache=True)
+    except Exception:
+        logger.exception(f"Error downloading artifact {art=}")
+
+
+def _clone_art(art: Artifact, root: Optional[str] = None):
+    if root is None:
+        # Currently, we would only ever clone a src artifact to move it to dst.
+        root = f"{SRC_ART_PATH}/{art.name}"
+
+    if (path := _download_art(art, root=root)) is None:
+        raise ValueError(f"Problem downloading {art=}")
+
+    name, _ = art.name.split(":v")
+
+    # Hack: skip naming validation check for wandb-* types
+    new_art = Artifact(name, ART_DUMMY_PLACEHOLDER_TYPE)
+    new_art._type = art.type
+    new_art._created_at = art.created_at
+
+    new_art._aliases = art.aliases
+    new_art._description = art.description
+
+    with patch("click.echo"):
+        new_art.add_dir(path)
+
+    return new_art
+
+
+def _create_files_if_not_exists() -> None:
+    fnames = [
+        ARTIFACT_ERRORS_FNAME,
+        ARTIFACT_SUCCESSES_FNAME,
+        RUN_ERRORS_FNAME,
+        RUN_SUCCESSES_FNAME,
+    ]
+
+    for fname in fnames:
+        logger.debug(f"Creating {fname=} if not exists")
+        with open(fname, "a"):
+            pass
+
+
+def _merge_dfs(dfs: List[pl.DataFrame]) -> pl.DataFrame:
+    # Ensure there are DataFrames in the list
+    if len(dfs) == 0:
+        return pl.DataFrame()
+
+    if len(dfs) == 1:
+        return dfs[0]
+
+    merged_df = dfs[0]
+    for df in dfs[1:]:
+        merged_df = merged_df.join(df, how="outer", on=["_step"])
+        col_pairs = [
+            (c, f"{c}_right")
+            for c in merged_df.columns
+            if f"{c}_right" in merged_df.columns
+        ]
+        for col, right in col_pairs:
+            new_col = merged_df[col].fill_null(merged_df[right])
+            merged_df = merged_df.with_columns(new_col).drop(right)
+
+    return merged_df

lib/python3.12/site-packages/wandb/apis/internal.py

@@ -0,0 +1,239 @@
+from typing import Any
+
+from wandb.sdk.internal.internal_api import Api as InternalApi
+
+
+class Api:
+    """Internal proxy to the official internal API."""
+
+    # TODO: Move these methods to PublicApi.
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        self._api_args = args
+        self._api_kwargs = kwargs
+        self._api = None
+
+    def __getstate__(self):
+        """Use for serializing.
+
+        self._api is not serializable, so it's dropped
+        """
+        state = self.__dict__.copy()
+        del state["_api"]
+        return state
+
+    def __setstate__(self, state):
+        """Used for deserializing.
+
+        Don't need to set self._api because it's constructed when needed.
+        """
+        self.__dict__.update(state)
+        self._api = None
+
+    @property
+    def api(self) -> InternalApi:
+        # This is a property in order to delay construction of Internal API
+        # for as long as possible. If constructed in constructor, then the
+        # whole InternalAPI is started when simply importing wandb.
+        if self._api is None:
+            self._api = InternalApi(*self._api_args, **self._api_kwargs)
+        return self._api
+
+    @property
+    def api_key(self):
+        return self.api.api_key
+
+    @property
+    def is_authenticated(self):
+        return self.api.access_token is not None or self.api.api_key is not None
+
+    @property
+    def api_url(self):
+        return self.api.api_url
+
+    @property
+    def app_url(self):
+        return self.api.app_url
+
+    @property
+    def default_entity(self):
+        return self.api.default_entity
+
+    @property
+    def git(self):
+        return self.api.git
+
+    def validate_api_key(self) -> bool:
+        """Returns whether the API key stored on initialization is valid."""
+        return self.api.validate_api_key()
+
+    def file_current(self, *args):
+        return self.api.file_current(*args)
+
+    def download_file(self, *args, **kwargs):
+        return self.api.download_file(*args, **kwargs)
+
+    def download_write_file(self, *args, **kwargs):
+        return self.api.download_write_file(*args, **kwargs)
+
+    def set_current_run_id(self, run_id):
+        return self.api.set_current_run_id(run_id)
+
+    def viewer(self):
+        return self.api.viewer()
+
+    def max_cli_version(self):
+        return self.api.max_cli_version()
+
+    def viewer_server_info(self):
+        return self.api.viewer_server_info()
+
+    def list_projects(self, entity=None):
+        return self.api.list_projects(entity=entity)
+
+    def format_project(self, project):
+        return self.api.format_project(project)
+
+    def upsert_project(self, project, id=None, description=None, entity=None):
+        return self.api.upsert_project(
+            project, id=id, description=description, entity=entity
+        )
+
+    def upsert_run(self, *args, **kwargs):
+        return self.api.upsert_run(*args, **kwargs)
+
+    def settings(self, *args, **kwargs):
+        return self.api.settings(*args, **kwargs)
+
+    def clear_setting(
+        self, key: str, globally: bool = False, persist: bool = False
+    ) -> None:
+        return self.api.clear_setting(key, globally, persist)
+
+    def set_setting(
+        self, key: str, value: Any, globally: bool = False, persist: bool = False
+    ) -> None:
+        return self.api.set_setting(key, value, globally, persist)
+
+    def parse_slug(self, *args, **kwargs):
+        return self.api.parse_slug(*args, **kwargs)
+
+    def download_url(self, *args, **kwargs):
+        return self.api.download_url(*args, **kwargs)
+
+    def download_urls(self, *args, **kwargs):
+        return self.api.download_urls(*args, **kwargs)
+
+    def create_anonymous_api_key(self) -> str:
+        return self.api.create_anonymous_api_key()
+
+    def push(self, *args, **kwargs):
+        return self.api.push(*args, **kwargs)
+
+    def sweep(self, *args, **kwargs):
+        return self.api.sweep(*args, **kwargs)
+
+    def upsert_sweep(self, *args, **kwargs):
+        return self.api.upsert_sweep(*args, **kwargs)
+
+    def set_sweep_state(self, *args, **kwargs):
+        return self.api.set_sweep_state(*args, **kwargs)
+
+    def get_sweep_state(self, *args, **kwargs):
+        return self.api.get_sweep_state(*args, **kwargs)
+
+    def stop_sweep(self, *args, **kwargs):
+        return self.api.stop_sweep(*args, **kwargs)
+
+    def cancel_sweep(self, *args, **kwargs):
+        return self.api.cancel_sweep(*args, **kwargs)
+
+    def pause_sweep(self, *args, **kwargs):
+        return self.api.pause_sweep(*args, **kwargs)
+
+    def resume_sweep(self, *args, **kwargs):
+        return self.api.resume_sweep(*args, **kwargs)
+
+    def register_agent(self, *args, **kwargs):
+        return self.api.register_agent(*args, **kwargs)
+
+    def agent_heartbeat(self, *args, **kwargs):
+        return self.api.agent_heartbeat(*args, **kwargs)
+
+    def use_artifact(self, *args, **kwargs):
+        return self.api.use_artifact(*args, **kwargs)
+
+    def create_artifact(self, *args, **kwargs):
+        return self.api.create_artifact(*args, **kwargs)
+
+    def complete_multipart_upload_artifact(self, *args, **kwargs):
+        return self.api.complete_multipart_upload_artifact(*args, **kwargs)
+
+    def run_config(self, *args, **kwargs):
+        return self.api.run_config(*args, **kwargs)
+
+    def upload_file_retry(self, *args, **kwargs):
+        return self.api.upload_file_retry(*args, **kwargs)
+
+    def upload_multipart_file_chunk_retry(self, *args, **kwargs):
+        return self.api.upload_multipart_file_chunk_retry(*args, **kwargs)
+
+    def get_run_info(self, *args, **kwargs):
+        return self.api.get_run_info(*args, **kwargs)
+
+    def get_run_state(self, *args, **kwargs):
+        return self.api.get_run_state(*args, **kwargs)
+
+    def entity_is_team(self, *args, **kwargs):
+        return self.api.entity_is_team(*args, **kwargs)
+
+    def get_project_run_queues(self, *args, **kwargs):
+        return self.api.get_project_run_queues(*args, **kwargs)
+
+    def push_to_run_queue(self, *args, **kwargs):
+        return self.api.push_to_run_queue(*args, **kwargs)
+
+    def pop_from_run_queue(self, *args, **kwargs):
+        return self.api.pop_from_run_queue(*args, **kwargs)
+
+    def ack_run_queue_item(self, *args, **kwargs):
+        return self.api.ack_run_queue_item(*args, **kwargs)
+
+    def create_launch_agent(self, *args, **kwargs):
+        return self.api.create_launch_agent(*args, **kwargs)
+
+    def create_default_resource_config(self, *args, **kwargs):
+        return self.api.create_default_resource_config(*args, **kwargs)
+
+    def create_run_queue(self, *args, **kwargs):
+        return self.api.create_run_queue(*args, **kwargs)
+
+    def upsert_run_queue(self, *args, **kwargs):
+        return self.api.upsert_run_queue(*args, **kwargs)
+
+    def create_custom_chart(self, *args, **kwargs):
+        return self.api.create_custom_chart(*args, **kwargs)
+
+    def update_launch_agent_status(self, *args, **kwargs):
+        return self.api.update_launch_agent_status(*args, **kwargs)
+
+    def launch_agent_introspection(self, *args, **kwargs):
+        return self.api.launch_agent_introspection(*args, **kwargs)
+
+    def fail_run_queue_item_introspection(self, *args, **kwargs):
+        return self.api.fail_run_queue_item_introspection(*args, **kwargs)
+
+    def fail_run_queue_item(self, *args, **kwargs):
+        return self.api.fail_run_queue_item(*args, **kwargs)
+
+    def update_run_queue_item_warning(self, *args, **kwargs):
+        return self.api.update_run_queue_item_warning(*args, **kwargs)
+
+    def get_launch_agent(self, *args, **kwargs):
+        return self.api.get_launch_agent(*args, **kwargs)
+
+    def stop_run(self, *args, **kwargs):
+        return self.api.stop_run(*args, **kwargs)
+
+
+__all__ = ["Api"]

lib/python3.12/site-packages/wandb/apis/normalize.py

@@ -0,0 +1,81 @@
+"""normalize."""
+
+import ast
+import sys
+from functools import wraps
+from typing import Callable, TypeVar
+
+import requests
+from wandb_gql.client import RetryError
+
+from wandb import env
+from wandb.errors import CommError, Error
+from wandb.util import parse_backend_error_messages
+
+_F = TypeVar("_F", bound=Callable)
+
+
+def normalize_exceptions(func: _F) -> _F:
+    """Function decorator for catching common errors and re-raising as wandb.Error."""
+
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        message = "Whoa, you found a bug."
+        try:
+            return func(*args, **kwargs)
+
+        except requests.HTTPError as error:
+            errors = parse_backend_error_messages(error.response)
+            status = error.response.status_code
+
+            if errors:
+                message = f"HTTP {status}: {'; '.join(errors)}"
+            elif error.response.text:
+                message = f"HTTP {status}: {error.response.text}"
+            elif error.response.reason:
+                # Visually different to distinguish backend errors from
+                # standard HTTP status descriptions.
+                message = f"HTTP {status} ({error.response.reason})"
+            else:
+                message = f"HTTP {status}"
+
+            raise CommError(message, error)
+
+        except RetryError as err:
+            if (
+                "response" in dir(err.last_exception)
+                and err.last_exception.response is not None
+            ):
+                try:
+                    message = err.last_exception.response.json().get(
+                        "errors", [{"message": message}]
+                    )[0]["message"]
+                except ValueError:
+                    message = err.last_exception.response.text
+            else:
+                message = err.last_exception
+
+            if env.is_debug():
+                raise err.last_exception.with_traceback(sys.exc_info()[2])
+            else:
+                raise CommError(message, err.last_exception).with_traceback(
+                    sys.exc_info()[2]
+                )
+        except Error:
+            raise
+        except Exception as err:
+            # gql raises server errors with dict's as strings...
+            if len(err.args) > 0:
+                payload = err.args[0]
+            else:
+                payload = err
+            if str(payload).startswith("{"):
+                message = ast.literal_eval(str(payload))["message"]
+            else:
+                message = str(err)
+            if env.is_debug():
+                raise
+            else:
+                raise CommError(message, err).with_traceback(sys.exc_info()[2])
+
+    return wrapper

lib/python3.12/site-packages/wandb/apis/paginator.py

@@ -0,0 +1,138 @@
+from __future__ import annotations
+
+from abc import abstractmethod
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    ClassVar,
+    Iterator,
+    Mapping,
+    Protocol,
+    Sized,
+    TypeVar,
+    overload,
+)
+
+import wandb
+
+if TYPE_CHECKING:
+    from wandb_graphql.language.ast import Document
+
+T = TypeVar("T")
+
+
+# Structural type hint for the client instance
+class _Client(Protocol):
+    def execute(self, *args: Any, **kwargs: Any) -> dict[str, Any]: ...
+
+
+class Paginator(Iterator[T]):
+    """An iterator for paginated objects from GraphQL requests."""
+
+    QUERY: ClassVar[Document | None] = None
+
+    def __init__(
+        self,
+        client: _Client,
+        variables: Mapping[str, Any],
+        per_page: int = 50,  # We don't allow unbounded paging
+    ):
+        self.client: _Client = client
+
+        # shallow copy partly guards against mutating the original input
+        self.variables: dict[str, Any] = dict(variables)
+
+        self.per_page: int = per_page
+        self.objects: list[T] = []
+        self.index: int = -1
+        self.last_response: object | None = None
+
+    def __iter__(self) -> Iterator[T]:
+        self.index = -1
+        return self
+
+    @property
+    @abstractmethod
+    def more(self) -> bool:
+        """Whether there are more pages to be fetched."""
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def cursor(self) -> str | None:
+        """The start cursor to use for the next fetched page."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def convert_objects(self) -> list[T]:
+        """Convert the last fetched response data into the iterated objects."""
+        raise NotImplementedError
+
+    def update_variables(self) -> None:
+        """Update the query variables for the next page fetch."""
+        self.variables.update({"perPage": self.per_page, "cursor": self.cursor})
+
+    def _update_response(self) -> None:
+        """Fetch and store the response data for the next page."""
+        self.last_response = self.client.execute(
+            self.QUERY, variable_values=self.variables
+        )
+
+    def _load_page(self) -> bool:
+        """Fetch the next page, if any, returning True and storing the response if there was one."""
+        if not self.more:
+            return False
+        self.update_variables()
+        self._update_response()
+        self.objects.extend(self.convert_objects())
+        return True
+
+    @overload
+    def __getitem__(self, index: int) -> T: ...
+    @overload
+    def __getitem__(self, index: slice) -> list[T]: ...
+
+    def __getitem__(self, index: int | slice) -> T | list[T]:
+        loaded = True
+        stop = index.stop if isinstance(index, slice) else index
+        while loaded and stop > len(self.objects) - 1:
+            loaded = self._load_page()
+        return self.objects[index]
+
+    def __next__(self) -> T:
+        self.index += 1
+        if len(self.objects) <= self.index:
+            if not self._load_page():
+                raise StopIteration
+            if len(self.objects) <= self.index:
+                raise StopIteration
+        return self.objects[self.index]
+
+    next = __next__
+
+
+class SizedPaginator(Paginator[T], Sized):
+    """A Paginator for objects with a known total count."""
+
+    @property
+    def length(self) -> int | None:
+        wandb.termwarn(
+            (
+                "`.length` is deprecated and will be removed in a future version. "
+                "Use `len(...)` instead."
+            ),
+            repeat=False,
+        )
+        return len(self)
+
+    def __len__(self) -> int:
+        if self._length is None:
+            self._load_page()
+        if self._length is None:
+            raise ValueError("Object doesn't provide length")
+        return self._length
+
+    @property
+    @abstractmethod
+    def _length(self) -> int | None:
+        raise NotImplementedError

lib/python3.12/site-packages/wandb/apis/public/__init__.py

@@ -0,0 +1,35 @@
+from wandb.apis.public.api import Api, RetryingClient, requests
+from wandb.apis.public.artifacts import (
+    ArtifactCollection,
+    ArtifactCollections,
+    ArtifactFiles,
+    Artifacts,
+    ArtifactType,
+    ArtifactTypes,
+    RunArtifacts,
+)
+from wandb.apis.public.automations import Automations
+from wandb.apis.public.files import FILE_FRAGMENT, File, Files
+from wandb.apis.public.history import HistoryScan, SampledHistoryScan
+from wandb.apis.public.integrations import SlackIntegrations, WebhookIntegrations
+from wandb.apis.public.jobs import (
+    Job,
+    QueuedRun,
+    RunQueue,
+    RunQueueAccessType,
+    RunQueuePrioritizationMode,
+    RunQueueResourceType,
+)
+from wandb.apis.public.projects import PROJECT_FRAGMENT, Project, Projects
+from wandb.apis.public.query_generator import QueryGenerator
+from wandb.apis.public.registries.registry import Registry
+from wandb.apis.public.reports import (
+    BetaReport,
+    PanelMetricsHelper,
+    PythonMongoishQueryGenerator,
+    Reports,
+)
+from wandb.apis.public.runs import RUN_FRAGMENT, Run, Runs
+from wandb.apis.public.sweeps import SWEEP_FRAGMENT, Sweep
+from wandb.apis.public.teams import Member, Team
+from wandb.apis.public.users import User