memory.artifacts

memory.artifacts

DVC-backed artifact helpers for reproducible experiment files.

This module provides the Phase 2 foundation for large/frequently updated files to live in DVC while Git keeps code, manifests, memory entries, and lightweight pointers. It intentionally does not import DVC at module import time and does not require AWS credentials unless a caller executes push/pull against S3.

CommandResult dataclass

Structured result from a DVC or Git command.

Source code in memory/artifacts.py

@dataclass(frozen=True)
class CommandResult:
    """Structured result from a DVC or Git command."""

    args: tuple[str, ...]
    cwd: Path
    returncode: int
    stdout: str = ""
    stderr: str = ""
    dry_run: bool = False

    @property
    def ok(self) -> bool:
        return self.returncode == 0

S3RemoteConfig dataclass

S3 settings needed for real DVC push/pull later.

Local initialization and tests can omit this entirely. Real remote use will need bucket/prefix/region plus IAM or other S3-compatible authentication.

Source code in memory/artifacts.py

@dataclass(frozen=True)
class S3RemoteConfig:
    """S3 settings needed for real DVC push/pull later.

    Local initialization and tests can omit this entirely. Real remote use will
    need bucket/prefix/region plus IAM or other S3-compatible authentication.
    """

    bucket: str
    prefix: str = ""
    region: str | None = None
    remote_name: str = DEFAULT_REMOTE_NAME
    endpoint_url: str | None = None

    @property
    def url(self) -> str:
        bucket = self.bucket.strip()
        if not bucket:
            raise ValueError("S3 bucket is required to configure a DVC remote.")
        prefix = self.prefix.strip().strip("/")
        return f"s3://{bucket}/{prefix}" if prefix else f"s3://{bucket}"

find_git_root

find_git_root(cwd: Path | str | None = None) -> Path

Return the Git worktree root for cwd.

DVC operations must run in the target repository, not the package install directory. This helper shells out to Git instead of walking for .git so worktrees and nested Git metadata files are handled correctly.

Source code in memory/artifacts.py

def find_git_root(cwd: Path | str | None = None) -> Path:
    """Return the Git worktree root for ``cwd``.

    DVC operations must run in the target repository, not the package install
    directory. This helper shells out to Git instead of walking for ``.git`` so
    worktrees and nested Git metadata files are handled correctly.
    """

    start = Path(cwd or Path.cwd()).resolve()
    result = _run(("git", "rev-parse", "--show-toplevel"), cwd=start, check=False)
    if result.returncode != 0:
        message = result.stderr.strip() or result.stdout.strip() or f"{start} is not inside a Git repository."
        raise RuntimeError(message)
    return Path(result.stdout.strip()).resolve()

resolve_repo_path

resolve_repo_path(path: Path | str, repo_root: Path | str | None = None) -> Path

Resolve path and reject values outside repo_root.

Source code in memory/artifacts.py

def resolve_repo_path(path: Path | str, repo_root: Path | str | None = None) -> Path:
    """Resolve ``path`` and reject values outside ``repo_root``."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    candidate = Path(path)
    if not candidate.is_absolute():
        candidate = root / candidate
    resolved = candidate.resolve(strict=False)
    try:
        resolved.relative_to(root)
    except ValueError as exc:
        raise ValueError(f"Artifact path must stay inside the repository: {path}") from exc
    return resolved

repo_relative_path

repo_relative_path(path: Path | str, repo_root: Path | str | None = None) -> str

Return a normalized POSIX-style path relative to the repository root.

Source code in memory/artifacts.py

def repo_relative_path(path: Path | str, repo_root: Path | str | None = None) -> str:
    """Return a normalized POSIX-style path relative to the repository root."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    resolved = resolve_repo_path(path, root)
    return resolved.relative_to(root).as_posix()

artifact_manifest_path

artifact_manifest_path(artifact_id: str, repo_root: Path | str | None = None) -> Path

Return the conventional manifest path for an artifact/run id.

Source code in memory/artifacts.py

def artifact_manifest_path(artifact_id: str, repo_root: Path | str | None = None) -> Path:
    """Return the conventional manifest path for an artifact/run id."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    safe_id = _safe_artifact_id(artifact_id)
    return root / MANIFESTS_DIR / f"{safe_id}.yaml"

list_artifacts

list_artifacts(repo_root: Path | str | None = None) -> list[dict[str, Any]]

List DVC pointer files and their declared outputs.

Source code in memory/artifacts.py

def list_artifacts(repo_root: Path | str | None = None) -> list[dict[str, Any]]:
    """List DVC pointer files and their declared outputs."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    artifacts: list[dict[str, Any]] = []
    for pointer in sorted(root.rglob("*.dvc")):
        if _is_internal_path(pointer, root):
            continue
        rel_pointer = pointer.relative_to(root).as_posix()
        try:
            data = yaml.safe_load(pointer.read_text(encoding="utf-8")) or {}
        except (OSError, yaml.YAMLError):
            data = {}
        outs = data.get("outs") or []
        normalized_outs = []
        for out in outs:
            if isinstance(out, Mapping):
                path = str(out.get("path", ""))
                normalized_outs.append({k: v for k, v in out.items() if k != "path"} | {"path": path})
            else:
                normalized_outs.append({"path": str(out)})
        artifacts.append({
            "pointer": rel_pointer,
            "outs": normalized_outs,
            "paths": [out.get("path", "") for out in normalized_outs if out.get("path")],
        })
    return artifacts

resolve_artifact

resolve_artifact(identifier: str, repo_root: Path | str | None = None) -> dict[str, Any]

Resolve an artifact id/path to a manifest, DVC pointer, or repo path.

Source code in memory/artifacts.py

