diff --git a/CHANGES/+container-build-error-sanitization.bugfix b/CHANGES/+container-build-error-sanitization.bugfix
new file mode 100644
index 000000000..baa3c23cf
--- /dev/null
+++ b/CHANGES/+container-build-error-sanitization.bugfix
@@ -0,0 +1 @@
+Raised a proper ``PulpException`` subclass instead of a bare ``Exception`` when building or pushing an OCI image fails, so the error is not sanitized away by pulpcore in a future release.
diff --git a/CHANGES/+task-resource-not-found.bugfix b/CHANGES/+task-resource-not-found.bugfix
new file mode 100644
index 000000000..26cc5156a
--- /dev/null
+++ b/CHANGES/+task-resource-not-found.bugfix
@@ -0,0 +1 @@
+Raised a proper ``PulpException`` subclass instead of a bare Django ``DoesNotExist`` when a task's referenced object (repository, remote, manifest, signing service, etc.) no longer exists, so the error is not sanitized away by pulpcore in a future release.
diff --git a/pulp_container/app/exceptions.py b/pulp_container/app/exceptions.py
index d8946477d..6522b3e34 100644
--- a/pulp_container/app/exceptions.py
+++ b/pulp_container/app/exceptions.py
@@ -1,5 +1,7 @@
 from rest_framework.exceptions import APIException, NotFound, ParseError
 
+from pulpcore.plugin.exceptions import PulpException
+
 
 class BadGateway(APIException):
     status_code = 502
@@ -162,6 +164,39 @@ def __init__(self, digest):
         )
 
 
+class TaskResourceNotFound(PulpException):
+    """Exception to signal that a resource a task depends on no longer exists.
+
+    Tasks look up their arguments' referenced objects by pk. If that object was
+    deleted between dispatch and execution (e.g. by a racing delete), a bare Django
+    DoesNotExist is not a PulpException, which pulpcore's task executor logs as
+    deprecated and will sanitize away in a future release. Raise this instead so the
+    real reason is preserved on the task result.
+    """
+
+    error_code = "PLP_CONTAINER_TASK_RESOURCE_MISSING"
+
+    def __init__(self, message):
+        """Initialize the exception with a description of the missing resource."""
+        self.message = message
+
+    def __str__(self):
+        return self.message
+
+
+class ContainerBuildError(PulpException):
+    """Exception to signal that building or pushing an OCI image failed."""
+
+    error_code = "PLP_CONTAINER_BUILD_0001"
+
+    def __init__(self, message):
+        """Initialize the exception with the decoded stderr of the failed podman command."""
+        self.message = message
+
+    def __str__(self):
+        return self.message
+
+
 class InvalidRequest(ParseError):
     """An exception to render an HTTP 400 response."""
 
diff --git a/pulp_container/app/tasks/builder.py b/pulp_container/app/tasks/builder.py
index 155abbc22..1296d4b9d 100644
--- a/pulp_container/app/tasks/builder.py
+++ b/pulp_container/app/tasks/builder.py
@@ -13,6 +13,7 @@
 )
 from pulpcore.plugin.util import get_domain
 
+from pulp_container.app.exceptions import ContainerBuildError
 from pulp_container.app.models import (
     Blob,
     BlobManifest,
@@ -20,7 +21,7 @@
     Manifest,
     Tag,
 )
-from pulp_container.app.utils import calculate_digest
+from pulp_container.app.utils import calculate_digest, get_task_resource_or_raise
 from pulp_container.constants import MEDIA_TYPE
 
 
