diff --git a/.github/workflows/testing.yml b/.github/workflows/testing.yml
index c0ba442db4..caaa84866e 100644
--- a/.github/workflows/testing.yml
+++ b/.github/workflows/testing.yml
@@ -106,6 +106,11 @@ jobs:
         if: matrix.package != 'emmet-core'
         run: python -m pip install --user --no-deps ./emmet-core
 
+      - name: Install editable emmet-archival for emmet-cli
+        shell: bash -l {0}
+        if: matrix.package == 'emmet-cli'
+        run: python -m pip install --user --no-deps ./emmet-archival
+
       - name: Install ${{ matrix.package }}
         shell: bash -l {0}
         run: python -m pip install --user --no-deps -e ./${{ matrix.package }}
diff --git a/emmet-cli/emmet/cli/state_manager.py b/emmet-cli/emmet/cli/state_manager.py
index f52c10b3dc..d02f41dd80 100644
--- a/emmet-cli/emmet/cli/state_manager.py
+++ b/emmet-cli/emmet/cli/state_manager.py
@@ -1,10 +1,12 @@
 from __future__ import annotations
 
+import fcntl
 import json
 import logging
+import os
 from pathlib import Path
 from typing import Any, Callable, Self, TextIO
-import fcntl
+from uuid import uuid4
 
 logger = logging.getLogger("emmet")
 
@@ -32,13 +34,14 @@ class StateManager:
     """Manages persistent state for the CLI application."""
 
     def __init__(self, state_dir: Path | str = Path.home() / ".emmet"):
-        # Store only the state file path
-        self.state_file = str(Path(state_dir) / "state.json")
+        self.state_dir = Path(state_dir)
+        self.state_file = str(self.state_dir / "state.json")
         self._ensure_state_dir()
 
     def _ensure_state_dir(self) -> None:
         """Ensures the state directory exists."""
-        Path(self.state_file).parent.mkdir(parents=True, exist_ok=True)
+        self.state_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
+        self.state_dir.chmod(0o700)
 
     def _load_state(self) -> dict[str, Any]:
         """Loads state from disk. Not thread safe."""
@@ -54,8 +57,37 @@ def _load_state(self) -> dict[str, Any]:
 
     def _save_state(self, state: dict[str, Any]) -> None:
         """Saves current state to disk. Not thread safe."""
-        with Path(self.state_file).open("w") as f:
-            json.dump(state, f, indent=2)
+        state_path = Path(self.state_file)
+        temporary_path = state_path.with_name(f".{state_path.name}.{uuid4().hex}.tmp")
+        descriptor = os.open(
+            temporary_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600
+        )
+        try:
+            try:
+                state_file = os.fdopen(descriptor, "w")
+            except Exception:
+                os.close(descriptor)
+                raise
+            with state_file as f:
+                json.dump(state, f, indent=2)
+                f.flush()
+                os.fsync(f.fileno())
+            os.replace(temporary_path, state_path)
+            try:
+                directory_descriptor = os.open(
+                    state_path.parent, os.O_RDONLY | os.O_DIRECTORY
+                )
+                try:
+                    os.fsync(directory_descriptor)
+                finally:
+                    os.close(directory_descriptor)
+            except OSError:
+                logger.warning(
+                    "State was saved, but its directory entry could not be synced."
+                )
+        except Exception:
+            temporary_path.unlink(missing_ok=True)
+            raise
 
     def get(self, key: str, default: Any = None) -> Any:
         """Gets a value from state."""
diff --git a/emmet-cli/emmet/cli/submission.py b/emmet-cli/emmet/cli/submission.py
index 16ad4ef8f8..0b1fe6da55 100644
--- a/emmet-cli/emmet/cli/submission.py
+++ b/emmet-cli/emmet/cli/submission.py
@@ -7,7 +7,7 @@
 from multiprocessing import get_context
 from os import PathLike, cpu_count
 from pathlib import Path
-from typing import ClassVar, Iterable
+from typing import ClassVar, Iterable, Literal, Protocol
 from uuid import UUID, uuid4
 
 from pydantic import BaseModel, Field, PrivateAttr
@@ -72,6 +72,35 @@ def refresh(self) -> None:
             self.calc_validation_errors.clear()
 
 
+class CalculationChange(BaseModel):
+    """A calculation-level change included in a submission snapshot."""
+
+    calculation_id: UUID
+    status: Literal["added", "changed", "removed"]
+    locator: CalculationLocator
+    calculation: CalculationMetadata | None = None
+    added_files: list[str] = Field(default_factory=list)
+    changed_files: list[str] = Field(default_factory=list)
+    removed_files: list[str] = Field(default_factory=list)
+
+
+class SubmissionChangeSet(BaseModel):
+    """The complete set of changes between two submission snapshots."""
+
+    current_calculations: list[tuple[CalculationLocator, CalculationMetadata]]
+    changes: list[CalculationChange]
+
+    @property
+    def has_changes(self) -> bool:
+        return bool(self.changes)
+
+
+class SubmissionUploader(Protocol):
+    """Upload a staged submission snapshot to a remote service."""
+
+    def upload(self, submission_id: UUID, changes: SubmissionChangeSet) -> None: ...
+
+
 def invoke_calc_refresh(args):
     path, cm = args
     cm.refresh()
@@ -110,9 +139,7 @@ class Submission(BaseModel):
         default=None,
     )
 
-    _pending_push: dict[CalculationLocator, FileMetadata] | None = PrivateAttr(
-        default=None
-    )
+    _pending_push: SubmissionChangeSet | None = PrivateAttr(default=None)
 
     def last_pushed(
         self,
@@ -315,8 +342,8 @@ def _create_calculations_copy(self, refresh: bool = False):
                     cm.refresh()
         return pending_calculations
 
-    def stage_for_push(self) -> list[FileMetadata]:
-        """Stages submission for push. Returns the list of files that will need to be (re)pushed."""
+    def stage_for_push(self) -> SubmissionChangeSet:
+        """Stage and validate the current snapshot for a remote push."""
         self.pending_calculations = self._create_calculations_copy()
 
         if not self.validate_submission():
@@ -327,46 +354,94 @@ def stage_for_push(self) -> list[FileMetadata]:
                 "Submission does not pass validation. Please fix validation errors prior to staging."
             )
 
-        changes = self.get_changed_files_per_calc_path(
+        self._pending_push = self.get_submission_changes(
             self.last_pushed(), self.pending_calculations
         )
-        self._pending_push = changes  # type: ignore[assignment]
+        return self._pending_push
 
-        return [item for sublist in changes.values() for item in sublist]
-
-    def get_changed_files_per_calc_path(
+    def get_submission_changes(
         self,
         previous: list[tuple[CalculationLocator, CalculationMetadata]] | None,
         current: list[tuple[CalculationLocator, CalculationMetadata]],
-    ) -> dict[CalculationLocator, list[FileMetadata]]:
-        changes: dict[CalculationLocator, list[FileMetadata]] = {}
-        if not previous:
-            changes = {k: v.files for k, v in current}
-        else:
-            for loc, cm in current:
-                prev_cm = next((cm_p for loc_p, cm_p in previous if loc_p == loc), None)
-                if prev_cm is None:
-                    changes[loc] = cm.files
-                else:
-                    file_changes = []
-                    for fm in cm.files:
-                        match = next(
-                            (item for item in prev_cm.files if item == fm), None
-                        )
-                        if match is None or fm.hash != match.hash:
-                            file_changes.append(fm)
-                    if file_changes:
-                        changes[loc] = file_changes
-        return changes
-
-    def push(self) -> None:
+    ) -> SubmissionChangeSet:
+        """Return added, changed, and removed calculations and files."""
+        previous_by_id = {
+            calculation.id: (locator, calculation)
+            for locator, calculation in (previous or [])
+        }
+        current_by_id = {
+            calculation.id: (locator, calculation) for locator, calculation in current
+        }
+        changes = []
+
+        for calculation_id in sorted(current_by_id, key=str):
+            locator, calculation = current_by_id[calculation_id]
+            previous_entry = previous_by_id.get(calculation_id)
+            current_files = {file.name: file for file in calculation.files}
+
+            if previous_entry is None:
+                changes.append(
+                    CalculationChange(
+                        calculation_id=calculation_id,
+                        status="added",
+                        locator=locator,
+                        calculation=calculation,
+                        added_files=sorted(current_files),
+                    )
+                )
+                continue
+
+            _, previous_calculation = previous_entry
+            previous_files = {file.name: file for file in previous_calculation.files}
+            added_files = sorted(current_files.keys() - previous_files.keys())
+            removed_files = sorted(previous_files.keys() - current_files.keys())
+            changed_files = sorted(
+                name
+                for name in current_files.keys() & previous_files.keys()
+                if current_files[name].hash != previous_files[name].hash
+            )
+            if added_files or changed_files or removed_files:
+                changes.append(
+                    CalculationChange(
+                        calculation_id=calculation_id,
+                        status="changed",
+                        locator=locator,
+                        calculation=calculation,
+                        added_files=added_files,
+                        changed_files=changed_files,
+                        removed_files=removed_files,
+                    )
+                )
+
+        for calculation_id in sorted(
+            previous_by_id.keys() - current_by_id.keys(), key=str
+        ):
+            locator, calculation = previous_by_id[calculation_id]
+            changes.append(
+                CalculationChange(
+                    calculation_id=calculation_id,
+                    status="removed",
+                    locator=locator,
+                    removed_files=sorted(file.name for file in calculation.files),
+                )
+            )
+
+        return SubmissionChangeSet(
+            current_calculations=current,
+            changes=changes,
+        )
+
+    def push(self, uploader: SubmissionUploader) -> None:
         """Performs the push. Returns info about the push"""
-        if not self.pending_calculations or not self._pending_push:
+        if (
+            self.pending_calculations is None
+            or self._pending_push is None
+            or not self._pending_push.has_changes
+        ):
             raise EmmetCliError("Nothing is staged. Please stage before pushing.")
 
-        if self.get_changed_files_per_calc_path(
-            self.pending_calculations, self._create_calculations_copy(refresh=True)
-        ):
+        current = self._create_calculations_copy(refresh=True)
+        if self.get_submission_changes(self.pending_calculations, current).has_changes:
             raise EmmetCliError(
                 "Files for submission have changed since staging. Please re-stage before pushing."
             )
@@ -378,12 +453,7 @@ def push(self) -> None:
                 "Submission does not pass validation. Please fix validation errors and re-stage."
             )
 