def resolve_artifact(identifier: str, repo_root: Path | str | None = None) -> dict[str, Any]:
    """Resolve an artifact id/path to a manifest, DVC pointer, or repo path."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    if not identifier.strip():
        raise ValueError("Artifact identifier is required.")

    manifest_path = artifact_manifest_path(identifier, root)
    if manifest_path.exists():
        manifest = read_manifest(identifier, root)
        return {
            "kind": "manifest",
            "id": manifest.get("id", identifier),
            "manifest_path": manifest_path.relative_to(root).as_posix(),
            "manifest": manifest,
        }

    candidate = resolve_repo_path(identifier, root)
    pointer = candidate if candidate.suffix == ".dvc" else Path(f"{candidate}.dvc")
    if pointer.exists():
        rel_pointer = pointer.relative_to(root).as_posix()
        return {
            "kind": "dvc_pointer",
            "path": candidate.relative_to(root).as_posix(),
            "pointer": rel_pointer,
            "metadata": _read_dvc_pointer(pointer),
        }
    if candidate.exists():
        return {
            "kind": "path",
            "path": candidate.relative_to(root).as_posix(),
            "exists": True,
            "tracked_by_dvc": False,
        }
    raise FileNotFoundError(f"No artifact manifest, DVC pointer, or repo path found for: {identifier}")

verify_artifacts

verify_artifacts(paths: Path | str | Iterable[Path | str] | None = None, repo_root: Path | str | None = None, *, dry_run: bool = False) -> dict[str, Any]

Check local DVC status and basic pointer/path presence for artifacts.

Source code in memory/artifacts.py

def verify_artifacts(
    paths: Path | str | Iterable[Path | str] | None = None,
    repo_root: Path | str | None = None,
    *,
    dry_run: bool = False,
) -> dict[str, Any]:
    """Check local DVC status and basic pointer/path presence for artifacts."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    rel_paths = _normalize_paths(paths, root) if paths is not None else []
    status = dvc_status(root, paths=rel_paths or None, dry_run=dry_run)
    checked = []
    for rel_path in rel_paths:
        path = root / rel_path
        pointer = path if path.suffix == ".dvc" else Path(f"{path}.dvc")
        exists = path.exists()
        pointer_exists = pointer.exists()
        checked.append({
            "path": rel_path,
            "exists": exists,
            "pointer": pointer.relative_to(root).as_posix() if pointer_exists else "",
            "pointer_exists": pointer_exists,
            "ok": exists or pointer_exists,
        })
    return {
        "ok": status.ok and all(item["ok"] for item in checked),
        "status": command_result_to_dict(status),
        "checked": checked,
    }

build_experiment_manifest

build_experiment_manifest(experiment_id: str, command: str, artifacts: Sequence[Mapping[str, Any]] | Sequence[str] | None = None, repo_root: Path | str | None = None, *, cwd: Path | str | None = None, status: str = 'planned', result_code: int | None = None, started_at: str | None = None, ended_at: str | None = None, include_dirty_diff: bool = False, dirty_note: str = '', environment: Mapping[str, Any] | None = None, artifact_state: str = ARTIFACT_STATE_LOCAL_ONLY, remote: Mapping[str, Any] | None = None, remote_error: str = '') -> dict[str, Any]

Build a reproducibility manifest without writing it to disk.

Source code in memory/artifacts.py

def build_experiment_manifest(
    experiment_id: str,
    command: str,
    artifacts: Sequence[Mapping[str, Any]] | Sequence[str] | None = None,
    repo_root: Path | str | None = None,
    *,
    cwd: Path | str | None = None,
    status: str = "planned",
    result_code: int | None = None,
    started_at: str | None = None,
    ended_at: str | None = None,
    include_dirty_diff: bool = False,
    dirty_note: str = "",
    environment: Mapping[str, Any] | None = None,
    artifact_state: str = ARTIFACT_STATE_LOCAL_ONLY,
    remote: Mapping[str, Any] | None = None,
    remote_error: str = "",
) -> dict[str, Any]:
    """Build a reproducibility manifest without writing it to disk."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root(cwd)
    working_dir = resolve_repo_path(cwd or root, root)
    normalized_artifacts = _normalize_artifact_records(artifacts or (), root)
    dirty_status = git_state.snapshot(root).to_dict()
    now = _utc_now()
    return {
        "schema_version": 1,
        "id": _safe_artifact_id(experiment_id),
        "created_at": now,
        "started_at": started_at or now,
        "ended_at": ended_at,
        "status": status,
        "result_code": result_code,
        "command": command,
        "cwd": working_dir.relative_to(root).as_posix(),
        "git_root": str(root),
        "git_commit": _git_output(("rev-parse", "HEAD"), root),
        "dirty": bool(dirty_status.get("dirty", False)),
        "dirty_status": dirty_status,
        "dirty_diff": _git_dirty_summary(root) if include_dirty_diff else "",
        "dirty_note": dirty_note,
        # environment/remote can carry caller-supplied values (e.g. os.environ or
        # an S3 remote dict with access keys). The manifest is committed and
        # pushed, so scrub secret-keyed values. These fields are informational --
        # reruns use command/git_commit/cwd/artifacts, never these -- so redaction
        # is safe. `command` is intentionally NOT redacted (it is executed
        # verbatim on rerun); callers must avoid inlining secrets there.
        "environment": _redact_secrets(dict(environment or _environment_summary())),
        "artifacts": normalized_artifacts,
        "artifact_state": artifact_state,
        "remote": _redact_secrets(dict(remote or {})),
        "remote_error": remote_error,
    }

write_manifest

write_manifest(manifest: Mapping[str, Any], repo_root: Path | str | None = None, *, dry_run: bool = False) -> dict[str, Any]

Write an experiment manifest under .memory/artifacts/manifests.

Source code in memory/artifacts.py

def write_manifest(
    manifest: Mapping[str, Any],
    repo_root: Path | str | None = None,
    *,
    dry_run: bool = False,
) -> dict[str, Any]:
    """Write an experiment manifest under ``.memory/artifacts/manifests``."""

    manifest_id = str(manifest.get("id", "")).strip()
    if not manifest_id:
        raise ValueError("Manifest must include a non-empty id.")
    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    path = artifact_manifest_path(manifest_id, root)
    if dry_run:
        return {
            "id": _safe_artifact_id(manifest_id),
            "path": path.relative_to(root).as_posix(),
            "dry_run": True,
            "manifest": dict(manifest),
        }
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(yaml.dump(dict(manifest), default_flow_style=False, sort_keys=False), encoding="utf-8")
    return {"id": _safe_artifact_id(manifest_id), "path": path.relative_to(root).as_posix(), "dry_run": False}

read_manifest

read_manifest(experiment_id: str, repo_root: Path | str | None = None) -> dict[str, Any]

Read an experiment manifest by id.

Source code in memory/artifacts.py

def read_manifest(experiment_id: str, repo_root: Path | str | None = None) -> dict[str, Any]:
    """Read an experiment manifest by id."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    path = artifact_manifest_path(experiment_id, root)
    if not path.exists():
        raise FileNotFoundError(f"Manifest not found: {path.relative_to(root).as_posix()}")
    try:
        data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    except yaml.YAMLError as exc:
        # Fail with a clear, path-anchored message instead of a raw parser error
        # so a corrupt manifest is diagnosable rather than cryptic.
        raise ValueError(
            f"Manifest is corrupt (invalid YAML): {path.relative_to(root).as_posix()}: {exc}"
        ) from exc
    if not isinstance(data, dict):
        raise ValueError(f"Manifest is not a mapping: {path.relative_to(root).as_posix()}")
    return data

list_manifests

list_manifests(repo_root: Path | str | None = None) -> list[dict[str, Any]]

List stored experiment manifests with lightweight metadata.

Source code in memory/artifacts.py

