diff --git a/src/aignostics/application/_cli.py b/src/aignostics/application/_cli.py
index 9ee527859..7e6d1e5fd 100644
--- a/src/aignostics/application/_cli.py
+++ b/src/aignostics/application/_cli.py
@@ -998,6 +998,9 @@ def run_describe(
 def run_dump_metadata(
     run_id: Annotated[str, typer.Argument(help="Id of the run to dump custom metadata for")],
     pretty: Annotated[bool, typer.Option(help="Pretty print JSON output with indentation")] = False,
+    show_checksum: Annotated[
+        bool, typer.Option("--show-checksum", help="Include custom_metadata_checksum in output for use with --checksum")
+    ] = False,
 ) -> None:
     """Dump custom metadata of a run as JSON to stdout."""
     logger.trace("Dumping custom metadata for run with ID '{}'", run_id)
@@ -1006,11 +1009,19 @@ def run_dump_metadata(
         run = Service().application_run(run_id).details()
         custom_metadata = run.custom_metadata if hasattr(run, "custom_metadata") else {}
 
+        if show_checksum:
+            output: object = {
+                "custom_metadata": custom_metadata,
+                "custom_metadata_checksum": run.custom_metadata_checksum,
+            }
+        else:
+            output = custom_metadata
+
         # Output JSON to stdout
         if pretty:
-            print(json.dumps(custom_metadata, indent=2))
+            print(json.dumps(output, indent=2))
         else:
-            print(json.dumps(custom_metadata))
+            print(json.dumps(output))
 
         logger.debug("Dumped custom metadata for run with ID '{}'", run_id)
     except NotFoundException:
@@ -1028,6 +1039,9 @@ def run_dump_item_metadata(
     run_id: Annotated[str, typer.Argument(help="Id of the run containing the item")],
     external_id: Annotated[str, typer.Argument(help="External ID of the item to dump custom metadata for")],
     pretty: Annotated[bool, typer.Option(help="Pretty print JSON output with indentation")] = False,
+    show_checksum: Annotated[
+        bool, typer.Option("--show-checksum", help="Include custom_metadata_checksum in output for use with --checksum")
+    ] = False,
 ) -> None:
     """Dump custom metadata of an item as JSON to stdout."""
     logger.trace("Dumping custom metadata for item '{}' in run with ID '{}'", external_id, run_id)
@@ -1052,11 +1066,19 @@ def run_dump_item_metadata(
 
         custom_metadata = item.custom_metadata if hasattr(item, "custom_metadata") else {}
 
+        if show_checksum:
+            item_output: object = {
+                "custom_metadata": custom_metadata,
+                "custom_metadata_checksum": item.custom_metadata_checksum,
+            }
+        else:
+            item_output = custom_metadata
+
         # Output JSON to stdout
         if pretty:
-            print(json.dumps(custom_metadata, indent=2))
+            print(json.dumps(item_output, indent=2))
         else:
-            print(json.dumps(custom_metadata))
+            print(json.dumps(item_output))
 
         logger.debug("Dumped custom metadata for item '{}' in run with ID '{}'", external_id, run_id)
     except NotFoundException:
@@ -1213,10 +1235,22 @@ def run_update_metadata(
     metadata_json: Annotated[
         str, typer.Argument(..., help='Custom metadata as JSON string (e.g., \'{"key": "value"}\')')
     ],
+    checksum: Annotated[
+        str | None,
+        typer.Option(
+            "--checksum",
+            help=(
+                "Optional checksum for optimistic concurrency control. "
+                "The server rejects the update with HTTP 412 if the metadata was modified since this checksum was read."
+            ),
+        ),
+    ] = None,
 ) -> None:
     """Update custom metadata for a run."""
     import json  # noqa: PLC0415
 
+    from aignostics.platform import ConcurrencyConflictError  # noqa: PLC0415
+
     logger.trace("Updating custom metadata for run with ID '{}'", run_id)
 
     try:
@@ -1230,13 +1264,17 @@ def run_update_metadata(
             console.print(f"[error]Error:[/error] Invalid JSON: {e}")
             sys.exit(1)
 
-        Service().application_run_update_custom_metadata(run_id, custom_metadata)
+        Service().application_run_update_custom_metadata(run_id, custom_metadata, custom_metadata_checksum=checksum)
         logger.debug("Updated custom metadata for run with ID '{}'.", run_id)
         console.print(f"Successfully updated custom metadata for run with ID '{run_id}'.")
     except NotFoundException:
         logger.warning(f"Run with ID '{run_id}' not found.")
         console.print(f"[warning]Warning:[/warning] Run with ID '{run_id}' not found.")
         sys.exit(2)
+    except ConcurrencyConflictError as e:
+        logger.warning(f"Concurrency conflict updating metadata for run '{run_id}': {e}")
+        console.print(f"[warning]Warning:[/warning] Metadata was modified by another process. Re-read and retry: {e}")
+        sys.exit(3)
     except ValueError as e:
         logger.warning(f"Run ID '{run_id}' invalid or metadata invalid: {e}")
         console.print(f"[warning]Warning:[/warning] Run ID '{run_id}' invalid or metadata invalid: {e}")
@@ -1254,10 +1292,22 @@ def run_update_item_metadata(
     metadata_json: Annotated[
         str, typer.Argument(..., help='Custom metadata as JSON string (e.g., \'{"key": "value"}\')')
     ],
+    checksum: Annotated[
+        str | None,
+        typer.Option(
+            "--checksum",
+            help=(
+                "Optional checksum for optimistic concurrency control. "
+                "The server rejects the update with HTTP 412 if the metadata was modified since this checksum was read."
+            ),
+        ),
+    ] = None,
 ) -> None:
     """Update custom metadata for an item in a run."""
     import json  # noqa: PLC0415
 
+    from aignostics.platform import ConcurrencyConflictError  # noqa: PLC0415
+
     logger.trace("Updating custom metadata for item '{}' in run with ID '{}'", external_id, run_id)
 
     try:
@@ -1271,13 +1321,19 @@ def run_update_item_metadata(
             console.print(f"[error]Error:[/error] Invalid JSON: {e}")
             sys.exit(1)
 
-        Service().application_run_update_item_custom_metadata(run_id, external_id, custom_metadata)
+        Service().application_run_update_item_custom_metadata(
+            run_id, external_id, custom_metadata, custom_metadata_checksum=checksum
+        )
         logger.debug("Updated custom metadata for item '{}' in run with ID '{}'.", external_id, run_id)
         console.print(f"Successfully updated custom metadata for item '{external_id}' in run with ID '{run_id}'.")
     except NotFoundException:
         logger.warning(f"Run with ID '{run_id}' or item '{external_id}' not found.")
         console.print(f"[warning]Warning:[/warning] Run with ID '{run_id}' or item '{external_id}' not found.")
         sys.exit(2)
+    except ConcurrencyConflictError as e:
+        logger.warning("Concurrency conflict updating metadata for item '{}' in run '{}': {}", external_id, run_id, e)
+        console.print(f"[warning]Warning:[/warning] Metadata was modified by another process. Re-read and retry: {e}")
+        sys.exit(3)
     except ValueError as e:
         logger.warning(
             "Run ID '{}' or item external ID '{}' invalid or metadata invalid: {}",
diff --git a/src/aignostics/application/_gui/_page_application_run_describe.py b/src/aignostics/application/_gui/_page_application_run_describe.py
index 926aa767d..cde6bd7af 100644
--- a/src/aignostics/application/_gui/_page_application_run_describe.py
+++ b/src/aignostics/application/_gui/_page_application_run_describe.py
@@ -19,7 +19,15 @@
 )
 from nicegui import run as nicegui_run
 
-from aignostics.platform import ArtifactOutput, ItemOutput, ItemResult, ItemState, Run, RunState
+from aignostics.platform import (
+    ArtifactOutput,
+    ConcurrencyConflictError,
+    ItemOutput,
+    ItemResult,
+    ItemState,
+    Run,
+    RunState,
+)
 from aignostics.third_party.showinfm.showinfm import show_in_file_manager
 from aignostics.utils import GUILocalFilePicker, get_user_data_directory
 
@@ -744,9 +752,15 @@ async def handle_metadata_change(e: Any) -> None:  # noqa: ANN401
                                     Service.application_run_update_custom_metadata_static,
                                     run_id=run_id,
                                     custom_metadata=new_metadata,
+                                    custom_metadata_checksum=run_data.custom_metadata_checksum,
                                 )
                                 ui.notify("Custom metadata updated successfully!", type="positive")
                                 ui.navigate.reload()
+                        except ConcurrencyConflictError:
+                            ui.notify(
+                                "Metadata was modified by another process — reload the page and retry.",
+                                type="warning",
+                            )
                         except Exception as ex:
                             ui.notify(f"Failed to update custom metadata: {ex!s}", type="negative")
 
diff --git a/src/aignostics/application/_service.py b/src/aignostics/application/_service.py
index 315c674f6..0a2423a25 100644
--- a/src/aignostics/application/_service.py
+++ b/src/aignostics/application/_service.py
@@ -23,6 +23,7 @@
     ApplicationSummary,
     ApplicationVersion,
     Client,
+    ConcurrencyConflictError,
     ForbiddenException,
     InputArtifact,
     InputItem,
@@ -1120,21 +1121,31 @@ def application_run_update_custom_metadata(
         self,
         run_id: str,
         custom_metadata: dict[str, Any],
+        *,
+        custom_metadata_checksum: str | None = None,
     ) -> None:
         """Update custom metadata for an existing application run.
 
         Args:
             run_id (str): The ID of the run to update
             custom_metadata (dict[str, Any]): The new custom metadata to attach to the run.
+            custom_metadata_checksum (str | None): Optional checksum for optimistic concurrency
+                control. When provided, the server returns HTTP 412 (and rejects the update) if
+                the metadata was modified since the checksum was read. Pass ``None`` to skip
+                the precondition check.
 
         Raises:
             NotFoundException: If the application run with the given ID is not found.
+            ConcurrencyConflictError: If the checksum precondition failed (HTTP 412).
             ValueError: If the run ID is invalid.
             RuntimeError: If updating the run metadata fails unexpectedly.
         """
         try:
             logger.trace("Updating custom metadata for run with ID '{}'", run_id)
-            self._get_platform_client().run(run_id).update_custom_metadata(custom_metadata)
+            self._get_platform_client().run(run_id).update_custom_metadata(
+                custom_metadata,
+                custom_metadata_checksum=custom_metadata_checksum,
+            )
             logger.trace("Updated custom metadata for run with ID '{}'", run_id)
         except ValueError as e:
             message = f"Failed to update custom metadata for run with ID '{run_id}': ValueError {e}"
@@ -1145,6 +1156,13 @@ def application_run_update_custom_metadata(
             logger.warning(message)
             raise NotFoundException(message) from e
         except ApiException as e:
+            if e.status == HTTPStatus.PRECONDITION_FAILED:
+                message = (
+                    f"Custom metadata for run '{run_id}' was modified since the checksum was read "
+                    f"(optimistic concurrency conflict): {e!s}."
+                )
+                logger.warning(message)
+                raise ConcurrencyConflictError(message) from e
             if e.status == HTTPStatus.UNPROCESSABLE_ENTITY:
                 message = f"Run ID '{run_id}' invalid: {e!s}."
                 logger.warning(message)
@@ -1161,25 +1179,36 @@ def application_run_update_custom_metadata(
     def application_run_update_custom_metadata_static(
         run_id: str,
         custom_metadata: dict[str, Any],
+        *,
+        custom_metadata_checksum: str | None = None,
     ) -> None:
         """Static wrapper for updating custom metadata for an application run.
 
         Args:
             run_id (str): The ID of the run to update
             custom_metadata (dict[str, Any]): The new custom metadata to attach to the run.
+            custom_metadata_checksum (str | None): Optional checksum for optimistic concurrency
+                control. When provided, the server returns HTTP 412 (and rejects the update) if
+                the metadata was modified since the checksum was read. Pass ``None`` to skip
+                the precondition check.
 
         Raises:
             NotFoundException: If the application run with the given ID is not found.
+            ConcurrencyConflictError: If the checksum precondition failed (HTTP 412).
             ValueError: If the run ID is invalid.
             RuntimeError: If updating the run metadata fails unexpectedly.
         """
-        Service().application_run_update_custom_metadata(run_id, custom_metadata)
+        Service().application_run_update_custom_metadata(
+            run_id, custom_metadata, custom_metadata_checksum=custom_metadata_checksum
+        )
 
     def application_run_update_item_custom_metadata(
         self,
         run_id: str,
         external_id: str,
         custom_metadata: dict[str, Any],
+        *,
+        custom_metadata_checksum: str | None = None,
     ) -> None:
         """Update custom metadata for an existing item in an application run.
 
@@ -1187,9 +1216,14 @@ def application_run_update_item_custom_metadata(
             run_id (str): The ID of the run containing the item
             external_id (str): The external ID of the item to update
             custom_metadata (dict[str, Any]): The new custom metadata to attach to the item.
+            custom_metadata_checksum (str | None): Optional checksum for optimistic concurrency
+                control. When provided, the server returns HTTP 412 (and rejects the update) if
+                the metadata was modified since the checksum was read. Pass ``None`` to skip
+                the precondition check.
 
         Raises:
             NotFoundException: If the application run or item with the given IDs is not found.
+            ConcurrencyConflictError: If the checksum precondition failed (HTTP 412).
             ValueError: If the run ID or item external ID is invalid.
             RuntimeError: If updating the item metadata fails unexpectedly.
         """
@@ -1202,6 +1236,7 @@ def application_run_update_item_custom_metadata(
             self._get_platform_client().run(run_id).update_item_custom_metadata(
                 external_id,
                 custom_metadata,
+                custom_metadata_checksum=custom_metadata_checksum,
             )
             logger.trace(
                 "Updated custom metadata for item '{}' in run with ID '{}'",
@@ -1219,6 +1254,13 @@ def application_run_update_item_custom_metadata(
             logger.warning(message)
             raise NotFoundException(message) from e
         except ApiException as e:
+            if e.status == HTTPStatus.PRECONDITION_FAILED:
+                message = (
+                    f"Custom metadata for item '{external_id}' in run '{run_id}' was modified since "
+                    f"the checksum was read (optimistic concurrency conflict): {e!s}."
+                )
+                logger.warning(message)
+                raise ConcurrencyConflictError(message) from e
             if e.status == HTTPStatus.UNPROCESSABLE_ENTITY:
                 message = f"Run ID '{run_id}' or item external ID '{external_id}' invalid: {e!s}."
                 logger.warning(message)
@@ -1236,6 +1278,8 @@ def application_run_update_item_custom_metadata_static(
         run_id: str,
         external_id: str,
         custom_metadata: dict[str, Any],
+        *,
+        custom_metadata_checksum: str | None = None,
     ) -> None:
         """Static wrapper for updating custom metadata for an item in an application run.
 
@@ -1243,13 +1287,20 @@ def application_run_update_item_custom_metadata_static(
             run_id (str): The ID of the run containing the item
             external_id (str): The external ID of the item to update
             custom_metadata (dict[str, Any]): The new custom metadata to attach to the item.
+            custom_metadata_checksum (str | None): Optional checksum for optimistic concurrency
+                control. When provided, the server returns HTTP 412 (and rejects the update) if
+                the metadata was modified since the checksum was read. Pass ``None`` to skip
+                the precondition check.
 
         Raises:
             NotFoundException: If the application run or item with the given IDs is not found.
+            ConcurrencyConflictError: If the checksum precondition failed (HTTP 412).
             ValueError: If the run ID or item external ID is invalid.
             RuntimeError: If updating the item metadata fails unexpectedly.
         """
-        Service().application_run_update_item_custom_metadata(run_id, external_id, custom_metadata)
+        Service().application_run_update_item_custom_metadata(
+            run_id, external_id, custom_metadata, custom_metadata_checksum=custom_metadata_checksum
+        )
 
     def application_run_cancel(self, run_id: str) -> None:
         """Cancel a run by its ID.
diff --git a/src/aignostics/platform/__init__.py b/src/aignostics/platform/__init__.py
index fea6f14f4..5de50f2e9 100644
--- a/src/aignostics/platform/__init__.py
+++ b/src/aignostics/platform/__init__.py
@@ -81,6 +81,7 @@
     TOKEN_URL_STAGING,
     TOKEN_URL_TEST,
 )
+from ._exceptions import ConcurrencyConflictError
 from ._messages import AUTHENTICATION_FAILED, NOT_YET_IMPLEMENTED, UNKNOWN_ENDPOINT_URL
 from ._sdk_metadata import (
     PipelineConfig,
@@ -155,6 +156,7 @@
     "Artifact",
     "ArtifactOutput",
     "Client",
+    "ConcurrencyConflictError",
     "Documents",
     "ForbiddenException",
     "InputArtifact",
diff --git a/src/aignostics/platform/_exceptions.py b/src/aignostics/platform/_exceptions.py
new file mode 100644
index 000000000..abab5bbff
--- /dev/null
+++ b/src/aignostics/platform/_exceptions.py
@@ -0,0 +1,10 @@
+"""Exceptions of platform module."""
+
+
+class ConcurrencyConflictError(ValueError):
+    """Raised when an optimistic concurrency precondition (HTTP 412) fails.
+
+    Subclasses ValueError so existing ``except ValueError`` callers still catch it,
+    while callers that need to distinguish a conflict from a bad-ID error can use
+    ``except ConcurrencyConflictError``.
+    """
diff --git a/src/aignostics/platform/resources/runs.py b/src/aignostics/platform/resources/runs.py
index 508a547bc..c518f8bbd 100644
--- a/src/aignostics/platform/resources/runs.py
+++ b/src/aignostics/platform/resources/runs.py
@@ -596,11 +596,17 @@ def ensure_artifacts_downloaded(
     def update_custom_metadata(
         self,
         custom_metadata: dict[str, Any],
+        *,
+        custom_metadata_checksum: str | None = None,
     ) -> None:
         """Update custom metadata for this application run.
 
         Args:
             custom_metadata (dict[str, Any]): The new custom metadata to attach to the run.
+            custom_metadata_checksum (str | None): Optional checksum for optimistic concurrency
+                control. When provided, the server returns HTTP 412 (and rejects the update) if
+                the metadata was modified since the checksum was read. Pass ``None`` to skip
+                the precondition check.
 
         Raises:
             Exception: If the API request fails.
@@ -615,7 +621,8 @@ def update_custom_metadata(
         self._api.put_run_custom_metadata_v1_runs_run_id_custom_metadata_put(
             self.run_id,
             custom_metadata_update_request=CustomMetadataUpdateRequest(
-                custom_metadata=cast("dict[str, Any]", convert_to_json_serializable(custom_metadata))
+                custom_metadata=cast("dict[str, Any]", convert_to_json_serializable(custom_metadata)),
+                custom_metadata_checksum=custom_metadata_checksum,
             ),
             _request_timeout=settings().run_submit_timeout,
             _headers={"User-Agent": user_agent()},
@@ -626,12 +633,18 @@ def update_item_custom_metadata(
         self,
         external_id: str,
         custom_metadata: dict[str, Any],
+        *,
+        custom_metadata_checksum: str | None = None,
     ) -> None:
         """Update custom metadata for an item in this application run.
 
         Args:
             external_id (str): The external ID of the item.
             custom_metadata (dict[str, Any]): The new custom metadata to attach to the item.
+            custom_metadata_checksum (str | None): Optional checksum for optimistic concurrency
+                control. When provided, the server returns HTTP 412 (and rejects the update) if
+                the metadata was modified since the checksum was read. Pass ``None`` to skip
+                the precondition check.
 
         Raises:
             Exception: If the API request fails.
@@ -647,7 +660,8 @@ def update_item_custom_metadata(
             self.run_id,
             external_id,
             custom_metadata_update_request=CustomMetadataUpdateRequest(
-                custom_metadata=cast("dict[str, Any]", convert_to_json_serializable(custom_metadata))
+                custom_metadata=cast("dict[str, Any]", convert_to_json_serializable(custom_metadata)),
+                custom_metadata_checksum=custom_metadata_checksum,
             ),
             _request_timeout=settings().run_submit_timeout,
             _headers={"User-Agent": user_agent()},
diff --git a/tests/aignostics/application/cli_test.py b/tests/aignostics/application/cli_test.py
index 567e0f29c..1c15233ca 100644
--- a/tests/aignostics/application/cli_test.py
+++ b/tests/aignostics/application/cli_test.py
@@ -67,6 +67,8 @@
 DOCUMENT_MODEL_CARD_PDF = "model_card.pdf"
 DOCUMENT_MISSING_PDF = "missing.pdf"
 APPLICATION_CLI_CLIENT_PATCH_TARGET = "aignostics.application._cli.Client"
+APPLICATION_CLI_SERVICE_PATCH_TARGET = "aignostics.application._cli.Service"
+_TEST_METADATA_JSON = '{"key": "value"}'
 
 # Stub values reused across the document CLI tests.
 DOCUMENT_TEST_FAILURE_MESSAGE = "kaboom"  # canonical exception body for unexpected-failure paths
@@ -957,7 +959,7 @@ def test_cli_run_describe_json_includes_items(runner: CliRunner) -> None:
 
     with (
         patch("aignostics.application._cli.PlatformService.get_user_info", return_value=mock_user_info),
-        patch("aignostics.application._cli.Service") as mock_service_cls,
+        patch(APPLICATION_CLI_SERVICE_PATCH_TARGET) as mock_service_cls,
     ):
         mock_service_cls.return_value.application_run.return_value = mock_run_handle
 
@@ -1224,6 +1226,172 @@ def test_cli_run_update_item_metadata_not_dict(runner: CliRunner) -> None:
     assert "Metadata must be a JSON object" in result.output
 
 
+@pytest.mark.unit
+def test_cli_run_update_metadata_success_with_checksum(runner: CliRunner) -> None:
+    """Check run update-metadata command succeeds and forwards --checksum to the service."""
+    with patch(APPLICATION_CLI_SERVICE_PATCH_TARGET) as mock_service_cls:
+        result = runner.invoke(
+            cli,
+            [
+                "application",
+                "run",
+                "update-metadata",
+                "run-123",
+                _TEST_METADATA_JSON,
+                "--checksum",
+                "abc123",
+            ],
+        )
+    assert result.exit_code == 0
+    assert "Successfully updated" in result.output
+    mock_service_cls.return_value.application_run_update_custom_metadata.assert_called_once_with(
+        "run-123", {"key": "value"}, custom_metadata_checksum="abc123"
+    )
+
+
+@pytest.mark.unit
+def test_cli_run_update_metadata_concurrency_conflict(runner: CliRunner) -> None:
+    """Check run update-metadata exits 3 with a clear message on ConcurrencyConflictError."""
+    from aignostics.platform import ConcurrencyConflictError
+
+    with patch(APPLICATION_CLI_SERVICE_PATCH_TARGET) as mock_service_cls:
+        mock_service_cls.return_value.application_run_update_custom_metadata.side_effect = ConcurrencyConflictError(
+            "stale checksum"
+        )
+        result = runner.invoke(
+            cli,
+            ["application", "run", "update-metadata", "run-123", _TEST_METADATA_JSON, "--checksum", "old"],
+        )
+    assert result.exit_code == 3
+    assert "modified by another process" in result.output
+
+
+@pytest.mark.unit
+def test_cli_run_update_item_metadata_success_with_checksum(runner: CliRunner) -> None:
+    """Check run update-item-metadata command succeeds and forwards --checksum to the service."""
+    with patch(APPLICATION_CLI_SERVICE_PATCH_TARGET) as mock_service_cls:
+        result = runner.invoke(
+            cli,
+            [
+                "application",
+                "run",
+                "update-item-metadata",
+                "run-123",
+                "item-ext-id",
+                _TEST_METADATA_JSON,
+                "--checksum",
+                "abc123",
+            ],
+        )
+    assert result.exit_code == 0
+    assert "Successfully updated" in result.output
+    mock_service_cls.return_value.application_run_update_item_custom_metadata.assert_called_once_with(
+        "run-123", "item-ext-id", {"key": "value"}, custom_metadata_checksum="abc123"
+    )
+
+
+@pytest.mark.unit
+def test_cli_run_update_item_metadata_concurrency_conflict(runner: CliRunner) -> None:
+    """Check run update-item-metadata exits 3 with a clear message on ConcurrencyConflictError."""
+    from aignostics.platform import ConcurrencyConflictError
+
+    with patch(APPLICATION_CLI_SERVICE_PATCH_TARGET) as mock_service_cls:
+        mock_service_cls.return_value.application_run_update_item_custom_metadata.side_effect = (
+            ConcurrencyConflictError("stale checksum")
+        )
+        result = runner.invoke(
+            cli,
+            [
+                "application",
+                "run",
+                "update-item-metadata",
+                "run-123",
+                "item-ext-id",
+                _TEST_METADATA_JSON,
+                "--checksum",
+                "old",
+            ],
+        )
+    assert result.exit_code == 3
+    assert "modified by another process" in result.output
+
+
+@pytest.mark.unit
+def test_cli_run_dump_metadata_default_output(runner: CliRunner) -> None:
+    """dump-metadata default output is bare custom_metadata dict (backward compat)."""
+    mock_run_handle = MagicMock()
+    mock_run_data = MagicMock()
+    mock_run_data.custom_metadata = {"key": "value"}
+    mock_run_data.custom_metadata_checksum = "abc123"
+    mock_run_handle.details.return_value = mock_run_data
+
+    with patch(APPLICATION_CLI_SERVICE_PATCH_TARGET) as mock_service_cls:
+        mock_service_cls.return_value.application_run.return_value = mock_run_handle
+        result = runner.invoke(cli, ["application", "run", "dump-metadata", "run-123"])
+
+    assert result.exit_code == 0
+    assert json.loads(result.output) == {"key": "value"}
+
+
+@pytest.mark.unit
+def test_cli_run_dump_metadata_show_checksum(runner: CliRunner) -> None:
+    """dump-metadata --show-checksum emits wrapped object with custom_metadata_checksum."""
+    mock_run_handle = MagicMock()
+    mock_run_data = MagicMock()
+    mock_run_data.custom_metadata = {"key": "value"}
+    mock_run_data.custom_metadata_checksum = "abc123"
+    mock_run_handle.details.return_value = mock_run_data
+
+    with patch(APPLICATION_CLI_SERVICE_PATCH_TARGET) as mock_service_cls:
+        mock_service_cls.return_value.application_run.return_value = mock_run_handle
+        result = runner.invoke(cli, ["application", "run", "dump-metadata", "run-123", "--show-checksum"])
+
+    assert result.exit_code == 0
+    parsed = json.loads(result.output)
+    assert parsed == {"custom_metadata": {"key": "value"}, "custom_metadata_checksum": "abc123"}
+
+
+@pytest.mark.unit
+def test_cli_run_dump_item_metadata_default_output(runner: CliRunner) -> None:
+    """dump-item-metadata default output is bare custom_metadata dict (backward compat)."""
+    mock_item = MagicMock()
+    mock_item.external_id = "item-ext-id"
+    mock_item.custom_metadata = {"item_key": "item_value"}
+    mock_item.custom_metadata_checksum = "def456"
+
+    mock_run_handle = MagicMock()
+    mock_run_handle.results.return_value = iter([mock_item])
+
+    with patch(APPLICATION_CLI_SERVICE_PATCH_TARGET) as mock_service_cls:
+        mock_service_cls.return_value.application_run.return_value = mock_run_handle
+        result = runner.invoke(cli, ["application", "run", "dump-item-metadata", "run-123", "item-ext-id"])
+
+    assert result.exit_code == 0
+    assert json.loads(result.output) == {"item_key": "item_value"}
+
+
+@pytest.mark.unit
+def test_cli_run_dump_item_metadata_show_checksum(runner: CliRunner) -> None:
+    """dump-item-metadata --show-checksum emits wrapped object with custom_metadata_checksum."""
+    mock_item = MagicMock()
+    mock_item.external_id = "item-ext-id"
+    mock_item.custom_metadata = {"item_key": "item_value"}
+    mock_item.custom_metadata_checksum = "def456"
+
+    mock_run_handle = MagicMock()
+    mock_run_handle.results.return_value = iter([mock_item])
+
+    with patch(APPLICATION_CLI_SERVICE_PATCH_TARGET) as mock_service_cls:
+        mock_service_cls.return_value.application_run.return_value = mock_run_handle
+        result = runner.invoke(
+            cli, ["application", "run", "dump-item-metadata", "run-123", "item-ext-id", "--show-checksum"]
+        )
+
+    assert result.exit_code == 0
+    parsed = json.loads(result.output)
+    assert parsed == {"custom_metadata": {"item_key": "item_value"}, "custom_metadata_checksum": "def456"}
+
+
 @pytest.mark.e2e
 @pytest.mark.timeout(timeout=180)
 @pytest.mark.sequential
diff --git a/tests/aignostics/application/service_test.py b/tests/aignostics/application/service_test.py
index ea023455b..90bf87c13 100644
--- a/tests/aignostics/application/service_test.py
+++ b/tests/aignostics/application/service_test.py
@@ -1,13 +1,14 @@
 """Tests to verify the service functionality of the application module."""
 
 from datetime import UTC, datetime, timedelta
+from http import HTTPStatus
 from unittest.mock import MagicMock, patch
 
 import pytest
 from typer.testing import CliRunner
 
 from aignostics.application import Service as ApplicationService
-from aignostics.platform import NotFoundException, RunData, RunOutput
+from aignostics.platform import ApiException, ConcurrencyConflictError, NotFoundException, RunData, RunOutput
 from tests.constants_test import (
     HETA_APPLICATION_ID,
     HETA_APPLICATION_VERSION,
@@ -511,7 +512,7 @@ def test_application_run_update_custom_metadata_success(mock_get_client: MagicMo
     # Verify the run() method was called with correct run_id
     mock_client.run.assert_called_once_with("run-123")
     # Verify the update_custom_metadata method was called with correct arguments
-    mock_run.update_custom_metadata.assert_called_once_with(custom_metadata)
+    mock_run.update_custom_metadata.assert_called_once_with(custom_metadata, custom_metadata_checksum=None)
 
 
 @pytest.mark.unit
@@ -548,7 +549,9 @@ def test_application_run_update_item_custom_metadata_success(mock_get_client: Ma
     # Verify the run() method was called with correct run_id
     mock_client.run.assert_called_once_with("run-123")
     # Verify the update_item_custom_metadata method was called with correct arguments
-    mock_run.update_item_custom_metadata.assert_called_once_with("item-ext-id", custom_metadata)
+    mock_run.update_item_custom_metadata.assert_called_once_with(
+        "item-ext-id", custom_metadata, custom_metadata_checksum=None
+    )
 
 
 @pytest.mark.unit
@@ -565,3 +568,71 @@ def test_application_run_update_item_custom_metadata_not_found(mock_get_client:
 
     with pytest.raises(NotFoundException, match="not found"):
         service.application_run_update_item_custom_metadata("run-123", "invalid-item-id", {"key": "value"})
+
+
+@pytest.mark.unit
+@patch("aignostics.application._service.Service._get_platform_client")
+def test_application_run_update_custom_metadata_with_checksum(mock_get_client: MagicMock) -> None:
+    """Checksum is forwarded through the service layer to Run.update_custom_metadata."""
+    mock_client = MagicMock()
+    mock_run = MagicMock()
+    mock_client.run.return_value = mock_run
+    mock_get_client.return_value = mock_client
+
+    service = ApplicationService()
+    service.application_run_update_custom_metadata("run-123", {"key": "value"}, custom_metadata_checksum="abc123")
+
+    mock_run.update_custom_metadata.assert_called_once_with({"key": "value"}, custom_metadata_checksum="abc123")
+
+
+@pytest.mark.unit
+@patch("aignostics.application._service.Service._get_platform_client")
+def test_application_run_update_custom_metadata_412_raises_concurrency_error(mock_get_client: MagicMock) -> None:
+    """HTTP 412 from server (optimistic concurrency conflict) raises ConcurrencyConflictError."""
+    mock_client = MagicMock()
+    mock_run = MagicMock()
+    mock_run.update_custom_metadata.side_effect = ApiException(status=HTTPStatus.PRECONDITION_FAILED)
+    mock_client.run.return_value = mock_run
+    mock_get_client.return_value = mock_client
+
+    service = ApplicationService()
+
+    with pytest.raises(ConcurrencyConflictError, match="optimistic concurrency conflict"):
+        service.application_run_update_custom_metadata("run-123", {"key": "value"}, custom_metadata_checksum="stale")
+
+
+@pytest.mark.unit
+@patch("aignostics.application._service.Service._get_platform_client")
+def test_application_run_update_item_custom_metadata_with_checksum(mock_get_client: MagicMock) -> None:
+    """Checksum is forwarded through the service layer to Run.update_item_custom_metadata."""
+    mock_client = MagicMock()
+    mock_run = MagicMock()
+    mock_client.run.return_value = mock_run
+    mock_get_client.return_value = mock_client
+
+    service = ApplicationService()
+    service.application_run_update_item_custom_metadata(
+        "run-123", "item-ext-id", {"key": "value"}, custom_metadata_checksum="abc123"
+    )
+
+    mock_run.update_item_custom_metadata.assert_called_once_with(
+        "item-ext-id", {"key": "value"}, custom_metadata_checksum="abc123"
+    )
+
+
+@pytest.mark.unit
+@patch("aignostics.application._service.Service._get_platform_client")
+def test_application_run_update_item_custom_metadata_412_raises_concurrency_error(mock_get_client: MagicMock) -> None:
+    """HTTP 412 from server raises ConcurrencyConflictError for item metadata update."""
+    mock_client = MagicMock()
+    mock_run = MagicMock()
+    mock_run.update_item_custom_metadata.side_effect = ApiException(status=HTTPStatus.PRECONDITION_FAILED)
+    mock_client.run.return_value = mock_run
+    mock_get_client.return_value = mock_client
+
+    service = ApplicationService()
+
+    with pytest.raises(ConcurrencyConflictError, match="optimistic concurrency conflict"):
+        service.application_run_update_item_custom_metadata(
+            "run-123", "item-ext-id", {"key": "value"}, custom_metadata_checksum="stale"
+        )
diff --git a/tests/aignostics/application/utils_test.py b/tests/aignostics/application/utils_test.py
index 1b30a006c..4bc165f75 100644
--- a/tests/aignostics/application/utils_test.py
+++ b/tests/aignostics/application/utils_test.py
@@ -1187,3 +1187,50 @@ def test_validate_scheduling_constraints_due_date_equal_deadline() -> None:
     same_time = (datetime.now(tz=UTC) + timedelta(hours=2)).isoformat()
     with pytest.raises(ValueError, match=r"due_date must be before deadline"):
         validate_scheduling_constraints(same_time, same_time)
+
+
+# Tests for assert_parquet_geojson_parity (conftest helper)
+
+
+@pytest.mark.unit
+def test_assert_parquet_geojson_parity_row_count_engine_agnostic(tmp_path: Path) -> None:
+    """Verify assert_parquet_geojson_parity counts rows correctly regardless of pyarrow version.
+
+    Guards the columns=['geometry'] fix: columns=[] returns 0 rows under pyarrow 23+
+    (the engine used on Python 3.14), which caused false-passing count assertions.
+    """
+    import json
+
+    import pandas as pd
+    import shapely.geometry
+    import shapely.wkb
+
+    from tests.conftest import assert_parquet_geojson_parity
+
+    n_rows = 3
+    # Build synthetic WKB geometries (simple boxes) for the parquet file
+    polygons = [shapely.geometry.box(i, i, i + 1, i + 1) for i in range(n_rows)]
+    wkb_bytes = [shapely.wkb.dumps(p) for p in polygons]
+
+    # Write the cell_classification parquet with a real geometry column
+    parquet_path = tmp_path / "cell_classification_parquet_polygons.parquet"
+    pd.DataFrame({"geometry": wkb_bytes}).to_parquet(parquet_path)
+
+    # Write matching GeoJSON (same polygons → count parity, area parity)
+    geojson_path = tmp_path / "cell_classification_geojson_polygons.json"
+    features = [{"type": "Feature", "geometry": shapely.geometry.mapping(p), "properties": {}} for p in polygons]
+    geojson_path.write_text(json.dumps({"type": "FeatureCollection", "features": features}))
+
+    # Stub out tissue_qc and tissue_segmentation pairs so the area loop is skipped
+    for stub_parquet, stub_json in [
+        ("tissue_qc_parquet_polygons.parquet", "tissue_qc_geojson_polygons.json"),
+        ("tissue_segmentation_parquet_polygons.parquet", "tissue_segmentation_geojson_polygons.json"),
+    ]:
+        stub_wkb = [shapely.wkb.dumps(shapely.geometry.box(0, 0, 1, 1))]
+        pd.DataFrame({"geometry": stub_wkb}).to_parquet(tmp_path / stub_parquet)
+        stub_geom = shapely.geometry.mapping(shapely.geometry.box(0, 0, 1, 1))
+        stub_features = [{"type": "Feature", "geometry": stub_geom, "properties": {}}]
+        (tmp_path / stub_json).write_text(json.dumps({"type": "FeatureCollection", "features": stub_features}))
+
+    # Must not raise — if columns=[] bug regresses, parquet_count would be 0 != 3
+    assert_parquet_geojson_parity(tmp_path)
diff --git a/tests/conftest.py b/tests/conftest.py
index 648f8d166..a9aa95158 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -370,7 +370,8 @@ def assert_parquet_geojson_parity(results_dir: Path) -> None:
             f"{parquet_name} ({parquet_area:.2f}) and {geojson_name} ({geojson_area:.2f})"
         )
 
-    parquet_count = len(pd.read_parquet(results_dir / "cell_classification_parquet_polygons.parquet", columns=[]))
+    cell_parquet = results_dir / "cell_classification_parquet_polygons.parquet"
+    parquet_count = len(pd.read_parquet(cell_parquet, columns=["geometry"]))
     with (results_dir / "cell_classification_geojson_polygons.json").open("rb") as f:
         geojson_count = sum(1 for _ in ijson.items(f, "features.item"))
     delta = abs(parquet_count - geojson_count)
diff --git a/tests/constants_test.py b/tests/constants_test.py
index 3fc1c2d35..c4a1af951 100644
--- a/tests/constants_test.py
+++ b/tests/constants_test.py
@@ -52,15 +52,6 @@
 SPOT_3_WIDTH = 4016
 SPOT_3_HEIGHT = 3952
 
-SPOT_4_GS_URL = (
-    "gs://aignostics-platform-ext-a4f7e9/python-sdk-tests/he-tme/slides/9375e3ed-28d2-4cf3-9fb9-8df9d11a6627.tiff"
-)
-SPOT_4_FILENAME = "9375e3ed-28d2-4cf3-9fb9-8df9d11a6627.tiff"
-SPOT_4_CRC32C = "9l3NNQ=="
-SPOT_4_FILESIZE = 14681750
-SPOT_4_RESOLUTION_MPP = 0.46499982
-SPOT_4_WIDTH = 3728
-SPOT_4_HEIGHT = 3640
 
 # To update file sizes: the tests print every file's actual size before asserting. Run with
 # -s to see them, then paste the printed byte values as the second element of each tuple.