-        # TODO: do push
-        for k, _ in self._pending_push.items():
-            # call RawArchive static method to create file_paths from list of FileMetadata for the pending_calc[k]
-            # construct a RawArchive file for writing
-            # push that file to S3
-            pass
+        uploader.upload(self.id, self._pending_push)
 
         # do bookkeeping
         self.calc_history.append(self.pending_calculations)
diff --git a/emmet-cli/emmet/cli/submit.py b/emmet-cli/emmet/cli/submit.py
index 591a790519..0de4c5a6d4 100644
--- a/emmet-cli/emmet/cli/submit.py
+++ b/emmet-cli/emmet/cli/submit.py
@@ -1,7 +1,11 @@
 import logging
 from pathlib import Path
+from uuid import UUID
+
 import click
-from emmet.cli.submission import Submission
+from emmet.cli.state_manager import StateManager
+from emmet.cli.submission import Submission, SubmissionUploader
+from emmet.cli.upload import HttpSubmissionUploader
 from emmet.cli.utils import EmmetCliError
 
 logger = logging.getLogger("emmet")
@@ -134,17 +138,26 @@ def validate(ctx: click.Context, submission: Path, check_all: bool) -> None:
     click.echo("Use 'emmet tasks status <task_id>' to check the status")
 
 
-def _push_submission(submission_path: Path) -> tuple[bool, str]:
+def _push_submission(
+    submission_path: Path,
+    state_dir: Path | None = None,
+    uploader: SubmissionUploader | None = None,
+) -> tuple[bool, str]:
     """Helper function to push a submission that can run in a separate process."""
     sub = Submission.load(submission_path)
-    updated_file_info = sub.stage_for_push()
-    if not updated_file_info:
+    changes = sub.stage_for_push()
+    if not changes.has_changes:
         return (
             False,
             "Files for submission have not changed since last update. Not pushing.",
         )
 
-    sub.push()
+    if uploader is None:
+        state_manager = StateManager(state_dir or Path.home() / ".emmet")
+        with HttpSubmissionUploader.from_environment(state_manager) as managed_uploader:
+            sub.push(managed_uploader)
+    else:
+        sub.push(uploader)
     sub.save(submission_path)
     return True, f"Successfully updated submission in {submission_path}"
 