@@ -138,9 +139,11 @@ def build_image(
         raise RuntimeError("Neither a name nor temporary file for the Containerfile was specified.")
 
     if containerfile_tempfile_pk:
-        containerfile_artifact = PulpTemporaryFile.objects.get(pk=containerfile_tempfile_pk)
+        containerfile_artifact = get_task_resource_or_raise(
+            PulpTemporaryFile, pk=containerfile_tempfile_pk
+        )
 
-    repository = ContainerRepository.objects.get(pk=repository_pk)
+    repository = get_task_resource_or_raise(ContainerRepository, pk=repository_pk)
     name = str(uuid4())
     with tempfile.TemporaryDirectory(dir=".") as working_directory:
         working_directory = os.path.abspath(working_directory)
@@ -179,7 +182,7 @@ def build_image(
             stderr=subprocess.PIPE,
         )
         if bud_cp.returncode != 0:
-            raise Exception(bud_cp.stderr)
+            raise ContainerBuildError(bud_cp.stderr.decode())
         image_dir = os.path.join(working_directory, "image")
         os.makedirs(image_dir, exist_ok=True)
         push_cp = subprocess.run(
@@ -188,7 +191,7 @@ def build_image(
             stderr=subprocess.PIPE,
         )
         if push_cp.returncode != 0:
-            raise Exception(push_cp.stderr)
+            raise ContainerBuildError(push_cp.stderr.decode())
         repository_version = add_image_from_directory_to_repository(image_dir, repository, tag)
         if isinstance(containerfile_artifact, PulpTemporaryFile):
             containerfile_artifact.delete()
@@ -229,15 +232,15 @@ def build_image_from_containerfile(
         image and tag.
 
     """
-    containerfile = Artifact.objects.get(pk=containerfile_pk)
-    repository = ContainerRepository.objects.get(pk=repository_pk)
+    containerfile = get_task_resource_or_raise(Artifact, pk=containerfile_pk)
+    repository = get_task_resource_or_raise(ContainerRepository, pk=repository_pk)
     name = str(uuid4())
     with tempfile.TemporaryDirectory(dir=".") as working_directory:
         working_directory = os.path.abspath(working_directory)
         context_path = os.path.join(working_directory, "context")
         os.makedirs(context_path, exist_ok=True)
         for key, val in artifacts.items():
-            artifact = Artifact.objects.get(pk=key)
+            artifact = get_task_resource_or_raise(Artifact, pk=key)
             dest_path = os.path.join(context_path, val)
             dirs = os.path.split(dest_path)[0]
             if dirs:
@@ -265,7 +268,7 @@ def build_image_from_containerfile(
             stderr=subprocess.PIPE,
         )
         if bud_cp.returncode != 0:
-            raise Exception(bud_cp.stderr)
+            raise ContainerBuildError(bud_cp.stderr.decode())
         image_dir = os.path.join(working_directory, "image")
         os.makedirs(image_dir, exist_ok=True)
         push_cp = subprocess.run(
@@ -274,7 +277,7 @@ def build_image_from_containerfile(
             stderr=subprocess.PIPE,
         )
         if push_cp.returncode != 0:
-            raise Exception(push_cp.stderr)
+            raise ContainerBuildError(push_cp.stderr.decode())
         repository_version = add_image_from_directory_to_repository(image_dir, repository, tag)
 
     return repository_version
diff --git a/pulp_container/app/tasks/download_image_data.py b/pulp_container/app/tasks/download_image_data.py
index 3eb712cb8..f0d456286 100644
--- a/pulp_container/app/tasks/download_image_data.py
+++ b/pulp_container/app/tasks/download_image_data.py
@@ -7,7 +7,7 @@
 from pulpcore.plugin.tasking import add_and_remove
 
 from pulp_container.app.models import ContainerRemote, ContainerRepository, Tag
-from pulp_container.app.utils import determine_media_type_from_json
+from pulp_container.app.utils import determine_media_type_from_json, get_task_resource_or_raise
 from pulp_container.constants import MEDIA_TYPE
 
 from .sync_stages import ContainerFirstStage
@@ -21,8 +21,8 @@ async def aadd_and_remove(*args, **kwargs):
 
 
 def download_image_data(repository_pk, remote_pk, raw_text_manifest_data, tag_name=None):
-    repository = ContainerRepository.objects.get(pk=repository_pk)
-    remote = ContainerRemote.objects.get(pk=remote_pk)
+    repository = get_task_resource_or_raise(ContainerRepository, pk=repository_pk)
+    remote = get_task_resource_or_raise(ContainerRemote, pk=remote_pk)
     log.info("Pulling cache: repository={r} remote={p}".format(r=repository.name, p=remote.name))
     first_stage = ContainerPullThroughFirstStage(remote, raw_text_manifest_data, tag_name)
     dv = ContainerDeclarativeVersion(first_stage, repository)
diff --git a/pulp_container/app/tasks/recursive_add.py b/pulp_container/app/tasks/recursive_add.py
index da0254bf7..f846ad0d3 100644
--- a/pulp_container/app/tasks/recursive_add.py
+++ b/pulp_container/app/tasks/recursive_add.py
@@ -5,6 +5,7 @@
     Manifest,
     Tag,
 )
+from pulp_container.app.utils import get_task_resource_or_raise
 
 
 def recursive_add_content(repository_pk, content_units):
@@ -22,7 +23,7 @@ def recursive_add_content(repository_pk, content_units):
             should be added to the previous Repository Version for this Repository.
 
     """
-    repository = ContainerRepository.objects.get(pk=repository_pk)
+    repository = get_task_resource_or_raise(ContainerRepository, pk=repository_pk)
 
     tags_to_add = Tag.objects.filter(pk__in=content_units)
 
diff --git a/pulp_container/app/tasks/recursive_remove.py b/pulp_container/app/tasks/recursive_remove.py
index 72835e227..7315cf408 100644
--- a/pulp_container/app/tasks/recursive_remove.py
+++ b/pulp_container/app/tasks/recursive_remove.py
@@ -9,6 +9,7 @@
     ManifestSignature,
     Tag,
 )
+from pulp_container.app.utils import get_task_resource_or_raise
 
 
 def recursive_remove_content(repository_pk, content_units):
@@ -36,7 +37,7 @@ def recursive_remove_content(repository_pk, content_units):
             should be removed from the Repository.
 
     """
-    repository = Repository.objects.get(pk=repository_pk).cast()
+    repository = get_task_resource_or_raise(Repository, pk=repository_pk).cast()
     latest_version = repository.latest_version()
     latest_content = latest_version.content.all() if latest_version else Content.objects.none()
     if "*" in content_units:
diff --git a/pulp_container/app/tasks/sign.py b/pulp_container/app/tasks/sign.py
index 74a66858d..b2c666b88 100644
--- a/pulp_container/app/tasks/sign.py
+++ b/pulp_container/app/tasks/sign.py
@@ -13,7 +13,7 @@
     ManifestSigningService,
     Tag,
 )
-from pulp_container.app.utils import extract_data_from_signature
+from pulp_container.app.utils import extract_data_from_signature, get_task_resource_or_raise
 from pulp_container.constants import (
     MANIFEST_MEDIA_TYPES,
     SIGNATURE_TYPE,
@@ -41,7 +41,7 @@ def sign(repository_pk, signing_service_pk, reference, tags_list=None):
                           should be signed.
 
     """
-    repository = Repository.objects.get(pk=repository_pk).cast()
+    repository = get_task_resource_or_raise(Repository, pk=repository_pk).cast()
     latest_version = repository.latest_version()
     if tags_list:
         latest_repo_content_tags = latest_version.content.filter(
@@ -55,7 +55,7 @@ def sign(repository_pk, signing_service_pk, reference, tags_list=None):
         .select_related("tagged_manifest")
         .exclude(Q(name__endswith=".sig") | Q(name__endswith=".att") | Q(name__endswith=".sbom"))
     )
-    signing_service = ManifestSigningService.objects.get(pk=signing_service_pk)
+    signing_service = get_task_resource_or_raise(ManifestSigningService, pk=signing_service_pk)
 
     async def sign_manifests():
         added_signatures = []
diff --git a/pulp_container/app/tasks/synchronize.py b/pulp_container/app/tasks/synchronize.py
index aa2ffa051..eab490485 100644
--- a/pulp_container/app/tasks/synchronize.py
+++ b/pulp_container/app/tasks/synchronize.py
@@ -12,6 +12,7 @@
 )
 
 from pulp_container.app.models import ContainerRemote, ContainerRepository
+from pulp_container.app.utils import get_task_resource_or_raise
 
 from .sync_stages import ContainerContentSaver, ContainerFirstStage
 
@@ -34,8 +35,8 @@ def synchronize(remote_pk, repository_pk, mirror, signed_only):
         ValueError: If the remote does not specify a URL to sync
 
     """
-    remote = ContainerRemote.objects.get(pk=remote_pk)
-    repository = ContainerRepository.objects.get(pk=repository_pk)
+    remote = get_task_resource_or_raise(ContainerRemote, pk=remote_pk)
+    repository = get_task_resource_or_raise(ContainerRepository, pk=repository_pk)
     log.info("Synchronizing: repository={r} remote={p}".format(r=repository.name, p=remote.name))
     first_stage = ContainerFirstStage(remote, signed_only)
     dv = ContainerDeclarativeVersion(first_stage, repository, mirror)
diff --git a/pulp_container/app/tasks/tag.py b/pulp_container/app/tasks/tag.py
index 0853e01b4..eee80d122 100644
--- a/pulp_container/app/tasks/tag.py
+++ b/pulp_container/app/tasks/tag.py
@@ -2,6 +2,7 @@
 from pulpcore.plugin.util import get_domain
 
 from pulp_container.app.models import Manifest, Tag
+from pulp_container.app.utils import get_task_resource_or_raise
 
 
 def tag_image(manifest_pk, tag, repository_pk):
@@ -14,9 +15,9 @@ def tag_image(manifest_pk, tag, repository_pk):
     a new repository version when a manifest contains a digest which is not equal to the
     digest passed with POST request.
     """
-    manifest = Manifest.objects.get(pk=manifest_pk)
+    manifest = get_task_resource_or_raise(Manifest, pk=manifest_pk)
 
-    repository = Repository.objects.get(pk=repository_pk).cast()
+    repository = get_task_resource_or_raise(Repository, pk=repository_pk).cast()
     latest_version = repository.latest_version()
 
     tags_to_remove = Tag.objects.filter(pk__in=latest_version.content.all(), name=tag).exclude(
diff --git a/pulp_container/app/tasks/untag.py b/pulp_container/app/tasks/untag.py
index 68cde6834..023d4ee96 100644
--- a/pulp_container/app/tasks/untag.py
+++ b/pulp_container/app/tasks/untag.py
@@ -1,13 +1,14 @@
 from pulpcore.plugin.models import Repository
 
 from pulp_container.app.models import Tag
+from pulp_container.app.utils import get_task_resource_or_raise
 
 
 def untag_image(tag, repository_pk):
     """
     Create a new repository version without a specified manifest's tag name.
     """
-    repository = Repository.objects.get(pk=repository_pk).cast()
+    repository = get_task_resource_or_raise(Repository, pk=repository_pk).cast()
     latest_version = repository.latest_version()
 
     tags_in_latest_repository = latest_version.content.filter(pulp_type=Tag.get_pulp_type())
diff --git a/pulp_container/app/utils.py b/pulp_container/app/utils.py
index d578bd62e..8e1e808fa 100644
--- a/pulp_container/app/utils.py
+++ b/pulp_container/app/utils.py
@@ -16,7 +16,7 @@
 from pulpcore.plugin.models import Artifact, Task
 from pulpcore.plugin.util import get_domain
 
-from pulp_container.app.exceptions import ManifestInvalid
+from pulp_container.app.exceptions import ManifestInvalid, TaskResourceNotFound
 from pulp_container.app.json_schemas import (
     DOCKER_MANIFEST_LIST_V2_SCHEMA,
     DOCKER_MANIFEST_V1_SCHEMA,
@@ -42,6 +42,24 @@ def get_full_path(base_path, pulp_domain=None):
     return base_path
 
 
+def get_task_resource_or_raise(model, **kwargs):
+    """
+    Fetch a single model instance by the given lookup, or raise TaskResourceNotFound.
+
+    Intended for the object lookups at the top of task functions (repository, remote,
+    manifest, etc. by pk), where a bare DoesNotExist would otherwise propagate as a
+    non-PulpException and get flagged by pulpcore as deprecated/sanitized-away.
+    """
+    try:
+        return model.objects.get(**kwargs)
+    except model.DoesNotExist:
+        lookup = ", ".join(f"{key}={value}" for key, value in kwargs.items())
+        raise TaskResourceNotFound(
+            f"{model.__name__} matching {{{lookup}}} does not exist. It may have been "
+            "deleted after this task was dispatched."
+        ) from None
+
+
 def get_accepted_media_types(headers):
     """
     Returns a list of media types from the Accept headers.
diff --git a/pulp_container/tests/functional/api/test_domains.py b/pulp_container/tests/functional/api/test_domains.py
index e90cb32db..d8bda092b 100644
--- a/pulp_container/tests/functional/api/test_domains.py
+++ b/pulp_container/tests/functional/api/test_domains.py
@@ -11,7 +11,7 @@
 
 
 @pytest.fixture
-def cdomain_factory(domain_factory, pulpcore_bindings):
+def cdomain_factory(domain_factory, pulpcore_bindings, container_bindings):
     domains = []
 
     def _domain_factory(*args, **kwargs):
@@ -22,6 +22,14 @@ def _domain_factory(*args, **kwargs):
     yield _domain_factory
 
     for domain in domains:
+        # ContainerNamespace has a PROTECT FK to Domain, so it must be cleared before
+        # domain_factory's own teardown deletes the domain, or that delete raises
+        # ProtectedError. Some tests already clean their own namespace up via
+        # add_to_cleanup; re-querying here is a no-op for those and a safety net for
+        # any test (e.g. one that creates multiple distributions) that doesn't.
+        namespaces = container_bindings.PulpContainerNamespacesApi.list(pulp_domain=domain.name)
+        for namespace in namespaces.results:
+            container_bindings.PulpContainerNamespacesApi.delete(namespace.pulp_href)
         guards = pulpcore_bindings.ContentguardsContentRedirectApi.list(pulp_domain=domain.name)
         for guard in guards.results:
             pulpcore_bindings.ContentguardsContentRedirectApi.delete(guard.pulp_href)