def list_manifests(repo_root: Path | str | None = None) -> list[dict[str, Any]]:
    """List stored experiment manifests with lightweight metadata."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    manifests_dir = root / MANIFESTS_DIR
    if not manifests_dir.exists():
        return []
    manifests: list[dict[str, Any]] = []
    for path in sorted(manifests_dir.glob("*.yaml")):
        try:
            data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
        except (OSError, yaml.YAMLError):
            data = {}
        manifests.append({
            "id": str(data.get("id") or path.stem),
            "path": path.relative_to(root).as_posix(),
            "command": data.get("command", ""),
            "status": data.get("status", ""),
            "artifact_state": data.get("artifact_state", ARTIFACT_STATE_LOCAL_ONLY),
            "remote": data.get("remote", {}),
            "remote_error": data.get("remote_error", ""),
            "git_commit": data.get("git_commit", ""),
            "dirty": bool(data.get("dirty", False)),
            "dirty_status": data.get("dirty_status", {}),
            "artifacts": data.get("artifacts", []),
            "created_at": data.get("created_at", ""),
            "experiment": str(data.get("experiment") or ""),
            "rerun_of": str(data.get("rerun_of") or ""),
            "replication": str(data.get("replication") or ""),
            "executor": str(data.get("executor") or ""),
            "run_stamp": str(data.get("run_stamp") or ""),
        })
    return manifests

run_experiment

run_experiment(experiment_id: str, command: str, artifacts: Sequence[Mapping[str, Any]] | Sequence[str] | None = None, repo_root: Path | str | None = None, *, cwd: Path | str | None = None, execute: bool = False, include_dirty_diff: bool = False, dirty_note: str = '', environment: Mapping[str, Any] | None = None, dry_run: bool = True, require_remote: bool = False) -> dict[str, Any]

Create a manifest for an experiment, optionally executing the command.

Source code in memory/artifacts.py

def run_experiment(
    experiment_id: str,
    command: str,
    artifacts: Sequence[Mapping[str, Any]] | Sequence[str] | None = None,
    repo_root: Path | str | None = None,
    *,
    cwd: Path | str | None = None,
    execute: bool = False,
    include_dirty_diff: bool = False,
    dirty_note: str = "",
    environment: Mapping[str, Any] | None = None,
    dry_run: bool = True,
    require_remote: bool = False,
) -> dict[str, Any]:
    """Create a manifest for an experiment, optionally executing the command."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root(cwd)
    working_dir = resolve_repo_path(cwd or root, root)
    started_at = _utc_now()
    result: CommandResult | None = None
    status = "planned"
    ended_at: str | None = None
    if execute and not dry_run:
        completed = _execute_command(command, cwd=working_dir)
        result = CommandResult((command,), working_dir, completed.returncode, completed.stdout, completed.stderr)
        status = "success" if completed.returncode == 0 else "failed"
        ended_at = _utc_now()
    elif dry_run:
        status = "dry_run"

    artifact_state = ARTIFACT_STATE_QUEUED if require_remote else ARTIFACT_STATE_LOCAL_ONLY

    manifest = build_experiment_manifest(
        experiment_id,
        command,
        artifacts,
        root,
        cwd=working_dir,
        status=status,
        result_code=result.returncode if result else None,
        started_at=started_at,
        ended_at=ended_at,
        include_dirty_diff=include_dirty_diff,
        dirty_note=dirty_note,
        environment=environment,
        artifact_state=artifact_state,
        remote=dvc_remote_summary(root, dry_run=True) if require_remote else {},
    )
    write_result = write_manifest(manifest, root, dry_run=False)
    push_result: dict[str, Any] | None = None
    if require_remote and not dry_run:
        try:
            push_result = push_manifest_artifacts(str(manifest["id"]), root, dry_run=False)
            manifest = push_result["manifest"]
        except Exception as exc:
            manifest = read_manifest(str(manifest["id"]), root)
            manifest["status"] = "failed"
            manifest["artifact_state"] = ARTIFACT_STATE_FAILED
            manifest["remote_error"] = str(exc)
            write_manifest(manifest, root, dry_run=False)
            raise
    return {
        "manifest": manifest,
        "manifest_path": write_result["path"],
        "executed": bool(result),
        "dry_run": dry_run,
        "require_remote": require_remote,
        "push": push_result,
        "result": command_result_to_dict(result) if result else None,
    }

list_pipeline_stages

list_pipeline_stages(repo_root: Path | str | None = None) -> list[dict[str, Any]]

Return normalized stage definitions from dvc.yaml.

Source code in memory/artifacts.py

def list_pipeline_stages(repo_root: Path | str | None = None) -> list[dict[str, Any]]:
    """Return normalized stage definitions from ``dvc.yaml``."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    pipeline_path = root / DVC_PIPELINE_FILE
    if not pipeline_path.exists():
        return []
    data = yaml.safe_load(pipeline_path.read_text(encoding="utf-8")) or {}
    stages = data.get("stages") if isinstance(data, Mapping) else {}
    if not isinstance(stages, Mapping):
        return []
    normalized: list[dict[str, Any]] = []
    for name, stage in sorted(stages.items()):
        if not isinstance(stage, Mapping):
            stage = {}
        normalized.append({
            "name": str(name),
            "cmd": str(stage.get("cmd", "")),
            "deps": _normalize_stage_values(stage.get("deps")),
            "outs": _normalize_stage_values(stage.get("outs")),
            "params": _normalize_stage_values(stage.get("params")),
            "metrics": _normalize_stage_values(stage.get("metrics")),
            "plots": _normalize_stage_values(stage.get("plots")),
            "wdir": str(stage.get("wdir", "")),
        })
    return normalized

dvc_stage_add

dvc_stage_add(name: str, command: str, repo_root: Path | str | None = None, *, deps: Path | str | Iterable[Path | str] | None = None, outs: Path | str | Iterable[Path | str] | None = None, params: str | Iterable[str] | None = None, metrics: Path | str | Iterable[Path | str] | None = None, plots: Path | str | Iterable[Path | str] | None = None, wdir: Path | str | None = None, force: bool = False, dry_run: bool = False) -> CommandResult

Add or update a DVC pipeline stage.

Source code in memory/artifacts.py

def dvc_stage_add(
    name: str,
    command: str,
    repo_root: Path | str | None = None,
    *,
    deps: Path | str | Iterable[Path | str] | None = None,
    outs: Path | str | Iterable[Path | str] | None = None,
    params: str | Iterable[str] | None = None,
    metrics: Path | str | Iterable[Path | str] | None = None,
    plots: Path | str | Iterable[Path | str] | None = None,
    wdir: Path | str | None = None,
    force: bool = False,
    dry_run: bool = False,
) -> CommandResult:
    """Add or update a DVC pipeline stage."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    stage_name = name.strip()
    if not stage_name:
        raise ValueError("DVC stage name is required.")
    stage_command = command.strip()
    if not stage_command:
        raise ValueError("DVC stage command is required.")

    args = ["stage", "add", "-n", stage_name]
    if force:
        args.append("--force")
    if wdir:
        args.extend(["--wdir", repo_relative_path(wdir, root)])
    for dep in _optional_repo_paths(deps, root):
        args.extend(["-d", dep])
    for out in _optional_repo_paths(outs, root):
        args.extend(["-o", out])
    for param in _optional_strings(params):
        args.extend(["-p", param])
    for metric in _optional_repo_paths(metrics, root):
        args.extend(["--metrics", metric])
    for plot in _optional_repo_paths(plots, root):
        args.extend(["--plots", plot])
    args.append(stage_command)
    return _run_dvc(args, root, dry_run=dry_run)