@@ -157,6 +170,57 @@ def push(ctx: click.Context, submission: Path) -> None:
 
     Returns a task ID that can be used to check the status."""
     task_manager = ctx.obj["task_manager"]
-    task_id = task_manager.start_task(_push_submission, Path(submission))
+    state_dir = task_manager.state_manager.state_dir
+    task_id = task_manager.start_task(_push_submission, Path(submission), state_dir)
     click.echo(f"Push started. Task ID: {task_id}")
     click.echo("Use 'emmet tasks status <task_id>' to check the status")
+
+
+def _submission_id_from_target(target: str) -> UUID:
+    """Resolve a submission metadata path or UUID to a submission ID."""
+    target_path = Path(target)
+    if target_path.exists():
+        if not target_path.is_file():
+            raise EmmetCliError(f"Submission target is not a file: {target}")
+        try:
+            return Submission.load(target_path).id
+        except (OSError, ValueError):
+            raise EmmetCliError(
+                f"Could not load submission metadata from {target}."
+            ) from None
+    try:
+        return UUID(target)
+    except ValueError:
+        raise EmmetCliError(
+            f"Submission target must be an existing metadata file or UUID: {target}"
+        ) from None
+
+
+def _echo_object_ids(label: str, object_ids: list[str]) -> None:
+    click.echo(f"{label}: {len(object_ids)}")
+    for object_id in object_ids:
+        click.echo(f"  {object_id}")
+
+
+@submit.command("contributor-status")
+@click.pass_context
+def contributor_status(ctx: click.Context) -> None:
+    """Checks whether the current user can contribute submissions."""
+    state_manager = ctx.obj["task_manager"].state_manager
+    with HttpSubmissionUploader.from_environment(state_manager) as client:
+        status = client.contributor_status()
+    click.echo(f"Contributor status: {status}")
+
+
+@submit.command("status")
+@click.argument("target", nargs=1, type=str)
+@click.pass_context
+def submission_status(ctx: click.Context, target: str) -> None:
+    """Checks remote status using a submission metadata file or UUID."""
+    submission_id = _submission_id_from_target(target)
+    state_manager = ctx.obj["task_manager"].state_manager
+    with HttpSubmissionUploader.from_environment(state_manager) as client:
+        status = client.submission_status(submission_id)
+    click.echo(f"Submission {status['submission_id']}: {status['status']}")
+    _echo_object_ids("Completed objects", status["completed_object_ids"])
+    _echo_object_ids("In-progress objects", status["in_progress_object_ids"])
diff --git a/emmet-cli/emmet/cli/upload.py b/emmet-cli/emmet/cli/upload.py
new file mode 100644
index 0000000000..ee97f427d2
--- /dev/null
+++ b/emmet-cli/emmet/cli/upload.py
@@ -0,0 +1,702 @@
+"""Remote upload support for calculation submission archives.
+
+The control-plane service exposes an idempotent three-step protocol:
+
+1. ``POST /submissions/{id}/upload-sessions`` with a snapshot manifest and the
+   objects that require presigned upload URLs.
+2. ``PUT`` each object to the returned URL using the returned headers.
+3. ``POST /submissions/{id}/upload-sessions/{session_id}/complete`` with the
+   uploaded object identifiers and checksums.
+
+Calling the prepare endpoint again with the same ``Idempotency-Key`` refreshes
+expired URLs and returns the same logical upload session.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import os
+from contextlib import suppress
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from typing import Any, Iterator
+from uuid import UUID
+
+import httpx
+
+from emmet.archival.vasp.raw import RawArchive, raw_archive_hierarchy_from_files
+from emmet.cli.state_manager import StateManager
+from emmet.cli.submission import CalculationMetadata, SubmissionChangeSet
+from emmet.cli.utils import EmmetCliError
+
+DEFAULT_API_URL = "https://api.materialsproject.org"
+UPLOAD_STATE_KEY = "submission_uploads"
+ARCHIVE_CONTENT_TYPE = "application/x-hdf5"
+MANIFEST_CONTENT_TYPE = "application/json"
+PROGRESS_CHECKPOINT_INTERVAL = 10
+
+logger = logging.getLogger("emmet")
+
+
+@dataclass(frozen=True)
+class _SessionParts:
+    uploads: dict[str, dict[str, Any]]
+    objects: dict[str, dict[str, Any]]
+    completed_ids: set[str]
+    completed_objects: dict[str, dict[str, Any]]
+
+
+@dataclass
+class _UploadObject:
+    object_id: str
+    path: Path
+    content_type: str
+    calculation_id: UUID | None = None
+    size: int | None = None
+    sha256: str | None = None
+
+    def metadata(self) -> dict[str, Any]:
+        metadata: dict[str, Any] = {
+            "object_id": self.object_id,
+            "content_type": self.content_type,
+        }
+        if self.size is not None:
+            metadata["size"] = self.size
+        if self.sha256 is not None:
+            metadata["sha256"] = self.sha256
+        return metadata
+
+
+@dataclass(frozen=True)
+class _UploadContext:
+    submission_id: UUID
+    snapshot_id: str
+    manifest: dict[str, Any]
+    calculations: dict[UUID, CalculationMetadata]
+
+
+def _canonical_json(value: Any) -> bytes:
+    return json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
+
+
+def _calculation_digest(calculation: CalculationMetadata) -> str:
+    files = sorted(
+        ({"name": file.name, "hash": file.hash} for file in calculation.files),
+        key=lambda file: file["name"] or "",
+    )
+    return hashlib.sha256(_canonical_json(files)).hexdigest()
+
+
+def _file_chunks(path: Path, digest: Any) -> Iterator[bytes]:
+    with path.open("rb") as file:
+        while chunk := file.read(1024 * 1024):
+            digest.update(chunk)
+            yield chunk
+
+
+def _is_unexpired(expires_at: str | None) -> bool:
+    if not expires_at:
+        return False
+    try:
+        expiry = datetime.fromisoformat(expires_at.replace("Z", "+00:00"))
+    except ValueError:
+        return False
+    if expiry.tzinfo is None:
+        expiry = expiry.replace(tzinfo=timezone.utc)
+    return expiry > datetime.now(timezone.utc)
+
+
+class HttpSubmissionUploader:
+    """Reusable uploader for service-provided presigned URLs.
+
+    Call :meth:`close` when finished, or use the uploader as a context manager.
+    Clients supplied by the caller remain owned by the caller and are not closed.
+    """
+
+    def __init__(
+        self,
+        state_manager: StateManager,
+        api_key: str,
+        api_url: str = DEFAULT_API_URL,
+        client: httpx.Client | None = None,
+    ) -> None:
+        if not api_key:
+            raise EmmetCliError(
+                "MP_API_KEY must be set before contacting the submission service."
+            )
+        self.state_manager = state_manager
+        self.api_key = api_key
+        self.api_url = api_url.rstrip("/")
+        self.client = client or httpx.Client(timeout=60.0)
+        self._owns_client = client is None
+
+    @classmethod
+    def from_environment(
+        cls, state_manager: StateManager, client: httpx.Client | None = None
+    ) -> HttpSubmissionUploader:
+        """Create an uploader using the CLI's supported environment settings."""
+        api_key = os.environ.get("MP_API_KEY", "")
+        if not api_key:
+            api_key = os.environ.get("EMMET_API_TOKEN", "")
+            if api_key:
+                logger.warning("EMMET_API_TOKEN is deprecated; set MP_API_KEY instead.")
+        return cls(
+            state_manager=state_manager,
+            api_key=api_key,
+            api_url=os.environ.get("EMMET_API_URL", DEFAULT_API_URL),
+            client=client,
+        )
+
+    def contributor_status(self) -> str:
+        """Return the authenticated user's contributor status."""
+        response = self._control_request(
+            "GET",
+            "/submissions/contributor-status",
+            action="Checking contributor status",
+        )
+        try:
+            payload = response.json()
+            contributor_status = payload["status"]
+            if contributor_status not in {"active", "inactive", "expired"}:
+                raise ValueError
+        except (KeyError, TypeError, ValueError):
+            raise EmmetCliError(
+                "Submission service returned an invalid contributor status response."
+            ) from None
+        return contributor_status
+
+    def submission_status(self, submission_id: UUID) -> dict[str, Any]:
+        """Return remote upload status for a submission."""
+        response = self._control_request(
+            "POST",
+            f"/submissions/{submission_id}/status",
+            action="Checking submission status",
+        )
+        try:
+            payload = response.json()
+            response_submission_id = UUID(payload["submission_id"])
+            submission_state = payload["status"]
+            completed = payload["completed_object_ids"]
+            in_progress = payload["in_progress_object_ids"]
+            if (
+                response_submission_id != submission_id
+                or submission_state not in {"complete", "incomplete"}
+                or not isinstance(completed, list)
+                or not all(isinstance(object_id, str) for object_id in completed)
+                or not isinstance(in_progress, list)
+                or not all(isinstance(object_id, str) for object_id in in_progress)
+            ):
+                raise ValueError
+        except (KeyError, TypeError, ValueError):
+            raise EmmetCliError(
+                "Submission service returned an invalid submission status response."
+            ) from None
+        return {
+            "submission_id": str(response_submission_id),
+            "status": submission_state,
+            "completed_object_ids": completed,
+            "in_progress_object_ids": in_progress,
+        }
+
+    def close(self) -> None:
+        """Close the internally-created HTTP client, if any."""
+        if self._owns_client and not self.client.is_closed:
+            self.client.close()
+
+    def __enter__(self) -> HttpSubmissionUploader:
+        return self
+
+    def __exit__(self, *args: Any) -> None:
+        self.close()
+
+    def upload(self, submission_id: UUID, changes: SubmissionChangeSet) -> None:
+        """Archive and upload a staged snapshot, then finalize it remotely."""
+        with TemporaryDirectory(prefix="emmet-upload-") as directory:
+            objects, manifest = self._build_objects(
+                submission_id, changes, Path(directory)
+            )
+            snapshot_id = manifest["snapshot_id"]
+            context = _UploadContext(
+                submission_id=submission_id,
+                snapshot_id=snapshot_id,
+                manifest=manifest,
+                calculations={
+                    calculation.id: calculation
+                    for _, calculation in changes.current_calculations
+                },
+            )
+            session = self._get_or_prepare_session(
+                submission_id, snapshot_id, manifest, objects, context
+            )
+            session_parts = self._parse_session(
+                session, "Upload session contains invalid details."
+            )
+            uploads = session_parts.uploads
+            completed = session_parts.completed_ids
+            session_objects = session_parts.objects
+            pending_checkpoint = 0
+
+            try:
+                for object_info in objects:
+                    object_id = object_info.object_id
+                    if object_id in completed:
+                        continue
+                    upload = uploads.get(object_id)
+                    if upload is None:
+                        raise EmmetCliError(
+                            f"Upload service did not return a URL for object {object_id}."
+                        )
+                    self._put_object(object_info, upload, context)
+                    completed.add(object_id)
+                    session_objects[object_id] = self._object_metadata(object_info)
+                    pending_checkpoint += 1
+                    if pending_checkpoint >= PROGRESS_CHECKPOINT_INTERVAL:
+                        self._checkpoint_session(
+                            submission_id, session, completed, session_objects
+                        )
+                        pending_checkpoint = 0
+            except Exception:
+                if pending_checkpoint:
+                    with suppress(Exception):
+                        self._checkpoint_session(
+                            submission_id, session, completed, session_objects
+                        )
+                raise
+
+            if pending_checkpoint:
+                self._checkpoint_session(
+                    submission_id, session, completed, session_objects
+                )
+            self._finalize_session(submission_id, session)
+            self._clear_session(submission_id)
+
+    def _build_objects(
+        self,
+        submission_id: UUID,
+        changes: SubmissionChangeSet,
+        directory: Path,
+    ) -> tuple[list[_UploadObject], dict[str, Any]]:
+        change_by_id = {change.calculation_id: change for change in changes.changes}
+        calculation_entries = []
+        archive_specs = []
+
+        for _, calculation in sorted(
+            changes.current_calculations, key=lambda item: str(item[1].id)
+        ):
+            digest = _calculation_digest(calculation)
+            object_id = f"calculations/{calculation.id}/{digest}.h5"
+            change = change_by_id.get(calculation.id)
+            calculation_entries.append(
+                {
+                    "calculation_id": str(calculation.id),
+                    "archive_object_id": object_id,
+                    "content_digest": digest,
+                    "change": change.status if change else "unchanged",
+                    "added_files": change.added_files if change else [],
+                    "changed_files": change.changed_files if change else [],
+                    "removed_files": change.removed_files if change else [],
+                }
+            )
+            if change is not None:
+                archive_specs.append((calculation, object_id))
+
+        manifest_body = {
+            "schema_version": 1,
+            "submission_id": str(submission_id),
+            "calculations": calculation_entries,
+            "removed_calculations": [
+                {
+                    "calculation_id": str(change.calculation_id),
+                    "removed_files": change.removed_files,
+                }
+                for change in changes.changes
+                if change.status == "removed"
+            ],
+        }
+        snapshot_id = hashlib.sha256(_canonical_json(manifest_body)).hexdigest()
+        manifest = {**manifest_body, "snapshot_id": snapshot_id}
+
+        objects = []
+        for calculation, object_id in archive_specs:
+            archive_path = directory / f"{calculation.id}.h5"
+            objects.append(
+                self._object_info(
+                    object_id,
+                    archive_path,
+                    ARCHIVE_CONTENT_TYPE,
+                    calculation_id=calculation.id,
+                )
+            )
+
+        manifest_path = directory / "manifest.json"
+        objects.append(
+            self._object_info(
+                f"manifests/{snapshot_id}.json",
+                manifest_path,
+                MANIFEST_CONTENT_TYPE,
+            )
+        )
+        return objects, manifest
+
+    @staticmethod
+    def _object_info(
+        object_id: str,
+        path: Path,
+        content_type: str,
+        calculation_id: UUID | None = None,
+    ) -> _UploadObject:
+        return _UploadObject(
+            object_id=object_id,
+            path=path,
+            content_type=content_type,
+            calculation_id=calculation_id,
+            size=path.stat().st_size if path.exists() else None,
+        )
+
+    @staticmethod
+    def _object_metadata(object_info: _UploadObject) -> dict[str, Any]:
+        """Return persistable metadata describing the bytes prepared for upload."""
+        return object_info.metadata()
+
+    @classmethod
+    def _parse_session(
+        cls,
+        session: dict[str, Any],
+        error: str,
+        *,
+        require_completed_objects: bool = False,
+    ) -> _SessionParts:
+        def items_by_id(
+            key: str, require_checksum: bool = False, required: bool = False
+        ):
+            if required and key not in session:
+                raise EmmetCliError(error)
+            items = session.get(key, [])
+            if not isinstance(items, list):
+                raise EmmetCliError(error)
+            parsed = {}
+            for item in items:
+                if (
+                    not isinstance(item, dict)
+                    or not isinstance(item.get("object_id"), str)
+                    or (require_checksum and not cls._has_checksum(item))
+                ):
+                    raise EmmetCliError(error)
+                parsed[item["object_id"]] = item
+            return parsed
+
+        completed_ids = session.get("completed_object_ids", [])
+        if not isinstance(completed_ids, list) or not all(
+            isinstance(object_id, str) for object_id in completed_ids
+        ):
+            raise EmmetCliError(error)
+        return _SessionParts(
+            uploads=items_by_id("uploads"),
+            objects=items_by_id("objects"),
+            completed_ids=set(completed_ids),
+            completed_objects=items_by_id(
+                "completed_objects",
+                require_checksum=True,
+                required=require_completed_objects,
+            ),
+        )
+
+    @staticmethod
+    def _has_checksum(item: dict[str, Any]) -> bool:
+        return isinstance(item.get("sha256"), str)
+
+    @staticmethod
+    def _objects_have_checksums(
+        completed: set[str], objects: dict[str, dict[str, Any]]
+    ) -> bool:
+        return all(
+            object_id in objects
+            and HttpSubmissionUploader._has_checksum(objects[object_id])
+            for object_id in completed
+        )
+
+    @classmethod
+    def _resolve_object_metadata(
+        cls,
+        base: dict[str, Any],
+        service_completed: set[str],
+        service_objects: dict[str, dict[str, Any]],
+        previous_objects: dict[str, dict[str, Any]],
+    ) -> tuple[dict[str, Any], bool]:
+        object_id = base["object_id"]
+        service_metadata = service_objects.get(object_id)
+        if service_metadata is not None:
+            return {**base, **service_metadata}, True
+        previous_metadata = previous_objects.get(object_id)
+        if (
+            object_id in service_completed
+            and previous_metadata is not None
+            and cls._has_checksum(previous_metadata)
+        ):
+            return {**base, **previous_metadata}, True
+        return base, False
+
+    def _get_or_prepare_session(
+        self,
+        submission_id: UUID,
+        snapshot_id: str,
+        manifest: dict[str, Any],
+        objects: list[_UploadObject],
+        context: _UploadContext,
+    ) -> dict[str, Any]:
+        previous_session = self._load_session(submission_id)
+        required_ids = {item.object_id for item in objects}
+        try:
+            previous_parts = self._parse_session(
+                previous_session, "Cached upload session is invalid."
+            )
+        except EmmetCliError:
+            previous_session = {}
+            previous_parts = _SessionParts({}, {}, set(), {})
+        if (
+            previous_session.get("snapshot_id") == snapshot_id
+            and _is_unexpired(previous_session.get("expires_at"))
+            and required_ids
+            <= previous_parts.uploads.keys() | previous_parts.completed_ids
+            and required_ids <= previous_parts.objects.keys()
+            and self._objects_have_checksums(
+                previous_parts.completed_ids, previous_parts.objects
+            )
+        ):
+            return previous_session
+
+        same_snapshot = previous_session.get("snapshot_id") == snapshot_id
+        for item in objects:
+            previous_metadata = previous_parts.objects.get(item.object_id, {})
+            previous_size = previous_metadata.get("size")
+            if same_snapshot and isinstance(previous_size, int):
+                item.size = previous_size
+            if item.size is None:
+                self._materialize_object(item, context)
+
+        payload = {
+            "snapshot_id": snapshot_id,
+            "manifest": manifest,
+            "objects": [self._object_metadata(item) for item in objects],
+        }
+        response = self._control_request(
+            "POST",
+            f"/submissions/{submission_id}/upload-sessions",
+            json=payload,
+            headers={"Idempotency-Key": snapshot_id},
+            action="Preparing upload session",
+        )
+        try:
+            session = response.json()
+            if not isinstance(session, dict) or not isinstance(
+                session["session_id"], str
+            ):
+                raise TypeError
+            service_parts = self._parse_session(
+                session,
+                "Upload service returned an invalid prepare response.",
+                require_completed_objects=True,
+            )
+        except (KeyError, TypeError, ValueError, EmmetCliError):
+            raise EmmetCliError(
+                "Upload service returned an invalid prepare response."
+            ) from None
+        session["snapshot_id"] = snapshot_id
+        if previous_session.get("snapshot_id") != snapshot_id:
+            previous_parts = _SessionParts({}, {}, set(), {})
+        service_completed = (
+            service_parts.completed_ids | service_parts.completed_objects.keys()
+        )
+        completed = set()
+        resolved_objects = []
+        for item in objects:
+            metadata, is_completed = self._resolve_object_metadata(
+                self._object_metadata(item),
+                service_completed,
+                service_parts.completed_objects,
+                previous_parts.objects,
+            )
+            resolved_objects.append(metadata)
+            if is_completed:
+                completed.add(item.object_id)
+        session["completed_object_ids"] = list(completed)
+        session["objects"] = resolved_objects
+        try:
+            self._save_session(submission_id, session)
+        except Exception:
+            raise EmmetCliError(
+                "Upload session was prepared remotely but could not be saved locally. "
+                "Retry the push to resume or refresh the session."
+            ) from None
+        return session
+
+    @staticmethod
+    def _materialize_object(
+        object_info: _UploadObject, context: _UploadContext
+    ) -> None:
+        path = object_info.path
+        if not path.exists():
+            if object_info.calculation_id is not None:
+                calculation = context.calculations[object_info.calculation_id]
+                RawArchive(
+                    file_paths=raw_archive_hierarchy_from_files(calculation.files)
+                ).to_archive(
+                    path,
+                    metadata={
+                        "submission_id": str(context.submission_id),
+                        "calculation_id": str(object_info.calculation_id),
+                        "snapshot_id": context.snapshot_id,
+                    },
+                )
+            else:
+                path.write_bytes(_canonical_json(context.manifest))
+        object_info.size = path.stat().st_size
+
+    def _put_object(
+        self,
+        object_info: _UploadObject,
+        upload: dict[str, Any],
+        context: _UploadContext,
+    ) -> None:
+        digest = hashlib.sha256()
+        try:
+            self._materialize_object(object_info, context)
+            response = self.client.put(
+                upload["url"],
+                headers=upload.get("headers", {}),
+                content=_file_chunks(object_info.path, digest),
+            )
+            response.raise_for_status()
+            object_info.sha256 = digest.hexdigest()
+        except (KeyError, TypeError) as exc:
+            raise EmmetCliError(
+                f"Upload service returned invalid details for object {object_info.object_id}."
+            ) from exc
+        except httpx.HTTPStatusError as exc:
+            raise EmmetCliError(
+                f"Uploading object {object_info.object_id} failed with HTTP "
+                f"{exc.response.status_code}."
+            ) from None
+        except httpx.RequestError:
+            raise EmmetCliError(
+                f"Uploading object {object_info.object_id} failed due to a network error."
+            ) from None
+        except OSError:
+            raise EmmetCliError(
+                f"Reading object {object_info.object_id} failed during upload. "
+                "Verify the local files are accessible and retry the push."
+            ) from None
+
+    def _finalize_session(
+        self,
+        submission_id: UUID,
+        session: dict[str, Any],
+    ) -> None:
+        try:
+            payload = {
+                "snapshot_id": session["snapshot_id"],
+                "objects": [
+                    {"object_id": item["object_id"], "sha256": item["sha256"]}
+                    for item in session["objects"]
+                ],
+            }
+            session_id = session["session_id"]
+        except (KeyError, TypeError):
+            raise EmmetCliError(
+                "Upload session is missing object checksums required for finalization. "
+                "Retry the push to reconcile remote upload progress."
+            ) from None
+        response = self._control_request(
+            "POST",
+            f"/submissions/{submission_id}/upload-sessions/{session_id}/complete",
+            json=payload,
+            headers={"Idempotency-Key": session["snapshot_id"]},
+            action="Finalizing upload session",
+        )
+        try:
+            result = response.json()
+            response_submission_id = UUID(result["submission_id"])
+            if (
+                response_submission_id != submission_id
+                or result["session_state"] != "complete"
+            ):
+                raise ValueError
+        except (KeyError, TypeError, ValueError):
+            raise EmmetCliError(
+                "Upload service returned an invalid finalization response."
+            ) from None
+
+    def _control_request(
+        self,
+        method: str,
+        path: str,
+        *,
+        action: str,
+        json: dict[str, Any] | None = None,
+        headers: dict[str, str] | None = None,
+    ) -> httpx.Response:
+        request_headers = {"X-API-KEY": self.api_key}
+        request_headers.update(headers or {})
+        try:
+            request_kwargs: dict[str, Any] = {"headers": request_headers}
+            if json is not None:
+                request_kwargs["json"] = json
+            response = self.client.request(
+                method, f"{self.api_url}{path}", **request_kwargs
+            )
+            response.raise_for_status()
+            return response
+        except httpx.HTTPStatusError as exc:
+            raise EmmetCliError(
+                f"{action} failed with HTTP {exc.response.status_code}."
+            ) from None
+        except httpx.RequestError:
+            raise EmmetCliError(f"{action} failed due to a network error.") from None
+
+    def _load_session(self, submission_id: UUID) -> dict[str, Any]:
+        sessions = self.state_manager.get(UPLOAD_STATE_KEY, {})
+        session = sessions.get(str(submission_id), {})
+        return session if isinstance(session, dict) else {}
+
+    def _save_session(self, submission_id: UUID, session: dict[str, Any]) -> None:
+        def save(sessions: Any) -> dict[str, Any]:
+            updated = dict(sessions or {})
+            updated[str(submission_id)] = session
+            return updated
+
+        self.state_manager.update(UPLOAD_STATE_KEY, save)
+
+    def _checkpoint_session(
+        self,
+        submission_id: UUID,
+        session: dict[str, Any],
+        completed: set[str],
+        session_objects: dict[str, dict[str, Any]],
+    ) -> None:
+        session["completed_object_ids"] = list(completed)
+        session["objects"] = list(session_objects.values())
+        try:
+            self._save_session(submission_id, session)
+        except Exception:
+            raise EmmetCliError(
+                "Remote upload progress could not be saved locally. Retry the push; "
+                "expired URLs will be refreshed and completed objects reconciled."
+            ) from None
+
+    def _clear_session(self, submission_id: UUID) -> None:
+        def clear(sessions: Any) -> dict[str, Any]:
+            updated = dict(sessions or {})
+            updated.pop(str(submission_id), None)
+            return updated
+
+        try:
+            self.state_manager.update(UPLOAD_STATE_KEY, clear)
+        except Exception:
+            logger.warning(
+                "Remote upload completed, but its local retry state could not be "
+                "cleared. A later retry may safely finalize the same snapshot again."
+            )
diff --git a/emmet-cli/pyproject.toml b/emmet-cli/pyproject.toml
index 9d842157dc..9f9f32c559 100644
--- a/emmet-cli/pyproject.toml
+++ b/emmet-cli/pyproject.toml
@@ -26,7 +26,9 @@ authors = [
 dependencies = [
   "click",
   "colorama",
+  "emmet-archival",
   "emmet-core>=0.85.1",
+  "httpx",
   "pymatgen-io-validation>=0.1.1",
   "psutil>=5.9.0",
 ]
diff --git a/emmet-cli/readme.md b/emmet-cli/readme.md
index ee60315dfa..d9b89ee3be 100644
--- a/emmet-cli/readme.md
+++ b/emmet-cli/readme.md
@@ -23,9 +23,11 @@ Options:
 
 Commands:
   add-to       Adds more files to the submission.
+  contributor-status  Checks whether the current user can contribute submissions.
   create       Creates a new MP data submission.
   push         Pushes the latest version of an MP data submission.
   remove-from  Removes files from the submission.
+  status       Checks the remote status of a submission.
   validate     Locally validates the latest version of an MP data...
 ```
 ### create
@@ -90,3 +92,30 @@ Usage: emmet submit push [OPTIONS] SUBMISSION
 Options:
   --help  Show this message and exit.
 ```
+
+#### Remote upload configuration
+
+Remote submission commands authenticate to the Materials Project submission service
+with the `MP_API_KEY` environment variable. The legacy `EMMET_API_TOKEN` variable
+is accepted with a deprecation warning. The CLI uses
+`https://api.materialsproject.org` by default; set `EMMET_API_URL` to target a
+development service. Tokens and presigned URLs are never written to submission
+metadata or logs. Active upload sessions are cached in the protected CLI state
+directory so interrupted pushes can resume.
+
+Use `emmet submit contributor-status` to check whether the authenticated user can
+contribute submissions. Use `emmet submit status TARGET` to inspect remote upload
+progress, where `TARGET` is either a local submission metadata file or a submission
+UUID.
+
+For each added or changed calculation, the CLI creates a complete RawArchive
+HDF5 object. A JSON snapshot manifest references current calculation archives
+and records removed files and calculations. The service contract is:
+
+1. `POST /submissions/{id}/upload-sessions` prepares or refreshes an
+   idempotent session and returns presigned object URLs.
+2. The CLI uploads each archive and manifest with `PUT`.
+3. `POST /submissions/{id}/upload-sessions/{session_id}/complete` confirms the
+   uploaded object identifiers and checksums.
+
+Local submission history advances only after the service confirms completion.
diff --git a/emmet-cli/requirements/ubuntu-latest_py3.11.txt b/emmet-cli/requirements/ubuntu-latest_py3.11.txt
index 986288e07e..505eaf7488 100644
--- a/emmet-cli/requirements/ubuntu-latest_py3.11.txt
+++ b/emmet-cli/requirements/ubuntu-latest_py3.11.txt
@@ -152,3 +152,25 @@ uncertainties==3.2.3
     # via pymatgen-core
 urllib3==2.7.0
     # via requests
+backports-zstd==1.5.0 ; python_version < "3.14"
+    # via emmet-archival
+anyio==4.13.0
+    # via httpx
+donfig==0.8.1.post1
+    # via zarr
+google-crc32c==1.8.0
+    # via zarr
+h11==0.16.0
+    # via httpcore
+h5py==3.16.0
+    # via emmet-archival
+httpcore==1.0.9
+    # via httpx
+httpx==0.28.1
+    # via emmet-cli (pyproject.toml)
+numcodecs==0.16.5
+    # via zarr
+pyarrow==24.0.0
+    # via emmet-archival
+zarr==3.1.6
+    # via emmet-archival
diff --git a/emmet-cli/requirements/ubuntu-latest_py3.11_extras.txt b/emmet-cli/requirements/ubuntu-latest_py3.11_extras.txt
index afdfeba55d..4b1641342d 100644
--- a/emmet-cli/requirements/ubuntu-latest_py3.11_extras.txt
+++ b/emmet-cli/requirements/ubuntu-latest_py3.11_extras.txt
@@ -332,3 +332,25 @@ wcmatch==11.0.1
     # via mkdocs-awesome-pages-plugin
 wincertstore==0.2.1
     # via emmet-cli (pyproject.toml)
+backports-zstd==1.5.0 ; python_version < "3.14"
+    # via emmet-archival
+anyio==4.13.0
+    # via httpx
+donfig==0.8.1.post1
+    # via zarr
+google-crc32c==1.8.0
+    # via zarr
+h11==0.16.0
+    # via httpcore
+h5py==3.16.0
+    # via emmet-archival
+httpcore==1.0.9
+    # via httpx
+httpx==0.28.1
+    # via emmet-cli (pyproject.toml)
+numcodecs==0.16.5
+    # via zarr
+pyarrow==24.0.0
+    # via emmet-archival
+zarr==3.1.6
+    # via emmet-archival
diff --git a/emmet-cli/requirements/ubuntu-latest_py3.12.txt b/emmet-cli/requirements/ubuntu-latest_py3.12.txt
index 93819bcba0..b3a59267f5 100644
--- a/emmet-cli/requirements/ubuntu-latest_py3.12.txt
+++ b/emmet-cli/requirements/ubuntu-latest_py3.12.txt
@@ -151,3 +151,25 @@ uncertainties==3.2.3
     # via pymatgen-core
 urllib3==2.7.0
     # via requests
+backports-zstd==1.5.0 ; python_version < "3.14"
+    # via emmet-archival
+anyio==4.13.0
+    # via httpx
+donfig==0.8.1.post1
+    # via zarr
+google-crc32c==1.8.0
+    # via zarr
+h11==0.16.0
+    # via httpcore
+h5py==3.16.0
+    # via emmet-archival
+httpcore==1.0.9
+    # via httpx
+httpx==0.28.1
+    # via emmet-cli (pyproject.toml)
+numcodecs==0.16.5
+    # via zarr
+pyarrow==24.0.0
+    # via emmet-archival
+zarr==3.2.1
+    # via emmet-archival
diff --git a/emmet-cli/requirements/ubuntu-latest_py3.12_extras.txt b/emmet-cli/requirements/ubuntu-latest_py3.12_extras.txt
index e6d9dc49e3..dc55822050 100644
--- a/emmet-cli/requirements/ubuntu-latest_py3.12_extras.txt
+++ b/emmet-cli/requirements/ubuntu-latest_py3.12_extras.txt
@@ -331,3 +331,25 @@ wcmatch==11.0.1
     # via mkdocs-awesome-pages-plugin
 wincertstore==0.2.1
     # via emmet-cli (pyproject.toml)
+backports-zstd==1.5.0 ; python_version < "3.14"
+    # via emmet-archival
+anyio==4.13.0
+    # via httpx
+donfig==0.8.1.post1
+    # via zarr
+google-crc32c==1.8.0
+    # via zarr
+h11==0.16.0
+    # via httpcore
+h5py==3.16.0
+    # via emmet-archival
+httpcore==1.0.9
+    # via httpx
+httpx==0.28.1
+    # via emmet-cli (pyproject.toml)
+numcodecs==0.16.5
+    # via zarr
+pyarrow==24.0.0
+    # via emmet-archival
+zarr==3.2.1
+    # via emmet-archival
diff --git a/emmet-cli/requirements/ubuntu-latest_py3.13.txt b/emmet-cli/requirements/ubuntu-latest_py3.13.txt
index 3c7225333e..1ed1fcc8a7 100644
--- a/emmet-cli/requirements/ubuntu-latest_py3.13.txt
+++ b/emmet-cli/requirements/ubuntu-latest_py3.13.txt
@@ -150,3 +150,25 @@ uncertainties==3.2.3
     # via pymatgen-core
 urllib3==2.7.0
     # via requests
+backports-zstd==1.5.0 ; python_version < "3.14"
+    # via emmet-archival
+anyio==4.13.0
+    # via httpx
+donfig==0.8.1.post1
+    # via zarr
+google-crc32c==1.8.0
+    # via zarr
+h11==0.16.0
+    # via httpcore
+h5py==3.16.0
+    # via emmet-archival
+httpcore==1.0.9
+    # via httpx
+httpx==0.28.1
+    # via emmet-cli (pyproject.toml)
+numcodecs==0.16.5
+    # via zarr
+pyarrow==24.0.0
+    # via emmet-archival
+zarr==3.2.1
+    # via emmet-archival
diff --git a/emmet-cli/requirements/ubuntu-latest_py3.13_extras.txt b/emmet-cli/requirements/ubuntu-latest_py3.13_extras.txt
index f505173824..37fe78eb2d 100644
--- a/emmet-cli/requirements/ubuntu-latest_py3.13_extras.txt
+++ b/emmet-cli/requirements/ubuntu-latest_py3.13_extras.txt
@@ -330,3 +330,25 @@ wcmatch==11.0.1
     # via mkdocs-awesome-pages-plugin
 wincertstore==0.2.1
     # via emmet-cli (pyproject.toml)
+backports-zstd==1.5.0 ; python_version < "3.14"
+    # via emmet-archival
+anyio==4.13.0
+    # via httpx
+donfig==0.8.1.post1
+    # via zarr
+google-crc32c==1.8.0
+    # via zarr
+h11==0.16.0
+    # via httpcore
+h5py==3.16.0
+    # via emmet-archival
+httpcore==1.0.9
+    # via httpx
+httpx==0.28.1
+    # via emmet-cli (pyproject.toml)
+numcodecs==0.16.5
+    # via zarr
+pyarrow==24.0.0
+    # via emmet-archival
+zarr==3.2.1
+    # via emmet-archival
diff --git a/emmet-cli/requirements/ubuntu-latest_py3.14.txt b/emmet-cli/requirements/ubuntu-latest_py3.14.txt
index ea342f6ee2..2ea986f703 100644
--- a/emmet-cli/requirements/ubuntu-latest_py3.14.txt
+++ b/emmet-cli/requirements/ubuntu-latest_py3.14.txt
@@ -150,3 +150,23 @@ uncertainties==3.2.3
     # via pymatgen-core
 urllib3==2.7.0
     # via requests
+anyio==4.13.0
+    # via httpx
+donfig==0.8.1.post1
+    # via zarr
+google-crc32c==1.8.0
+    # via zarr
+h11==0.16.0
+    # via httpcore
+h5py==3.16.0
+    # via emmet-archival
+httpcore==1.0.9
+    # via httpx
+httpx==0.28.1
+    # via emmet-cli (pyproject.toml)
+numcodecs==0.16.5
+    # via zarr
+pyarrow==24.0.0
+    # via emmet-archival
+zarr==3.2.1
+    # via emmet-archival
diff --git a/emmet-cli/requirements/ubuntu-latest_py3.14_extras.txt b/emmet-cli/requirements/ubuntu-latest_py3.14_extras.txt
index 2de87d06b2..2e61d60e44 100644
--- a/emmet-cli/requirements/ubuntu-latest_py3.14_extras.txt
+++ b/emmet-cli/requirements/ubuntu-latest_py3.14_extras.txt
@@ -330,3 +330,23 @@ wcmatch==11.0.1
     # via mkdocs-awesome-pages-plugin
 wincertstore==0.2.1
     # via emmet-cli (pyproject.toml)
+anyio==4.13.0
+    # via httpx
+donfig==0.8.1.post1
+    # via zarr
+google-crc32c==1.8.0
+    # via zarr
+h11==0.16.0
+    # via httpcore
+h5py==3.16.0
+    # via emmet-archival
+httpcore==1.0.9
+    # via httpx
+httpx==0.28.1
+    # via emmet-cli (pyproject.toml)
+numcodecs==0.16.5
+    # via zarr
+pyarrow==24.0.0
+    # via emmet-archival
+zarr==3.2.1
+    # via emmet-archival
diff --git a/emmet-cli/tests/test_state_manager.py b/emmet-cli/tests/test_state_manager.py
index 4e14de8de7..f1491dbccc 100644
--- a/emmet-cli/tests/test_state_manager.py
+++ b/emmet-cli/tests/test_state_manager.py
@@ -1,13 +1,28 @@
 import json
+import os
+import stat
 from pathlib import Path
+import pytest
 from emmet.cli.state_manager import StateManager
 
 
 def test_init_creates_state_dir(temp_state_dir):
     """Test that initialization creates the state directory."""
-    StateManager(state_dir=temp_state_dir)
+    manager = StateManager(state_dir=temp_state_dir)
     assert temp_state_dir.exists()
     assert temp_state_dir.is_dir()
+    assert manager.state_dir == temp_state_dir
+
+
+def test_state_dir_is_private_with_permissive_umask(tmp_path):
+    state_dir = tmp_path / "permissive-umask"
+    previous_umask = os.umask(0)
+    try:
+        StateManager(state_dir=state_dir)
+    finally:
+        os.umask(previous_umask)
+
+    assert stat.S_IMODE(state_dir.stat().st_mode) == 0o700
 
 
 def test_load_empty_state(state_manager):
@@ -41,6 +56,7 @@ def test_set_and_get(state_manager):
         json.loads(Path(state_manager.state_file).read_text())["test_key"]
         == "test_value"
     )
+    assert stat.S_IMODE(Path(state_manager.state_file).stat().st_mode) == 0o600
 
 
 def test_update_atomically_transforms_value(state_manager, monkeypatch):
@@ -72,6 +88,85 @@ def save_state(state):
     assert state_manager.get("other_key") == "preserved"
 
 
+def test_save_state_closes_descriptor_if_fdopen_fails(state_manager, monkeypatch):
+    original_open = os.open
+    original_close = os.close
+    opened_descriptors = []
+    closed_descriptors = []
+
+    def tracked_open(*args, **kwargs):
+        descriptor = original_open(*args, **kwargs)
+        opened_descriptors.append(descriptor)
+        return descriptor
+
+    def tracked_close(descriptor):
+        closed_descriptors.append(descriptor)
+        original_close(descriptor)
+
+    def fail_fdopen(*args, **kwargs):
+        raise OSError("fdopen failed")
+
+    monkeypatch.setattr(os, "open", tracked_open)
+    monkeypatch.setattr(os, "close", tracked_close)
+    monkeypatch.setattr(os, "fdopen", fail_fdopen)
+
+    with pytest.raises(OSError, match="fdopen failed"):
+        state_manager._save_state({"key": "value"})
+
+    assert closed_descriptors == opened_descriptors
+
+
+def test_save_state_failure_preserves_existing_state(state_manager, monkeypatch):
+    state_manager._save_state({"session": "existing"})
+    state_path = Path(state_manager.state_file)
+    original_contents = state_path.read_text()
+
+    def fail_dump(*args, **kwargs):
+        raise OSError("disk full")
+
+    monkeypatch.setattr(json, "dump", fail_dump)
+
+    with pytest.raises(OSError, match="disk full"):
+        state_manager._save_state({"session": "replacement"})
+
+    assert state_path.read_text() == original_contents
+    assert list(state_path.parent.glob(".state.json.*.tmp")) == []
+
+
+def test_save_state_fsyncs_file_and_directory(state_manager, monkeypatch):
+    original_fsync = os.fsync
+    fsynced_modes = []
+
+    def track_fsync(descriptor):
+        fsynced_modes.append(os.fstat(descriptor).st_mode)
+        original_fsync(descriptor)
+
+    monkeypatch.setattr(os, "fsync", track_fsync)
+
+    state_manager._save_state({"session": "durable"})
+
+    assert stat.S_ISREG(fsynced_modes[0])
+    assert stat.S_ISDIR(fsynced_modes[1])
+
+
+def test_directory_fsync_failure_does_not_report_save_failure(
+    state_manager, monkeypatch, caplog
+):
+    original_fsync = os.fsync
+
+    def fail_directory_fsync(descriptor):
+        if stat.S_ISDIR(os.fstat(descriptor).st_mode):
+            raise OSError("directory sync failed")
+        original_fsync(descriptor)
+
+    monkeypatch.setattr(os, "fsync", fail_directory_fsync)
+
+    state_manager._save_state({"session": "persisted"})
+
+    assert state_manager._load_state() == {"session": "persisted"}
+    assert "directory entry could not be synced" in caplog.text
+
+
 def test_save_and_load_state(temp_state_dir):
     """Test that state is properly saved and loaded."""
     manager1 = StateManager(state_dir=temp_state_dir)