dvc_repro

dvc_repro(repo_root: Path | str | None = None, *, targets: str | Iterable[str] | None = None, dry_run: bool = False) -> CommandResult

Run DVC pipeline reproduction for all stages or selected targets.

Source code in memory/artifacts.py

def dvc_repro(
    repo_root: Path | str | None = None,
    *,
    targets: str | Iterable[str] | None = None,
    dry_run: bool = False,
) -> CommandResult:
    """Run DVC pipeline reproduction for all stages or selected targets."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    args = ["repro"]
    args.extend(_optional_strings(targets))
    return _run_dvc(args, root, dry_run=dry_run)

run_pipeline

run_pipeline(pipeline_id: str, repo_root: Path | str | None = None, *, targets: str | Iterable[str] | None = None, dry_run: bool = True, require_remote: bool = False, environment: Mapping[str, Any] | None = None) -> dict[str, Any]

Run or plan a DVC pipeline and write a reproducibility manifest.

Source code in memory/artifacts.py

def run_pipeline(
    pipeline_id: str,
    repo_root: Path | str | None = None,
    *,
    targets: str | Iterable[str] | None = None,
    dry_run: bool = True,
    require_remote: bool = False,
    environment: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    """Run or plan a DVC pipeline and write a reproducibility manifest."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    normalized_targets = _optional_strings(targets)
    command_parts = ["dvc", "repro", *normalized_targets]
    repro_result = dvc_repro(root, targets=normalized_targets, dry_run=dry_run)
    output_paths = _pipeline_output_paths(root, normalized_targets)
    status = "dry_run" if dry_run else ("success" if repro_result.ok else "failed")
    artifact_state = ARTIFACT_STATE_QUEUED if require_remote else ARTIFACT_STATE_LOCAL_ONLY
    manifest = build_experiment_manifest(
        pipeline_id,
        " ".join(command_parts),
        artifacts=output_paths,
        repo_root=root,
        status=status,
        result_code=repro_result.returncode if not dry_run else None,
        started_at=_utc_now(),
        ended_at="" if dry_run else _utc_now(),
        environment=environment,
        artifact_state=artifact_state,
        remote=dvc_remote_summary(root, dry_run=True) if require_remote else {},
    )
    manifest["kind"] = "pipeline"
    manifest["pipeline"] = {
        "targets": normalized_targets,
        "stages": list_pipeline_stages(root),
    }
    write_result = write_manifest(manifest, root, dry_run=False)
    push_result: dict[str, Any] | None = None
    if require_remote and not dry_run:
        try:
            push_result = push_manifest_artifacts(str(manifest["id"]), root, dry_run=False)
            manifest = push_result["manifest"]
        except Exception as exc:
            manifest = read_manifest(str(manifest["id"]), root)
            manifest["status"] = "failed"
            manifest["artifact_state"] = ARTIFACT_STATE_FAILED
            manifest["remote_error"] = str(exc)
            write_manifest(manifest, root)
            raise
    return {
        "manifest": manifest,
        "manifest_path": write_result["path"],
        "targets": normalized_targets,
        "dry_run": dry_run,
        "require_remote": require_remote,
        "repro": command_result_to_dict(repro_result),
        "push": push_result,
    }

reproduce_experiment

reproduce_experiment(experiment_id: str, repo_root: Path | str | None = None, *, pull: bool = False, execute: bool = False, dry_run: bool = True) -> dict[str, Any]

Prepare or run reproduction steps from a stored experiment manifest.

Source code in memory/artifacts.py

def reproduce_experiment(
    experiment_id: str,
    repo_root: Path | str | None = None,
    *,
    pull: bool = False,
    execute: bool = False,
    dry_run: bool = True,
) -> dict[str, Any]:
    """Prepare or run reproduction steps from a stored experiment manifest."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    manifest = read_manifest(experiment_id, root)
    cwd = resolve_repo_path(str(manifest.get("cwd") or "."), root)
    artifact_paths = [a.get("path", "") for a in manifest.get("artifacts", []) if isinstance(a, Mapping) and a.get("path")]
    pull_result = dvc_pull(root, targets=artifact_paths, dry_run=dry_run) if pull and artifact_paths else None
    command = str(manifest.get("command", ""))
    run_result: CommandResult | None = None
    if execute and not dry_run:
        completed = _execute_command(command, cwd=cwd)
        run_result = CommandResult((command,), cwd, completed.returncode, completed.stdout, completed.stderr)
    return {
        "id": manifest.get("id", experiment_id),
        "manifest": manifest,
        "planned_command": command,
        "cwd": repo_relative_path(cwd, root),
        "pull": command_result_to_dict(pull_result) if pull_result else None,
        "executed": bool(run_result),
        "dry_run": dry_run,
        "result": command_result_to_dict(run_result) if run_result else None,
    }

ensure_dvc_repo

ensure_dvc_repo(repo_root: Path | str | None = None, *, dry_run: bool = False) -> CommandResult

Initialize DVC metadata in the target Git repository when missing.

Source code in memory/artifacts.py

def ensure_dvc_repo(
    repo_root: Path | str | None = None,
    *,
    dry_run: bool = False,
) -> CommandResult:
    """Initialize DVC metadata in the target Git repository when missing."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    if (root / ".dvc").exists():
        return CommandResult(("dvc", "init"), root, 0, stdout="DVC already initialized.\n", dry_run=dry_run)
    return _run_dvc(("init",), root, dry_run=dry_run)

configure_s3_remote

configure_s3_remote(config: S3RemoteConfig, repo_root: Path | str | None = None, *, force: bool = False, dry_run: bool = False) -> list[CommandResult]

Configure the repository's DVC remote for S3 artifact storage.

This writes only DVC config when executed. It does not contact AWS and does not validate credentials; later push/pull calls perform remote I/O.

Source code in memory/artifacts.py

def configure_s3_remote(
    config: S3RemoteConfig,
    repo_root: Path | str | None = None,
    *,
    force: bool = False,
    dry_run: bool = False,
) -> list[CommandResult]:
    """Configure the repository's DVC remote for S3 artifact storage.

    This writes only DVC config when executed. It does not contact AWS and does
    not validate credentials; later push/pull calls perform remote I/O.
    """

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    add_args = ["remote", "add"]
    if force:
        add_args.append("--force")
    add_args.extend(["-d", config.remote_name, config.url])

    results = [_run_dvc(add_args, root, dry_run=dry_run)]
    if config.region:
        results.append(_run_dvc(("remote", "modify", config.remote_name, "region", config.region), root, dry_run=dry_run))
    if config.endpoint_url:
        results.append(
            _run_dvc(("remote", "modify", config.remote_name, "endpointurl", config.endpoint_url), root, dry_run=dry_run)
        )
    return results