diff --git a/emmet-cli/tests/test_submission.py b/emmet-cli/tests/test_submission.py
index 4a7f197666..91e0a5573d 100644
--- a/emmet-cli/tests/test_submission.py
+++ b/emmet-cli/tests/test_submission.py
@@ -5,6 +5,17 @@
 import pytest
 
 
+class RecordingUploader:
+    def __init__(self, error=None):
+        self.error = error
+        self.uploads = []
+
+    def upload(self, submission_id, changes):
+        if self.error:
+            raise self.error
+        self.uploads.append((submission_id, changes))
+
+
 @pytest.fixture(scope="session")
 def tmp_structure(tmp_path_factory):
     directory_structure = {
@@ -148,25 +159,6 @@ def test_remove_from(sub_file, tmp_structure):
     assert len(removed) == 9
 
 
-def test_changed_files(sub_file):
-    sub = Submission.load(Path(sub_file))
-    changed = sub.get_changed_files_per_calc_path(
-        sub.calculations, sub._create_calculations_copy(refresh=True)
-    )
-    assert len(changed) == 7
-
-    sub.calculations = sub._create_calculations_copy(refresh=True)
-
-    changed = sub.get_changed_files_per_calc_path(
-        sub.calculations, sub._create_calculations_copy(refresh=True)
-    )
-    assert len(changed) == 0
-    changed = sub.get_changed_files_per_calc_path(
-        sub.last_pushed(), sub._create_calculations_copy(refresh=True)
-    )
-    assert len(changed) == 7
-
-
 @pytest.mark.parametrize("changed_index", [0, 1, 2])
 def test_refresh_invalidates_cached_validation(calculation_metadata, changed_index):
     original_hashes = [file.hash for file in calculation_metadata.files]
@@ -234,19 +226,71 @@ def test_validate_submission(sub_file, validation_sub_file):
 
 def test_changed_files_to_push(validation_sub_file):
     sub = Submission.load(Path(validation_sub_file))
+    uploader = RecordingUploader()
 
     with pytest.raises(EmmetCliError) as ex_info:
-        sub.push()
+        sub.push(uploader)
 
     assert "Nothing is staged" in str(ex_info.value)
 
     changed = sub.stage_for_push()
-    assert len(changed) == 10
+    assert changed.has_changes
+    assert sum(len(change.added_files) for change in changed.changes) == 10
     changed = sub.stage_for_push()
-    assert len(changed) == 10
+    assert changed.has_changes
 
-    sub.push()
+    sub.push(uploader)
+    assert len(uploader.uploads) == 1
     changed = sub.stage_for_push()
-    assert len(changed) == 0
+    assert not changed.has_changes
+
+
+def test_submission_changes_include_file_and_calculation_removals(
+    calculation_metadata,
+):
+    locator = CalculationLocator(path=Path("/calculation"), modifier="standard")
+    second = calculation_metadata.model_copy(deep=True)
+    second.id = calculation_metadata.id
+    submission = Submission(calculations=[(locator, calculation_metadata)])
+    submission.calc_history.append([(locator, second)])
+
+    removed_file = calculation_metadata.files.pop()
+    changes = submission.get_submission_changes(
+        submission.last_pushed(), submission.calculations
+    )
+
+    assert changes.has_changes
+    assert changes.changes[0].status == "changed"
+    assert changes.changes[0].removed_files == [removed_file.name]
+
+    removed_calculation_changes = submission.get_submission_changes(
+        submission.calculations, []
+    )
+    assert removed_calculation_changes.changes[0].status == "removed"
+    assert removed_calculation_changes.changes[0].calculation is None
+
+
+def test_failed_upload_does_not_advance_history(validation_sub_file):
+    sub = Submission.load(Path(validation_sub_file))
+    sub.stage_for_push()
+
+    with pytest.raises(EmmetCliError, match="remote failure"):
+        sub.push(RecordingUploader(EmmetCliError("remote failure")))
+
+    assert sub.calc_history == []
+    assert sub.pending_calculations is not None
+
+
+def test_files_changed_after_staging_block_push(validation_sub_file):
+    sub = Submission.load(Path(validation_sub_file))
+    uploader = RecordingUploader()
+    sub.stage_for_push()
+    changed_file = sub.calculations[0][1].files[0].path
+    changed_file.write_bytes(changed_file.read_bytes() + b"changed after staging")
+
+    with pytest.raises(EmmetCliError, match="changed since staging"):
+        sub.push(uploader)
 
-    # check that if file changed after stage then push raises exception
+    assert uploader.uploads == []
+    assert sub.calc_history == []
+    assert sub.pending_calculations is not None
diff --git a/emmet-cli/tests/test_submit.py b/emmet-cli/tests/test_submit.py
index 5aebbccf5f..e773912f59 100644
--- a/emmet-cli/tests/test_submit.py
+++ b/emmet-cli/tests/test_submit.py
@@ -1,6 +1,7 @@
 import os
-import pytest
 from pathlib import Path
+from uuid import uuid4
+
 from emmet.cli.submission import Submission
 from emmet.cli.submit import (
     submit,
@@ -9,6 +10,7 @@
     _remove_from_submission,
 )
 from emmet.cli.utils import EmmetCliError
+from emmet.cli.upload import HttpSubmissionUploader
 from conftest import (
     wait_for_task_completion_and_assert_success,
 )
@@ -131,9 +133,116 @@ def test_validate_failure(invalid_validation_sub_file, cli_runner, inline_task_m
     )
 
 
-@pytest.mark.skip(reason="Push coverage is deferred to issue #1486.")
-def test_push(validation_sub_file, cli_runner, task_manager):
+def test_push(
+    validation_sub_file,
+    cli_runner,
+    inline_task_manager,
+    monkeypatch,
+):
+    uploads = []
+
+    class FakeUploader:
+        def upload(self, submission_id, changes):
+            uploads.append((submission_id, changes))
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *args):
+            pass
+
+    monkeypatch.setattr(
+        HttpSubmissionUploader,
+        "from_environment",
+        lambda state_manager: FakeUploader(),
+    )
     result = cli_runner(submit, ["push", validation_sub_file])
 
     assert result.exit_code == 0
     assert "Push started." in result.output
+    final_status = wait_for_task_completion_and_assert_success(
+        result, inline_task_manager
+    )
+    assert final_status["result"][0] is True
+    assert len(uploads) == 1
+    assert len(Submission.load(Path(validation_sub_file)).calc_history) == 1
+
+
+class FakeStatusClient:
+    def __init__(self):
+        self.submission_ids = []
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        pass
+
+    def contributor_status(self):
+        return "active"
+
+    def submission_status(self, submission_id):
+        self.submission_ids.append(submission_id)
+        return {
+            "submission_id": str(submission_id),
+            "status": "incomplete",
+            "completed_object_ids": ["complete-object"],
+            "in_progress_object_ids": ["pending-object"],
+        }
+
+
+def test_contributor_status(cli_runner, monkeypatch):
+    client = FakeStatusClient()
+    monkeypatch.setattr(
+        HttpSubmissionUploader,
+        "from_environment",
+        lambda state_manager: client,
+    )
+
+    result = cli_runner(submit, ["contributor-status"])
+
+    assert result.exit_code == 0
+    assert result.output == "Contributor status: active\n"
+
+
+def test_submission_status_accepts_uuid(cli_runner, monkeypatch):
+    client = FakeStatusClient()
+    submission_id = uuid4()
+    monkeypatch.setattr(
+        HttpSubmissionUploader,
+        "from_environment",
+        lambda state_manager: client,
+    )
+
+    result = cli_runner(submit, ["status", str(submission_id)])
+
+    assert result.exit_code == 0
+    assert client.submission_ids == [submission_id]
+    assert f"Submission {submission_id}: incomplete" in result.output
+    assert "Completed objects: 1\n  complete-object" in result.output
+    assert "In-progress objects: 1\n  pending-object" in result.output
+
+
+def test_submission_status_accepts_metadata_file(
+    validation_sub_file, cli_runner, monkeypatch
+):
+    client = FakeStatusClient()
+    submission_id = Submission.load(Path(validation_sub_file)).id
+    monkeypatch.setattr(
+        HttpSubmissionUploader,
+        "from_environment",
+        lambda state_manager: client,
+    )
+
+    result = cli_runner(submit, ["status", validation_sub_file])
+
+    assert result.exit_code == 0
+    assert client.submission_ids == [submission_id]
+
+
+def test_submission_status_rejects_invalid_target(cli_runner):
+    result = cli_runner(submit, ["status", "not-a-submission"])
+
+    assert result.exit_code != 0
+    assert isinstance(result.exception, EmmetCliError)
+    assert "existing metadata file or UUID" in str(result.exception)
diff --git a/emmet-cli/tests/test_upload.py b/emmet-cli/tests/test_upload.py
new file mode 100644
index 0000000000..2353428fed
--- /dev/null
+++ b/emmet-cli/tests/test_upload.py
@@ -0,0 +1,583 @@
+import hashlib
+import json
+import traceback
+import builtins
+from datetime import datetime, timedelta, timezone
+from uuid import uuid4
+
+import httpx
+import pytest
+
+import emmet.cli.upload as upload_module
+from emmet.cli.state_manager import StateManager
+from emmet.cli.submission import CalculationMetadata, Submission
+from emmet.cli.upload import HttpSubmissionUploader, UPLOAD_STATE_KEY
+from emmet.cli.utils import EmmetCliError
+from emmet.core.vasp.utils import CalculationLocator, FileMetadata
+
+
+def _submission_with_raw_files(tmp_path):
+    files = []
+    for name in ("INCAR", "POSCAR"):
+        path = tmp_path / name
+        path.write_text(f"contents of {name}")
+        metadata = FileMetadata(name=name, path=path)
+        metadata.compute_hash()
+        files.append(metadata)
+    calculation = CalculationMetadata(files=files, calc_valid=True)
+    locator = CalculationLocator(path=tmp_path, modifier="standard")
+    return Submission(calculations=[(locator, calculation)])
+
+
+class UploadService:
+    def __init__(
+        self,
+        fail_object=None,
+        fail_finalize=False,
+        malformed_upload=False,
+        omit_completed_objects=False,
+        finalize_state="complete",
+    ):
+        self.fail_object = fail_object
+        self.fail_finalize = fail_finalize
+        self.malformed_upload = malformed_upload
+        self.omit_completed_objects = omit_completed_objects
+        self.finalize_state = finalize_state
+        self.prepare_requests = []
+        self.prepare_headers = []
+        self.put_attempts = []
+        self.puts = {}
+        self.finalize_requests = []
+        self.finalize_headers = []
+
+    def __call__(self, request):
+        content = request.read()
+        if request.url.host == "uploads.test":
+            object_id = request.headers["x-object-id"]
+            self.put_attempts.append(object_id)
+            if object_id == self.fail_object or (
+                self.fail_object == "manifest" and object_id.endswith(".json")
+            ):
+                return httpx.Response(500)
+            self.puts[object_id] = content
+            return httpx.Response(200)
+
+        payload = json.loads(content)
+        if request.url.path.endswith("/complete"):
+            self.finalize_requests.append(payload)
+            self.finalize_headers.append(request.headers)
+            if self.fail_finalize:
+                self.fail_finalize = False
+                return httpx.Response(500)
+            for item in payload["objects"]:
+                uploaded_sha256 = hashlib.sha256(
+                    self.puts[item["object_id"]]
+                ).hexdigest()
+                if item["sha256"] != uploaded_sha256:
+                    return httpx.Response(422)
+            response_submission_id = request.url.path.split("/")[2]
+            return httpx.Response(
+                200,
+                json={
+                    "submission_id": response_submission_id,
+                    "session_state": self.finalize_state,
+                },
+            )
+
+        self.prepare_requests.append(payload)
+        self.prepare_headers.append(request.headers)
+        uploads = [
+            {
+                "object_id": item["object_id"],
+                "url": f"https://uploads.test/{index}",
+                "headers": {"x-object-id": item["object_id"]},
+            }
+            for index, item in enumerate(payload["objects"])
+        ]
+        if self.malformed_upload:
+            uploads[0].pop("object_id")
+        response_payload = {
+            "session_id": "session-1",
+            "expires_at": (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat(),
+            "uploads": uploads,
+            "completed_object_ids": list(self.puts),
+        }
+        if not self.omit_completed_objects:
+            response_payload["completed_objects"] = [
+                {
+                    "object_id": object_id,
+                    "sha256": hashlib.sha256(uploaded).hexdigest(),
+                }
+                for object_id, uploaded in self.puts.items()
+            ]
+        return httpx.Response(200, json=response_payload)
+
+
+def _uploader(state_manager, service):
+    client = httpx.Client(transport=httpx.MockTransport(service))
+    return HttpSubmissionUploader(
+        state_manager=state_manager,
+        api_key="secret-token",
+        api_url="https://api.test",
+        client=client,
+    )
+
+
+def test_environment_configuration_requires_token(tmp_path, monkeypatch):
+    monkeypatch.delenv("MP_API_KEY", raising=False)
+    monkeypatch.delenv("EMMET_API_TOKEN", raising=False)
+
+    with pytest.raises(EmmetCliError, match="MP_API_KEY"):
+        HttpSubmissionUploader.from_environment(StateManager(tmp_path / "state"))
+
+
+def test_environment_prefers_mp_api_key(tmp_path, monkeypatch):
+    monkeypatch.setenv("MP_API_KEY", "current-key")
+    monkeypatch.setenv("EMMET_API_TOKEN", "legacy-key")
+
+    with HttpSubmissionUploader.from_environment(
+        StateManager(tmp_path / "state")
+    ) as uploader:
+        assert uploader.api_key == "current-key"
+
+
+def test_environment_supports_deprecated_token(tmp_path, monkeypatch, caplog):
+    monkeypatch.delenv("MP_API_KEY", raising=False)
+    monkeypatch.setenv("EMMET_API_TOKEN", "legacy-key")
+
+    with HttpSubmissionUploader.from_environment(
+        StateManager(tmp_path / "state")
+    ) as uploader:
+        assert uploader.api_key == "legacy-key"
+    assert "deprecated" in caplog.text
+
+
+def test_context_manager_closes_owned_client(tmp_path):
+    uploader = HttpSubmissionUploader(
+        state_manager=StateManager(tmp_path / "state"),
+        api_key="secret-token",
+    )
+    client = uploader.client
+
+    with uploader:
+        assert not client.is_closed
+
+    assert client.is_closed
+
+
+def test_contributor_status_matches_server_route(tmp_path):
+    def service(request):
+        assert request.method == "GET"
+        assert request.url.path == "/submissions/contributor-status"
+        assert request.headers["x-api-key"] == "secret-token"
+        assert request.content == b""
+        return httpx.Response(200, json={"status": "active"})
+
+    client = httpx.Client(transport=httpx.MockTransport(service))
+    uploader = HttpSubmissionUploader(
+        state_manager=StateManager(tmp_path / "state"),
+        api_key="secret-token",
+        api_url="https://api.test",
+        client=client,
+    )
+
+    assert uploader.contributor_status() == "active"
+
+
+def test_submission_status_matches_server_route(tmp_path):
+    submission_id = uuid4()
+
+    def service(request):
+        assert request.method == "POST"
+        assert request.url.path == f"/submissions/{submission_id}/status"
+        assert request.content == b""
+        return httpx.Response(
+            200,
+            json={
+                "submission_id": str(submission_id),
+                "status": "incomplete",
+                "completed_object_ids": ["completed-object"],
+                "in_progress_object_ids": ["pending-object"],
+            },
+        )
+
+    client = httpx.Client(transport=httpx.MockTransport(service))
+    uploader = HttpSubmissionUploader(
+        state_manager=StateManager(tmp_path / "state"),
+        api_key="secret-token",
+        api_url="https://api.test",
+        client=client,
+    )
+
+    assert uploader.submission_status(submission_id) == {
+        "submission_id": str(submission_id),
+        "status": "incomplete",
+        "completed_object_ids": ["completed-object"],
+        "in_progress_object_ids": ["pending-object"],
+    }
+
+
+def test_uploads_raw_archive_and_snapshot_manifest(tmp_path):
+    submission = _submission_with_raw_files(tmp_path)
+    changes = submission.stage_for_push()
+    state_manager = StateManager(tmp_path / "state")
+    service = UploadService()
+
+    uploader = _uploader(state_manager, service)
+    submission.push(uploader)
+
+    assert len(service.prepare_requests) == 1
+    assert service.prepare_headers[0]["x-api-key"] == "secret-token"
+    assert "authorization" not in service.prepare_headers[0]
+    assert len(service.puts) == 2
+    archive = next(
+        content
+        for object_id, content in service.puts.items()
+        if object_id.endswith(".h5")
+    )
+    assert archive.startswith(b"\x89HDF")
+    manifest = json.loads(
+        next(
+            content
+            for object_id, content in service.puts.items()
+            if object_id.endswith(".json")
+        )
+    )
+    assert service.prepare_headers[0]["idempotency-key"] == manifest["snapshot_id"]
+    assert manifest["submission_id"] == str(submission.id)
+    assert manifest["calculations"][0]["change"] == "added"
+    assert manifest["calculations"][0]["added_files"] == ["INCAR", "POSCAR"]
+    assert len(service.finalize_requests) == 1
+    assert service.finalize_headers[0]["idempotency-key"] == manifest["snapshot_id"]
+    assert all("sha256" not in item for item in service.prepare_requests[0]["objects"])
+    assert all(
+        isinstance(item.get("size"), int) and item["size"] > 0
+        for item in service.prepare_requests[0]["objects"]
+    )
+    assert state_manager.get(UPLOAD_STATE_KEY) == {}
+    assert len(submission.calc_history) == 1
+    assert changes.has_changes
+    assert not uploader.client.is_closed
+
+    second_directory = tmp_path / "second"
+    second_directory.mkdir()
+    second_submission = _submission_with_raw_files(second_directory)
+    uploader.upload(second_submission.id, second_submission.stage_for_push())
+    assert len(service.finalize_requests) == 2
+    assert not uploader.client.is_closed
+
+
+def test_partial_upload_resumes_saved_session(tmp_path):
+    submission = _submission_with_raw_files(tmp_path)
+    submission.stage_for_push()
+    state_manager = StateManager(tmp_path / "state")
+    service = UploadService(fail_object="manifest")
+
+    with pytest.raises(EmmetCliError, match="Uploading object") as exc_info:
+        submission.push(_uploader(state_manager, service))
+    rendered_error = "".join(traceback.format_exception(exc_info.value))
+    assert "secret-token" not in rendered_error
+    assert "uploads.test" not in rendered_error
+
+    saved = state_manager.get(UPLOAD_STATE_KEY)[str(submission.id)]
+    assert len(saved["completed_object_ids"]) == 1
+    assert saved["completed_object_ids"][0].endswith(".h5")
+    assert "secret-token" not in json.dumps(saved)
+    assert len(submission.calc_history) == 0
+
+    service.fail_object = None
+    submission.push(_uploader(state_manager, service))
+
+    assert len(service.prepare_requests) == 1
+    assert len(submission.calc_history) == 1
+    assert state_manager.get(UPLOAD_STATE_KEY) == {}
+
+
+def test_malformed_prepare_upload_is_wrapped(tmp_path):
+    submission = _submission_with_raw_files(tmp_path)
+    submission.stage_for_push()
+    service = UploadService(malformed_upload=True)
+
+    with pytest.raises(EmmetCliError, match="invalid prepare response"):
+        submission.push(_uploader(StateManager(tmp_path / "state"), service))
+
+
+def test_prepare_requires_completed_objects_field(tmp_path):
+    submission = _submission_with_raw_files(tmp_path)
+    submission.stage_for_push()
+    service = UploadService(omit_completed_objects=True)
+
+    with pytest.raises(EmmetCliError, match="invalid prepare response"):
+        submission.push(_uploader(StateManager(tmp_path / "state"), service))
+
+
+def test_finalize_requires_complete_session_state(tmp_path):
+    submission = _submission_with_raw_files(tmp_path)
+    submission.stage_for_push()
+    service = UploadService(finalize_state="in_progress")
+
+    with pytest.raises(EmmetCliError, match="invalid finalization response"):
+        submission.push(_uploader(StateManager(tmp_path / "state"), service))
+
+    assert len(submission.calc_history) == 0
+
+
+def test_streaming_file_read_error_is_retryable(tmp_path, monkeypatch):
+    submission = _submission_with_raw_files(tmp_path)
+    submission.stage_for_push()
+    service = UploadService()
+
+    def broken_file_chunks(path, digest):
+        raise OSError("file disappeared")
+        yield b""  # pragma: no cover
+
+    monkeypatch.setattr(upload_module, "_file_chunks", broken_file_chunks)
+
+    with pytest.raises(EmmetCliError, match="accessible and retry"):
+        submission.push(_uploader(StateManager(tmp_path / "state"), service))
+
+
+def test_prepare_state_failure_is_retryable(tmp_path, monkeypatch):
+    submission = _submission_with_raw_files(tmp_path)
+    submission.stage_for_push()
+    state_manager = StateManager(tmp_path / "state")
+    service = UploadService()
+
+    def fail_update(*args):
+        raise TypeError("not serializable")
+
+    monkeypatch.setattr(state_manager, "update", fail_update)
+
+    with pytest.raises(EmmetCliError, match="prepared remotely"):
+        submission.push(_uploader(state_manager, service))
+
+    assert len(service.prepare_requests) == 1
+    assert service.puts == {}
+
+
+def test_checkpoint_wraps_non_os_state_error(tmp_path, monkeypatch):
+    state_manager = StateManager(tmp_path / "state")
+    uploader = _uploader(state_manager, UploadService())
+
+    def fail_update(*args):
+        raise TypeError("not serializable")
+
+    monkeypatch.setattr(state_manager, "update", fail_update)
+
+    with pytest.raises(EmmetCliError, match="progress could not be saved"):
+        uploader._checkpoint_session(
+            submission_id=uuid4(),
+            session={},
+            completed=set(),
+            session_objects={},
+        )
+
+
+def test_upload_progress_uses_batched_atomic_checkpoints(tmp_path, monkeypatch):
+    submission = _submission_with_raw_files(tmp_path)
+    changes = submission.stage_for_push()
+    state_manager = StateManager(tmp_path / "state")
+    service = UploadService()
+    uploader = _uploader(state_manager, service)
+    objects = []
+    for index in range(25):
+        path = tmp_path / f"object-{index}"
+        path.write_bytes(f"object {index}".encode())
+        objects.append(
+            uploader._object_info(f"objects/{index}", path, "application/octet-stream")
+        )
+    monkeypatch.setattr(
+        uploader,
+        "_build_objects",
+        lambda *args: (objects, {"snapshot_id": "snapshot-1"}),
+    )
+    original_update = state_manager.update
+    update_calls = 0
+    sort_calls = 0
+
+    def count_update(key, updater):
+        nonlocal update_calls
+        update_calls += 1
+        return original_update(key, updater)
+
+    def count_sorted(*args, **kwargs):
+        nonlocal sort_calls
+        sort_calls += 1
+        return builtins.sorted(*args, **kwargs)
+
+    monkeypatch.setattr(state_manager, "update", count_update)
+    monkeypatch.setattr(upload_module, "sorted", count_sorted, raising=False)
+
+    uploader.upload(submission.id, changes)
+
+    assert len(service.puts) == 25
+    assert update_calls == 5  # prepare, 2 batches, final partial batch, and clear
+    assert sort_calls == 0
+
+
+def test_expired_session_refreshes_urls_without_reuploading_completed_objects(
+    tmp_path, monkeypatch
+):
+    submission = _submission_with_raw_files(tmp_path)
+    submission.stage_for_push()
+    state_manager = StateManager(tmp_path / "state")
+    service = UploadService(fail_object="manifest")
+    original_to_archive = upload_module.RawArchive.to_archive
+    archive_writes = 0
+
+    def count_archive_writes(*args, **kwargs):
+        nonlocal archive_writes
+        archive_writes += 1
+        return original_to_archive(*args, **kwargs)
+
+    monkeypatch.setattr(upload_module.RawArchive, "to_archive", count_archive_writes)
+
+    with pytest.raises(EmmetCliError):
+        submission.push(_uploader(state_manager, service))
+
+    sessions = state_manager.get(UPLOAD_STATE_KEY)
+    sessions[str(submission.id)]["expires_at"] = "2000-01-01T00:00:00+00:00"
+    state_manager.set(UPLOAD_STATE_KEY, sessions)
+    archive_id = sessions[str(submission.id)]["completed_object_ids"][0]
+    service.fail_object = None
+
+    submission.push(_uploader(state_manager, service))
+
+    assert len(service.prepare_requests) == 2
+    assert service.put_attempts.count(archive_id) == 1
+    assert archive_writes == 1
+
+
+def test_checkpoint_failure_preserves_upload_error_and_remote_progress(
+    tmp_path, monkeypatch
+):
+    submission = _submission_with_raw_files(tmp_path)
+    submission.stage_for_push()
+    state_manager = StateManager(tmp_path / "state")
+    service = UploadService(fail_object="manifest")
+    original_update = state_manager.update
+    update_calls = 0
+
+    def fail_progress_checkpoint(key, updater):
+        nonlocal update_calls
+        update_calls += 1
+        if update_calls == 2:
+            raise OSError("disk full")
+        return original_update(key, updater)
+
+    monkeypatch.setattr(state_manager, "update", fail_progress_checkpoint)
+
+    with pytest.raises(EmmetCliError, match="Uploading object"):
+        submission.push(_uploader(state_manager, service))
+
+    sessions = state_manager.get(UPLOAD_STATE_KEY)
+    archive_id = next(
+        object_id for object_id in service.puts if object_id.endswith(".h5")
+    )
+    assert sessions[str(submission.id)]["completed_object_ids"] == []
+
+    monkeypatch.setattr(state_manager, "update", original_update)
+    sessions[str(submission.id)]["expires_at"] = "2000-01-01T00:00:00+00:00"
+    state_manager.set(UPLOAD_STATE_KEY, sessions)
+    service.fail_object = None
+
+    submission.push(_uploader(state_manager, service))
+
+    assert len(service.prepare_requests) == 2
+    assert service.put_attempts.count(archive_id) == 1
+
+
+def test_retry_after_lost_finalize_uses_uploaded_checksums(tmp_path, monkeypatch):
+    submission = _submission_with_raw_files(tmp_path)
+    submission.stage_for_push()
+    state_manager = StateManager(tmp_path / "state")
+    service = UploadService(fail_finalize=True)
+    original_to_archive = upload_module.RawArchive.to_archive
+    archive_writes = 0
+
+    def count_archive_writes(*args, **kwargs):
+        nonlocal archive_writes
+        archive_writes += 1
+        return original_to_archive(*args, **kwargs)
+
+    monkeypatch.setattr(upload_module.RawArchive, "to_archive", count_archive_writes)
+
+    with pytest.raises(EmmetCliError, match="Finalizing upload session"):
+        submission.push(_uploader(state_manager, service))
+    first_put_attempts = list(service.put_attempts)
+    uploaded_archive_id = next(
+        object_id for object_id in service.puts if object_id.endswith(".h5")
+    )
+    uploaded_archive_sha256 = hashlib.sha256(
+        service.puts[uploaded_archive_id]
+    ).hexdigest()
+    assert archive_writes == 1
+    assert len(submission.calc_history) == 0
+
+    submission.push(_uploader(state_manager, service))
+
+    assert archive_writes == 1
+    assert service.put_attempts == first_put_attempts
+    assert len(service.finalize_requests) == 2
+    finalized_archive = next(
+        item
+        for item in service.finalize_requests[-1]["objects"]
+        if item["object_id"] == uploaded_archive_id
+    )
+    assert finalized_archive["sha256"] == uploaded_archive_sha256
+    assert len(submission.calc_history) == 1
+
+
+def test_clear_failure_does_not_hide_successful_push(tmp_path, monkeypatch, caplog):
+    submission = _submission_with_raw_files(tmp_path)
+    changes = submission.stage_for_push()
+    state_manager = StateManager(tmp_path / "state")
+    service = UploadService()
+    original_update = state_manager.update
+    update_calls = 0
+
+    def fail_clear(key, updater):
+        nonlocal update_calls
+        update_calls += 1
+        if update_calls == 3:
+            raise OSError("disk full")
+        return original_update(key, updater)
+
+    monkeypatch.setattr(state_manager, "update", fail_clear)
+
+    submission.push(_uploader(state_manager, service))
+
+    assert len(submission.calc_history) == 1
+    assert str(submission.id) in state_manager.get(UPLOAD_STATE_KEY)
+    assert "local retry state could not be cleared" in caplog.text
+
+    monkeypatch.setattr(state_manager, "update", original_update)
+    _uploader(state_manager, service).upload(submission.id, changes)
+
+    assert len(service.put_attempts) == 2
+    assert len(service.finalize_requests) == 2
+    assert (
+        service.finalize_headers[0]["idempotency-key"]
+        == service.finalize_headers[1]["idempotency-key"]
+    )
+    assert state_manager.get(UPLOAD_STATE_KEY) == {}
+
+
+def test_removal_only_push_uploads_manifest(tmp_path):
+    submission = _submission_with_raw_files(tmp_path)
+    previous = submission._create_calculations_copy()
+    removed_id = previous[0][1].id
+    submission.calc_history.append(previous)
+    submission.calculations = []
+    changes = submission.stage_for_push()
+    state_manager = StateManager(tmp_path / "state")
+    service = UploadService()
+
+    submission.push(_uploader(state_manager, service))
+
+    assert [change.status for change in changes.changes] == ["removed"]
+    assert list(service.puts) == [
+        f"manifests/{service.prepare_requests[0]['snapshot_id']}.json"
+    ]
+    manifest = json.loads(next(iter(service.puts.values())))
+    assert manifest["removed_calculations"][0]["calculation_id"] == str(removed_id)