dvc_add

dvc_add(paths: Path | str | Iterable[Path | str], repo_root: Path | str | None = None, *, dry_run: bool = False) -> CommandResult

Track one or more repository-local paths with DVC.

Source code in memory/artifacts.py

def dvc_add(
    paths: Path | str | Iterable[Path | str],
    repo_root: Path | str | None = None,
    *,
    dry_run: bool = False,
) -> CommandResult:
    """Track one or more repository-local paths with DVC."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    rel_paths = _normalize_paths(paths, root)
    return _run_dvc(("add", *rel_paths), root, dry_run=dry_run)

dvc_status

dvc_status(repo_root: Path | str | None = None, *, paths: Path | str | Iterable[Path | str] | None = None, dry_run: bool = False) -> CommandResult

Return DVC status for the repository or selected paths.

Source code in memory/artifacts.py

def dvc_status(
    repo_root: Path | str | None = None,
    *,
    paths: Path | str | Iterable[Path | str] | None = None,
    dry_run: bool = False,
) -> CommandResult:
    """Return DVC status for the repository or selected paths."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    args = ["status"]
    if paths is not None:
        args.extend(_normalize_paths(paths, root))
    return _run_dvc(args, root, dry_run=dry_run)

dvc_push

dvc_push(repo_root: Path | str | None = None, *, targets: Path | str | Iterable[Path | str] | None = None, dry_run: bool = False) -> CommandResult

Push DVC-tracked artifacts to the configured remote.

Source code in memory/artifacts.py

def dvc_push(
    repo_root: Path | str | None = None,
    *,
    targets: Path | str | Iterable[Path | str] | None = None,
    dry_run: bool = False,
) -> CommandResult:
    """Push DVC-tracked artifacts to the configured remote."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    _ensure_remote_for_io(root, dry_run=dry_run)
    args = ["push"]
    if targets is not None:
        args.extend(_normalize_paths(targets, root))
    return _run_dvc(args, root, dry_run=dry_run)

dvc_pull

dvc_pull(repo_root: Path | str | None = None, *, targets: Path | str | Iterable[Path | str] | None = None, dry_run: bool = False) -> CommandResult

Pull DVC-tracked artifacts from the configured remote.

Source code in memory/artifacts.py

def dvc_pull(
    repo_root: Path | str | None = None,
    *,
    targets: Path | str | Iterable[Path | str] | None = None,
    dry_run: bool = False,
) -> CommandResult:
    """Pull DVC-tracked artifacts from the configured remote."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    _ensure_remote_for_io(root, dry_run=dry_run)
    args = ["pull"]
    if targets is not None:
        args.extend(_normalize_paths(targets, root))
    return _run_dvc(args, root, dry_run=dry_run)

dvc_remote_summary

dvc_remote_summary(repo_root: Path | str | None = None, *, dry_run: bool = False) -> dict[str, Any]

Return lightweight configured DVC remote metadata.

Source code in memory/artifacts.py

def dvc_remote_summary(repo_root: Path | str | None = None, *, dry_run: bool = False) -> dict[str, Any]:
    """Return lightweight configured DVC remote metadata."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    if dry_run:
        return {
            "name": DEFAULT_REMOTE_NAME,
            "url": "",
            "verified_at": "",
            "dry_run": True,
        }
    result = _run_dvc(("remote", "list"), root, check=False)
    remotes = []
    for line in result.stdout.splitlines():
        parts = line.split(None, 1)
        if len(parts) == 2:
            remotes.append({"name": parts[0], "url": parts[1]})
    default_result = _run_dvc(("config", "core.remote"), root, check=False)
    default_name = default_result.stdout.strip() if default_result.returncode == 0 else ""
    selected = next((remote for remote in remotes if remote["name"] == default_name), None)
    if selected is None and remotes:
        selected = remotes[0]
    remote_name = selected["name"] if selected else default_name
    remote_url = selected["url"] if selected else ""
    if remote_name and not remote_url:
        url_result = _run_dvc(("config", f"remote.{remote_name}.url"), root, check=False)
        remote_url = url_result.stdout.strip() if url_result.returncode == 0 else ""
    profile = ""
    region = ""
    if remote_name:
        profile_result = _run_dvc(("config", f"remote.{remote_name}.profile"), root, check=False)
        profile = profile_result.stdout.strip() if profile_result.returncode == 0 else ""
        region_result = _run_dvc(("config", f"remote.{remote_name}.region"), root, check=False)
        region = region_result.stdout.strip() if region_result.returncode == 0 else ""
    return {
        "name": remote_name,
        "url": remote_url,
        "profile": profile,
        "region": region,
        "verified_at": "",
        "dry_run": False,
    }

push_manifest_artifacts

push_manifest_artifacts(experiment_id: str, repo_root: Path | str | None = None, *, dry_run: bool = False) -> dict[str, Any]

Push all DVC artifacts referenced by a manifest and update remote state.

Source code in memory/artifacts.py

def push_manifest_artifacts(
    experiment_id: str,
    repo_root: Path | str | None = None,
    *,
    dry_run: bool = False,
) -> dict[str, Any]:
    """Push all DVC artifacts referenced by a manifest and update remote state."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    manifest = read_manifest(experiment_id, root)
    artifact_paths = _artifact_paths_from_manifest(manifest)
    if not artifact_paths:
        remote = dvc_remote_summary(root, dry_run=dry_run)
        remote["verified_at"] = "" if dry_run else _utc_now()
        manifest["artifact_state"] = ARTIFACT_STATE_REMOTE_VERIFIED
        manifest["remote"] = remote
        manifest["remote_error"] = ""
        if not dry_run:
            write_manifest(manifest, root)
        return {
            "id": manifest.get("id", experiment_id),
            "artifact_state": manifest["artifact_state"],
            "pushed": False,
            "push": None,
            "manifest": manifest,
        }

    try:
        push_result = dvc_push(root, targets=artifact_paths, dry_run=dry_run)
    except Exception as exc:
        manifest["artifact_state"] = ARTIFACT_STATE_FAILED
        manifest["remote"] = dvc_remote_summary(root, dry_run=True)
        manifest["remote_error"] = str(exc)
        if not dry_run:
            write_manifest(manifest, root)
        raise

    remote = dvc_remote_summary(root, dry_run=dry_run)
    remote["verified_at"] = "" if dry_run else _utc_now()
    manifest["artifact_state"] = ARTIFACT_STATE_QUEUED if dry_run else ARTIFACT_STATE_REMOTE_VERIFIED
    manifest["remote"] = remote
    manifest["remote_error"] = ""
    if not dry_run:
        write_manifest(manifest, root)
    return {
        "id": manifest.get("id", experiment_id),
        "artifact_state": manifest["artifact_state"],
        "pushed": not dry_run,
        "push": command_result_to_dict(push_result),
        "manifest": manifest,
    }

verify_manifest

verify_manifest(experiment_id: str, repo_root: Path | str | None = None, *, dry_run: bool = False) -> dict[str, Any]

Verify local pointer/path presence for all artifacts in a manifest.

Source code in memory/artifacts.py

def verify_manifest(
    experiment_id: str,
    repo_root: Path | str | None = None,
    *,
    dry_run: bool = False,
) -> dict[str, Any]:
    """Verify local pointer/path presence for all artifacts in a manifest."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    manifest = read_manifest(experiment_id, root)
    artifact_paths = _artifact_paths_from_manifest(manifest)
    verification = verify_artifacts(artifact_paths or None, root, dry_run=dry_run)
    return {
        "id": manifest.get("id", experiment_id),
        "artifact_state": manifest.get("artifact_state", ARTIFACT_STATE_LOCAL_ONLY),
        "remote": manifest.get("remote", {}),
        "remote_error": manifest.get("remote_error", ""),
        "ok": verification["ok"],
        "verification": verification,
        "manifest_path": artifact_manifest_path(str(manifest.get("id", experiment_id)), root).relative_to(root).as_posix(),
    }

resolve_rerun_target

resolve_rerun_target(experiment_id: str, repo_root: Path | str | None = None) -> dict[str, Any]

Resolve the manifest a rerun should actually execute.

Cards carry the latest run's id, which may be a failed run. Failed runs have no usable provenance to replay, so walk the rerun_of chain back to the nearest successful run (or the family root).

Source code in memory/artifacts.py

def resolve_rerun_target(experiment_id: str, repo_root: Path | str | None = None) -> dict[str, Any]:
    """Resolve the manifest a rerun should actually execute.

    Cards carry the latest run's id, which may be a failed run. Failed runs
    have no usable provenance to replay, so walk the ``rerun_of`` chain back
    to the nearest successful run (or the family root).
    """

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    manifest = read_manifest(experiment_id, root)
    seen: set[str] = {str(manifest.get("id") or experiment_id)}
    while str(manifest.get("status") or "") != "success" and str(manifest.get("rerun_of") or ""):
        parent_id = str(manifest["rerun_of"])
        if parent_id in seen:
            break
        seen.add(parent_id)
        try:
            manifest = read_manifest(parent_id, root)
        except FileNotFoundError:
            break
    return manifest

rerun_experiment

rerun_experiment(experiment_id: str, repo_root: Path | str | None = None, *, executor: str = 'local', run_stamp: str = '', dry_run: bool = True, require_remote: bool = True, keep_sandbox: bool = False, log_fn: Any = None) -> dict[str, Any]

Rerun an experiment in a git-worktree sandbox at its original commit.

Outputs are copied to a per-run folder (.memory/artifacts/experiments/<family>/runs/<stamp>/), DVC-pushed, and recorded in a new dated run manifest carrying rerun_of, executor, and a replication check against the original output hashes.

Source code in memory/artifacts.py

def rerun_experiment(
    experiment_id: str,
    repo_root: Path | str | None = None,
    *,
    executor: str = "local",
    run_stamp: str = "",
    dry_run: bool = True,
    require_remote: bool = True,
    keep_sandbox: bool = False,
    log_fn: Any = None,
) -> dict[str, Any]:
    """Rerun an experiment in a git-worktree sandbox at its original commit.

    Outputs are copied to a per-run folder
    (``.memory/artifacts/experiments/<family>/runs/<stamp>/``), DVC-pushed,
    and recorded in a new dated run manifest carrying ``rerun_of``,
    ``executor``, and a replication check against the original output hashes.
    """

    def _log(message: str) -> None:
        if log_fn is not None:
            log_fn(message)

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    original = resolve_rerun_target(experiment_id, root)
    original_id = str(original.get("id") or experiment_id)
    if original_id != experiment_id:
        _log(f"Requested run {experiment_id} is not rerunnable; resolved to {original_id}")
    family = _safe_artifact_id(str(original.get("experiment") or original_id))
    commit = str(original.get("git_commit") or "").strip()
    command = str(original.get("command") or "").strip()
    if not commit:
        raise ValueError(f"Manifest {original_id} has no git_commit; cannot sandbox-rerun.")
    if not command:
        raise ValueError(f"Manifest {original_id} has no command; cannot rerun.")

    # Suffix with a short random token: multiple runners (or rapid clicks)
    # within the same second must never collide on sandbox/run paths.
    stamp = run_stamp or (
        datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S") + "-" + uuid.uuid4().hex[:4]
    )
    run_id = _safe_artifact_id(f"{family}-run-{stamp}")
    sandbox = root / SANDBOXES_DIR / run_id
    run_dir_rel = (EXPERIMENT_RUNS_DIR / family / "runs" / stamp).as_posix()

    input_artifacts = [
        artifact for artifact in original.get("artifacts", [])
        if isinstance(artifact, Mapping) and "input" in str(artifact.get("role", "")).lower()
    ]
    output_artifacts = [
        artifact for artifact in original.get("artifacts", [])
        if isinstance(artifact, Mapping) and artifact.get("path")
        and "input" not in str(artifact.get("role", "")).lower()
    ]

    plan = {
        "run_id": run_id,
        "experiment": family,
        "rerun_of": original_id,
        "git_commit": commit,
        "command": command,
        "sandbox": sandbox.relative_to(root).as_posix(),
        "run_dir": run_dir_rel,
        "inputs": [str(a.get("path")) for a in input_artifacts],
        "outputs": [str(a.get("path")) for a in output_artifacts],
        "executor": executor,
    }
    if dry_run:
        return {"dry_run": True, "plan": plan}

    if sandbox.exists():
        # Stamps are unique, so an existing dir is debris from a crashed run.
        _log(f"Removing stale sandbox debris at {plan['sandbox']}")
        _remove_sandbox(root, sandbox)
    sandbox.parent.mkdir(parents=True, exist_ok=True)

    replication_caveat = ""
    if original.get("dirty"):
        replication_caveat = (
            "Original run was dirty (uncommitted changes); the sandbox reruns the "
            "committed code, so outputs may legitimately differ."
        )

    started_at = _utc_now()
    try:
        _log(f"Creating sandbox worktree at {plan['sandbox']} ({commit[:12]})")
        _run(("git", "worktree", "add", "--detach", str(sandbox), commit), cwd=root)

        # Code files referenced by the command may have been committed after
        # the run (same lag as the DVC pointers). Copy missing ones from the
        # working tree so the rerun can proceed, but flag the provenance gap.
        artifact_path_set = {str(a.get("path") or "").replace("\\", "/") for a in original.get("artifacts", [])}
        copied_code = _copy_missing_command_files(command, root, sandbox, exclude=artifact_path_set)
        if copied_code:
            _log(
                f"Command file(s) missing at {commit[:12]}: {', '.join(copied_code)}; "
                "copied current versions from the working tree"
            )
            note = (
                f"Command file(s) {', '.join(copied_code)} were not present at the recorded "
                "commit; the rerun used the current working-tree versions."
            )
            replication_caveat = f"{replication_caveat} {note}".strip()

        _prepare_sandbox_dvc(root, sandbox)

        # Manifests capture HEAD at run time, but pointer files are usually
        # committed right after the run — so the sandbox commit often lacks
        # them. Reconstruct missing pointers from the manifest's recorded
        # hashes (preferred) or the current working tree.
        pull_targets: list[str] = []
        for artifact in input_artifacts:
            rel = str(artifact.get("path") or "")
            if not rel:
                continue
            pointer_rel = str(artifact.get("pointer") or f"{rel}.dvc")
            if not (sandbox / pointer_rel).exists():
                restored = _restore_sandbox_pointer(root, sandbox, artifact, rel, pointer_rel)
                if restored:
                    _log(f"Pointer {pointer_rel} missing at {commit[:12]}; reconstructed {restored}")
                else:
                    _log(f"WARNING: no pointer or hash available for input {rel}; skipping pull")
                    continue
            pull_targets.append(rel)
        if pull_targets:
            _log(f"Pulling {len(pull_targets)} input(s) from DVC remote")
            _run(("dvc", "pull", *pull_targets), cwd=sandbox)

        work_cwd = sandbox / str(original.get("cwd") or ".")
        _log(f"Running: {command}")
        completed = _execute_command(command, cwd=work_cwd)
        result = CommandResult((command,), work_cwd, completed.returncode, completed.stdout, completed.stderr)
        if result.stdout:
            _log(result.stdout[-4000:])
        if result.stderr:
            _log(result.stderr[-4000:])
        ended_at = _utc_now()
        status = "success" if result.ok else "failed"

        run_dir = root / EXPERIMENT_RUNS_DIR / family / "runs" / stamp
        new_artifacts: list[dict[str, Any]] = []
        replication_checks: list[dict[str, Any]] = []
        if result.ok:
            run_dir.mkdir(parents=True, exist_ok=True)
            for artifact in output_artifacts:
                rel = str(artifact.get("path"))
                produced = sandbox / rel
                if not produced.exists():
                    replication_checks.append({"path": rel, "status": "missing"})
                    continue
                destination = run_dir / Path(rel).name
                shutil.copy2(produced, destination)
                new_rel = destination.relative_to(root).as_posix()
                record: dict[str, Any] = {"path": new_rel, "role": str(artifact.get("role") or "output")}
                original_md5 = _pointer_md5_at_commit(root, commit, str(artifact.get("pointer") or f"{rel}.dvc"))
                new_md5 = _file_md5(destination)
                if original_md5:
                    match = original_md5 == new_md5
                    replication_checks.append({
                        "path": rel,
                        "status": "exact" if match else "divergent",
                        "original_md5": original_md5,
                        "new_md5": new_md5,
                    })
                else:
                    replication_checks.append({"path": rel, "status": "unknown", "new_md5": new_md5})
                new_artifacts.append(record)

        replication = "failed"
        if result.ok and replication_checks:
            statuses = {check["status"] for check in replication_checks}
            if statuses <= {"exact"}:
                replication = "exact"
            elif "unknown" in statuses and "divergent" not in statuses and "missing" not in statuses:
                replication = "unknown"
            else:
                replication = "divergent"
        elif result.ok:
            replication = "unknown"

        push_error = ""
        artifact_state = ARTIFACT_STATE_LOCAL_ONLY
        if new_artifacts and require_remote:
            try:
                _log("Tracking and pushing run outputs via DVC")
                paths = [record["path"] for record in new_artifacts]
                dvc_add(paths, root)
                dvc_push(root, targets=paths)
                artifact_state = ARTIFACT_STATE_REMOTE_VERIFIED
            except Exception as exc:
                push_error = str(exc)
                artifact_state = ARTIFACT_STATE_FAILED
        new_artifacts = _normalize_artifact_records(new_artifacts, root)

        # Carry input provenance forward so each run manifest is itself
        # rerunnable (inputs are restored by content hash, not by run dir).
        carried_inputs = [
            {
                key: artifact[key]
                for key in ("path", "role", "pointer", "md5", "size")
                if artifact.get(key) is not None
            }
            for artifact in input_artifacts
        ]

        manifest = build_experiment_manifest(
            run_id,
            command,
            carried_inputs + new_artifacts,
            root,
            cwd=root / str(original.get("cwd") or "."),
            status=status,
            result_code=result.returncode,
            started_at=started_at,
            ended_at=ended_at,
            artifact_state=artifact_state,
            remote=dvc_remote_summary(root) if require_remote and new_artifacts else {},
            remote_error=push_error,
        )
        # The rerun executed the original commit inside a clean sandbox, so
        # provenance reflects that commit rather than the live checkout.
        manifest["git_commit"] = commit
        manifest["dirty"] = False
        manifest["dirty_status"] = {
            "branch": "",
            "head": commit,
            "dirty": False,
            "severity": "clean",
            "message": "Sandbox rerun used a clean checkout.",
            "requested_paths": [],
            "dirty_paths": [],
            "staged_paths": [],
            "unstaged_paths": [],
            "untracked_paths": [],
            "requested_dirty_paths": [],
            "unrelated_dirty_paths": [],
            "unrelated_staged_paths": [],
        }
        manifest["dirty_diff"] = ""
        # Inherit classification from the original so the run stays in the
        # same dashboard family/tab.
        for field in ("kind", "subtype", "initiated_by", "title"):
            if original.get(field):
                manifest[field] = original[field]
        manifest["experiment"] = family
        manifest["rerun_of"] = original_id
        manifest["executor"] = executor
        manifest["run_stamp"] = stamp
        manifest["replication"] = replication
        manifest["replication_checks"] = replication_checks
        if replication_caveat:
            manifest["replication_caveat"] = replication_caveat
        write_result = write_manifest(manifest, root)

        # Mark the original manifest as part of the same family so the
        # dashboard groups the runs together.
        if not original.get("experiment"):
            original["experiment"] = family
            write_manifest(original, root)

        _log(f"Rerun {status}; replication: {replication}")
        return {
            "dry_run": False,
            "plan": plan,
            "status": status,
            "replication": replication,
            "replication_checks": replication_checks,
            "replication_caveat": replication_caveat,
            "manifest": manifest,
            "manifest_path": write_result["path"],
            "artifact_state": artifact_state,
            "remote_error": push_error,
            "result": command_result_to_dict(result),
        }
    finally:
        if not keep_sandbox:
            _remove_sandbox(root, sandbox)

backup_artifact

backup_artifact(path: Path | str, repo_root: Path | str | None = None, *, dry_run: bool = False) -> dict[str, Any]

One-step backup: DVC add + push + verify for a local file.

The local file stays exactly where it is; this just tracks it and ships a copy to S3 so the path is restorable on any machine.

Source code in memory/artifacts.py

def backup_artifact(
    path: Path | str,
    repo_root: Path | str | None = None,
    *,
    dry_run: bool = False,
) -> dict[str, Any]:
    """One-step backup: DVC add + push + verify for a local file.

    The local file stays exactly where it is; this just tracks it and ships a
    copy to S3 so the path is restorable on any machine.
    """

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    rel = repo_relative_path(path, root)
    add_result = dvc_add(rel, root, dry_run=dry_run)
    push_result = dvc_push(root, targets=rel, dry_run=dry_run)
    pointer = root / f"{rel}.dvc"
    pointer_data = _read_dvc_pointer(pointer) if pointer.exists() else {}
    outs = pointer_data.get("outs") if isinstance(pointer_data, Mapping) else []
    first_out = outs[0] if isinstance(outs, list) and outs and isinstance(outs[0], Mapping) else {}
    verified = bool(push_result.ok) and not dry_run
    remote = dvc_remote_summary(root, dry_run=dry_run)
    return {
        "path": rel,
        "pointer": f"{rel}.dvc" if pointer.exists() else "",
        "md5": str(first_out.get("md5") or ""),
        "size": first_out.get("size"),
        "artifact_state": ARTIFACT_STATE_REMOTE_VERIFIED if verified else ARTIFACT_STATE_QUEUED,
        "remote": remote,
        "dvc_cache_uri": _dvc_cache_uri(str(remote.get("url") or ""), str(first_out.get("md5") or "")),
        "add": command_result_to_dict(add_result),
        "push": command_result_to_dict(push_result),
        "dry_run": dry_run,
    }

export_experiment_catalog

export_experiment_catalog(experiment_id: str, repo_root: Path | str | None = None, *, catalog_prefix: str = '', copy_small_artifacts: bool = True, max_artifact_bytes: int = DEFAULT_CATALOG_MAX_BYTES, dry_run: bool = True) -> dict[str, Any]

Export a human-readable S3 catalog beside DVC's content-addressed cache.

Source code in memory/artifacts.py

def export_experiment_catalog(
    experiment_id: str,
    repo_root: Path | str | None = None,
    *,
    catalog_prefix: str = "",
    copy_small_artifacts: bool = True,
    max_artifact_bytes: int = DEFAULT_CATALOG_MAX_BYTES,
    dry_run: bool = True,
) -> dict[str, Any]:
    """Export a human-readable S3 catalog beside DVC's content-addressed cache."""

    root = Path(repo_root).resolve() if repo_root is not None else find_git_root()
    manifest = read_manifest(experiment_id, root)
    safe_id = _safe_artifact_id(str(manifest.get("id", experiment_id)))
    remote = dvc_remote_summary(root, dry_run=dry_run)
    if not remote.get("url") and isinstance(manifest.get("remote"), Mapping):
        remote["url"] = str(manifest["remote"].get("url") or "")
    remote_url = str(remote.get("url") or "").rstrip("/")
    target_prefix = (catalog_prefix or _s3_join(remote_url, "experiments", safe_id)).rstrip("/")
    if not target_prefix.startswith("s3://"):
        raise ValueError("Experiment catalog export requires an S3 catalog prefix or configured S3 DVC remote.")

    local_dir = root / CATALOGS_DIR / safe_id
    manifest_path = artifact_manifest_path(safe_id, root)
    artifacts_payload = {
        "experiment_id": safe_id,
        "generated_at": _utc_now(),
        "dvc_remote": remote,
        "artifacts": _catalog_artifacts(manifest, root, remote_url),
    }
    catalog_payload = {
        "experiment_id": safe_id,
        "title": manifest.get("title") or manifest.get("subtype") or safe_id,
        "status": manifest.get("status", ""),
        "artifact_state": manifest.get("artifact_state", ARTIFACT_STATE_LOCAL_ONLY),
        "manifest": manifest_path.relative_to(root).as_posix(),
        "catalog_prefix": target_prefix,
        "dvc_remote": remote,
        "files": {
            "manifest": _s3_join(target_prefix, "manifest.yaml"),
            "artifacts": _s3_join(target_prefix, "artifacts.json"),
            "catalog": _s3_join(target_prefix, "catalog.json"),
        },
    }
    selected_artifacts = _small_catalog_artifact_copies(
        artifacts_payload["artifacts"],
        root,
        max_artifact_bytes=max_artifact_bytes,
    ) if copy_small_artifacts else []
    catalog_payload["copied_artifacts"] = [
        item | {"catalog_uri": _s3_join(target_prefix, "files", Path(item["path"]).name)}
        for item in selected_artifacts
    ]

    local_files = {
        "manifest": local_dir / "manifest.yaml",
        "artifacts": local_dir / "artifacts.json",
        "catalog": local_dir / "catalog.json",
    }
    if not dry_run:
        local_dir.mkdir(parents=True, exist_ok=True)
        local_files["manifest"].write_text(
            yaml.dump(manifest, default_flow_style=False, sort_keys=False),
            encoding="utf-8",
        )
        local_files["artifacts"].write_text(json.dumps(artifacts_payload, indent=2), encoding="utf-8")
        local_files["catalog"].write_text(json.dumps(catalog_payload, indent=2), encoding="utf-8")

    uploads = [
        _s3_upload(local_files["manifest"], _s3_join(target_prefix, "manifest.yaml"), root, remote, dry_run=dry_run),
        _s3_upload(local_files["artifacts"], _s3_join(target_prefix, "artifacts.json"), root, remote, dry_run=dry_run),
        _s3_upload(local_files["catalog"], _s3_join(target_prefix, "catalog.json"), root, remote, dry_run=dry_run),
    ]
    for item in selected_artifacts:
        uploads.append(_s3_upload(
            root / item["path"],
            _s3_join(target_prefix, "files", Path(item["path"]).name),
            root,
            remote,
            dry_run=dry_run,
        ))

    return {
        "experiment_id": safe_id,
        "catalog_prefix": target_prefix,
        "dry_run": dry_run,
        "local_dir": local_dir.relative_to(root).as_posix(),
        "catalog": catalog_payload,
        "artifacts": artifacts_payload,
        "uploads": uploads,
    }

command_result_to_dict

command_result_to_dict(result: CommandResult | None) -> dict[str, Any] | None

Convert a command result to JSON-serializable data.

Source code in memory/artifacts.py

def command_result_to_dict(result: CommandResult | None) -> dict[str, Any] | None:
    """Convert a command result to JSON-serializable data."""

    if result is None:
        return None
    return {
        "args": list(result.args),
        "cwd": str(result.cwd),
        "returncode": result.returncode,
        "stdout": result.stdout,
        "stderr": result.stderr,
        "dry_run": result.dry_run,
        "ok": result.ok,
    }