diff --git a/CHANGES/+alias-tags.feature b/CHANGES/+alias-tags.feature
new file mode 100644
index 000000000..565226d50
--- /dev/null
+++ b/CHANGES/+alias-tags.feature
@@ -0,0 +1 @@
+Added named alias support for digest-based sync. Entries in `includes` using the format `name=sha256:digest` create human-readable tags alongside the digest, enabling friendly image references in disconnected environments.
diff --git a/CHANGES/2474.feature b/CHANGES/2474.feature
new file mode 100644
index 000000000..8f97ff9c1
--- /dev/null
+++ b/CHANGES/2474.feature
@@ -0,0 +1 @@
+Added tag list bypass optimization for digest-based sync. When all entries in a remote's `includes` field are sha256 digests, the sync pipeline skips the expensive `/tags/list` enumeration and fetches manifests directly by digest.
diff --git a/pulp_container/app/tasks/sync_stages.py b/pulp_container/app/tasks/sync_stages.py
index d24e62bb4..50a73caaa 100644
--- a/pulp_container/app/tasks/sync_stages.py
+++ b/pulp_container/app/tasks/sync_stages.py
@@ -3,6 +3,7 @@
 import hashlib
 import json
 import logging
+import re
 from collections import defaultdict
 from urllib.parse import urljoin, urlparse, urlunparse
 
@@ -52,12 +53,13 @@ class ContainerFirstStage(Stage):
 
     """
 
-    def __init__(self, remote, signed_only):
+    def __init__(self, remote, signed_only, mirror=False):
         """Initialize the stage."""
         super().__init__()
         self.remote = remote
         self.deferred_download = self.remote.policy != Remote.IMMEDIATE
         self.signed_only = signed_only
+        self.mirror = mirror
 
         self.tag_dcs = []
         self.manifest_list_dcs = []
@@ -66,6 +68,9 @@ def __init__(self, remote, signed_only):
         self._synced_digests = defaultdict(list)
         self._full_tag_list = []
         self._cosign_tags = []
+        self._bypass_taglist = False
+        self._probed_cosign_digests = set()
+        self._cosign_probe_semaphore = asyncio.Semaphore(20)
 
     async def _download_manifest_data(self, manifest_url):
         downloader = self.remote.get_downloader(url=manifest_url)
@@ -116,12 +121,127 @@ async def _check_for_existing_manifest(self, head_manifest_task):
 
         return content_data, raw_text_data, response, original_reference
 
+    @staticmethod
+    def _parse_include_entry(entry):
+        """
+        Parse an includes entry into (digest, alias_name).
+
+        Supports two formats:
+        - "sha256:abc123..." — plain digest, no alias
+        - "name=sha256:abc123..." — named alias that creates a local tag
+
+        Returns (digest, alias) where alias is None for plain digests,
+        or (None, None) if the entry is not a valid digest reference.
+        """
+        if "=" in entry and not entry.startswith("sha256:"):
+            name, _, digest = entry.partition("=")
+            if re.fullmatch(r"sha256:[0-9a-f]{64}", digest):
+                return digest, name
+            return None, None
+        if re.fullmatch(r"sha256:[0-9a-f]{64}", entry):
+            return entry, None
+        return None, None
+
+    def _can_bypass_taglist(self):
+        """
+        Check if we can safely bypass /tags/list enumeration.
+
+        Returns True only if all includes are explicit sha256 digests
+        (or name=sha256:digest aliases) and we are not in mirror mode.
+        """
+        includes = self.remote.includes or []
+
+        if not includes:
+            return False
+
+        if self.mirror:
+            return False
+
+        return all(self._parse_include_entry(entry)[0] is not None for entry in includes)
+
+    async def _tag_exists(self, tag_name):
+        """Check if a tag exists upstream via lightweight HEAD request."""
+        relative_url = "/v2/{name}/manifests/{tag}".format(
+            name=self.remote.namespaced_upstream_name, tag=tag_name
+        )
+        manifest_url = urljoin(self.remote.url, relative_url)
+        downloader = self.remote.get_downloader(url=manifest_url)
+
+        try:
+            await downloader.run(extra_data={"headers": V2_ACCEPT_HEADERS, "http_method": "head"})
+            return True
+        except aiohttp.ClientResponseError as exc:
+            if exc.status == 404:
+                return False
+            raise
+
+    async def _ensure_cosign_tags(self, digests):
+        """
+        Populate _cosign_tags by probing for cosign companion tags via HEAD requests.
+
+        Only used when _bypass_taglist is True. Results are cached per digest
+        so repeated calls (e.g. from _has_cosign_signature and then the post-sync
+        companion pass) do not re-probe.
+        """
+        if not self._bypass_taglist:
+            return
+
+        pending = sorted(set(digests) - self._probed_cosign_digests)
+        candidates = []
+
+        for digest in pending:
+            if not digest.startswith("sha256:"):
+                continue
+
+            stem = digest.replace("sha256:", "sha256-", 1)
+            candidates.extend(f"{stem}{suffix}" for suffix in (*COSIGN_TAG_SUFFIXES, ""))
+
+        candidates = filter_resources(candidates, [], self.remote.excludes or [])
+
+        async def probe(tag):
+            async with self._cosign_probe_semaphore:
+                return tag if await self._tag_exists(tag) else None
+
+        results = await asyncio.gather(*(probe(tag) for tag in candidates))
+        self._cosign_tags.extend(tag for tag in results if tag is not None)
+        self._probed_cosign_digests.update(pending)
+
     async def run(self):
         """
         ContainerFirstStage.
         """
         signature_source = await self.get_signature_source()
 
+        self._bypass_taglist = self._can_bypass_taglist()
+        if self._bypass_taglist:
+            digest_list = []
+            self._tag_aliases = {}
+            for entry in self.remote.includes:
+                digest, alias = self._parse_include_entry(entry)
+                digest_list.append(digest)
+                if alias:
+                    self._tag_aliases[digest] = alias
+
+            log.info(
+                "Bypassing /tags/list enumeration - syncing %d explicit digests directly",
+                len(digest_list),
+            )
+            await self._process_manifests(digest_list, signature_source, "Processing Manifests")
+
+            await self._ensure_cosign_tags(list(self._synced_digests))
+            companion_tags = self._find_cosign_companion_tags()
+            if companion_tags:
+                log.info(
+                    "Found %d cosign companion tag(s) for synced manifests",
+                    len(companion_tags),
+                )
+                await self._process_manifests(
+                    companion_tags,
+                    signature_source,
+                    "Processing Cosign Companion Tags",
+                )
+            return
+
         async with ProgressReport(
             message="Downloading tag list", code="sync.downloading.tag_list", total=1
         ) as pb:
@@ -216,6 +336,11 @@ async def _process_manifests(self, manifests, signature_source, msg):
                 if is_tag:
                     tag_dc = DeclarativeContent(Tag(name=manifest_ref))
 
+                alias_name = getattr(self, "_tag_aliases", {}).get(manifest_ref)
+                alias_dc = None
+                if alias_name:
+                    alias_dc = DeclarativeContent(Tag(name=alias_name))
+
                 if media_type in (MEDIA_TYPE.MANIFEST_LIST, MEDIA_TYPE.INDEX_OCI):
                     list_dc = self.create_manifest_list(
                         content_data, raw_text_data, media_type, digest=digest
@@ -264,6 +389,9 @@ async def _process_manifests(self, manifests, signature_source, msg):
                         if is_tag:
                             tag_dc.extra_data["tagged_manifest_dc"] = list_dc
                             self.tag_dcs.append(tag_dc)
+                        if alias_dc:
+                            alias_dc.extra_data["tagged_manifest_dc"] = list_dc
+                            self.tag_dcs.append(alias_dc)
 
                 else:
                     # Simple tagged manifest
@@ -280,6 +408,9 @@ async def _process_manifests(self, manifests, signature_source, msg):
                     if is_tag:
                         tag_dc.extra_data["tagged_manifest_dc"] = man_dc
                         self.tag_dcs.append(tag_dc)
+                    if alias_dc:
+                        alias_dc.extra_data["tagged_manifest_dc"] = man_dc
+                        self.tag_dcs.append(alias_dc)
                     self.manifest_dcs.append(man_dc)
 
                 # Count the skipped tasks as parsed too.
@@ -300,6 +431,7 @@ async def _process_manifests(self, manifests, signature_source, msg):
 
     async def _has_cosign_signature(self, digest):
         """Check if a digest has a cosign signature."""
+        await self._ensure_cosign_tags([digest])
         cosign_digest = digest.replace("sha256:", "sha256-")
         if f"{cosign_digest}.sig" in self._cosign_tags:
             return True
diff --git a/pulp_container/app/tasks/synchronize.py b/pulp_container/app/tasks/synchronize.py
index aa2ffa051..9a3cad78d 100644
--- a/pulp_container/app/tasks/synchronize.py
+++ b/pulp_container/app/tasks/synchronize.py
@@ -37,7 +37,7 @@ def synchronize(remote_pk, repository_pk, mirror, signed_only):
     remote = ContainerRemote.objects.get(pk=remote_pk)
     repository = ContainerRepository.objects.get(pk=repository_pk)
     log.info("Synchronizing: repository={r} remote={p}".format(r=repository.name, p=remote.name))
-    first_stage = ContainerFirstStage(remote, signed_only)
+    first_stage = ContainerFirstStage(remote, signed_only, mirror=mirror)
     dv = ContainerDeclarativeVersion(first_stage, repository, mirror)
     rv = dv.create()
     if rv:
diff --git a/pulp_container/tests/unit/test_sync_stages.py b/pulp_container/tests/unit/test_sync_stages.py
index 835347cb7..3f50fd976 100644
--- a/pulp_container/tests/unit/test_sync_stages.py
+++ b/pulp_container/tests/unit/test_sync_stages.py
@@ -1,8 +1,10 @@
-"""Unit tests for cosign companion tag helpers on ContainerFirstStage."""
+"""Unit tests for ContainerFirstStage: cosign companion tag helpers and tag list bypass."""
 
 import unittest
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import aiohttp
+
 from pulp_container.app.tasks.sync_stages import COSIGN_TAG_SUFFIXES, ContainerFirstStage
 from pulp_container.constants import MEDIA_TYPE
 
@@ -98,8 +100,16 @@ async def test_has_cosign_signature_true_when_sig_tag_present(self):
         cosign_key = digest.replace("sha256:", "sha256-")
         self.stage._cosign_tags = [f"{cosign_key}.sig"]
 
+        self.stage._ensure_cosign_tags = AsyncMock()
         self.assertTrue(await self.stage._has_cosign_signature(digest))
-        self.stage.remote.get_downloader.assert_not_called()
+
+    async def test_has_cosign_signature_calls_ensure_cosign_tags(self):
+        _, digest = _bare_cosign_digest()
+        self.stage._cosign_tags = []
+        self.stage._ensure_cosign_tags = AsyncMock()
+
+        await self.stage._has_cosign_signature(digest)
+        self.stage._ensure_cosign_tags.assert_awaited_once_with([digest])
 
     async def test_has_cosign_signature_true_after_fetching_v3_index(self):
         tag, digest = _bare_cosign_digest()
@@ -115,6 +125,7 @@ async def test_has_cosign_signature_true_after_fetching_v3_index(self):
         mock_response = MagicMock()
         mock_response.url = f"https://registry.example/v2/foo/manifests/{tag}"
 
+        self.stage._ensure_cosign_tags = AsyncMock()
         self.stage._download_manifest_data = AsyncMock(
             return_value=(content_data, raw, mock_response)
         )
@@ -136,6 +147,7 @@ async def test_has_cosign_signature_false_when_bare_tag_not_companion(self):
         mock_response = MagicMock()
         mock_response.url = f"https://registry.example/v2/foo/manifests/{tag}"
 
+        self.stage._ensure_cosign_tags = AsyncMock()
         self.stage._download_manifest_data = AsyncMock(
             return_value=(content_data, raw, mock_response)
         )
@@ -149,8 +161,284 @@ async def test_has_cosign_signature_false_when_bare_tag_not_companion(self):
     async def test_has_cosign_signature_false_when_no_cosign_tags(self):
         _, digest = _bare_cosign_digest()
         self.stage._cosign_tags = []
+        self.stage._ensure_cosign_tags = AsyncMock()
         self.assertFalse(await self.stage._has_cosign_signature(digest))
 
 
+class TestParseIncludeEntry(unittest.TestCase):
+    """Test _parse_include_entry static method."""
+
+    def test_plain_digest(self):
+        digest, alias = ContainerFirstStage._parse_include_entry("sha256:" + "a" * 64)
+        assert digest == "sha256:" + "a" * 64
+        assert alias is None
+
+    def test_named_alias(self):
+        entry = "cli=sha256:" + "b" * 64
+        digest, alias = ContainerFirstStage._parse_include_entry(entry)
+        assert digest == "sha256:" + "b" * 64
+        assert alias == "cli"
+
+    def test_named_alias_with_hyphens(self):
+        entry = "cluster-dns-operator=sha256:" + "c" * 64
+        digest, alias = ContainerFirstStage._parse_include_entry(entry)
+        assert digest == "sha256:" + "c" * 64
+        assert alias == "cluster-dns-operator"
+
+    def test_non_digest_tag(self):
+        digest, alias = ContainerFirstStage._parse_include_entry("latest")
+        assert digest is None
+        assert alias is None
+
+    def test_wildcard_pattern(self):
+        digest, alias = ContainerFirstStage._parse_include_entry("v1.*")
+        assert digest is None
+        assert alias is None
+
+    def test_name_equals_non_digest(self):
+        digest, alias = ContainerFirstStage._parse_include_entry("name=latest")
+        assert digest is None
+        assert alias is None
+
+    def test_short_hex_rejected(self):
+        digest, alias = ContainerFirstStage._parse_include_entry("sha256:" + "a" * 63)
+        assert digest is None
+        assert alias is None
+
+    def test_uppercase_hex_rejected(self):
+        digest, alias = ContainerFirstStage._parse_include_entry("sha256:" + "A" * 64)
+        assert digest is None
+        assert alias is None
+
+
+class TestCanBypassTaglist(unittest.TestCase):
+    """Test _can_bypass_taglist decision logic."""
+
+    def _make_stage(self, includes=None, excludes=None, mirror=False):
+        remote = MagicMock()
+        remote.policy = MagicMock()
+        remote.includes = includes
+        remote.excludes = excludes
+        stage = ContainerFirstStage(remote=remote, signed_only=False, mirror=mirror)
+        return stage
+
+    def test_all_digests_no_excludes(self):
+        stage = self._make_stage(includes=["sha256:" + "a" * 64, "sha256:" + "b" * 64])
+        assert stage._can_bypass_taglist() is True
+
+    def test_non_digest_entry_prevents_bypass(self):
+        stage = self._make_stage(includes=["latest", "sha256:" + "a" * 64])
+        assert stage._can_bypass_taglist() is False
+
+    def test_alias_entry_allows_bypass(self):
+        stage = self._make_stage(includes=["cli=sha256:" + "a" * 64])
+        assert stage._can_bypass_taglist() is True
+
+    def test_mixed_digests_and_aliases(self):
+        stage = self._make_stage(includes=["sha256:" + "a" * 64, "cli=sha256:" + "b" * 64])
+        assert stage._can_bypass_taglist() is True
+
+    def test_wildcard_entry_prevents_bypass(self):
+        stage = self._make_stage(includes=["v1.*"])
+        assert stage._can_bypass_taglist() is False
+
+    def test_empty_includes_prevents_bypass(self):
+        stage = self._make_stage(includes=[])
+        assert stage._can_bypass_taglist() is False
+
+    def test_none_includes_prevents_bypass(self):
+        stage = self._make_stage(includes=None)
+        assert stage._can_bypass_taglist() is False
+
+    def test_mirror_mode_prevents_bypass(self):
+        stage = self._make_stage(includes=["sha256:" + "a" * 64], mirror=True)
+        assert stage._can_bypass_taglist() is False
+
+    def test_uppercase_hex_prevents_bypass(self):
+        stage = self._make_stage(includes=["sha256:" + "A" * 64])
+        assert stage._can_bypass_taglist() is False
+
+    def test_short_hex_prevents_bypass(self):
+        stage = self._make_stage(includes=["sha256:" + "a" * 63])
+        assert stage._can_bypass_taglist() is False
+
+    def test_excludes_do_not_affect_bypass_decision(self):
+        stage = self._make_stage(
+            includes=["sha256:" + "a" * 64],
+            excludes=["sha256-*"],
+        )
+        assert stage._can_bypass_taglist() is True
+
+
+class TestEnsureCosignTags(unittest.IsolatedAsyncioTestCase):
+    """Test _ensure_cosign_tags cached HEAD-probe logic."""
+
+    def setUp(self):
+        remote = MagicMock()
+        remote.policy = MagicMock()
+        remote.namespaced_upstream_name = "library/test"
+        remote.url = "https://registry.example/"
+        remote.get_downloader = MagicMock()
+        remote.excludes = None
+        self.stage = ContainerFirstStage(remote=remote, signed_only=False)
+        self.stage._bypass_taglist = True
+
+    async def test_probes_all_cosign_patterns(self):
+        digest = "sha256:" + "a" * 64
+        hex_part = "a" * 64
+
+        self.stage._tag_exists = AsyncMock(return_value=True)
+        await self.stage._ensure_cosign_tags([digest])
+
+        expected_tags = {
+            f"sha256-{hex_part}.sig",
+            f"sha256-{hex_part}.att",
+            f"sha256-{hex_part}.sbom",
+            f"sha256-{hex_part}",
+        }
+        assert set(self.stage._cosign_tags) == expected_tags
+
+    async def test_returns_only_existing_tags(self):
+        digest = "sha256:" + "a" * 64
+        hex_part = "a" * 64
+
+        async def selective_exists(tag):
+            return tag.endswith(".sig")
+
+        self.stage._tag_exists = selective_exists
+        await self.stage._ensure_cosign_tags([digest])
+
+        assert self.stage._cosign_tags == [f"sha256-{hex_part}.sig"]
+
+    async def test_skips_non_sha256_digests(self):
+        self.stage._tag_exists = AsyncMock(return_value=True)
+        await self.stage._ensure_cosign_tags(["not-a-digest"])
+        assert self.stage._cosign_tags == []
+        self.stage._tag_exists.assert_not_called()
+
+    async def test_empty_input(self):
+        self.stage._tag_exists = AsyncMock(return_value=True)
+        await self.stage._ensure_cosign_tags([])
+        assert self.stage._cosign_tags == []
+
+    async def test_caches_probed_digests(self):
+        digest = "sha256:" + "a" * 64
+        self.stage._tag_exists = AsyncMock(return_value=False)
+
+        await self.stage._ensure_cosign_tags([digest])
+        call_count = self.stage._tag_exists.await_count
+
+        await self.stage._ensure_cosign_tags([digest])
+        assert self.stage._tag_exists.await_count == call_count
+
+    async def test_noop_when_bypass_taglist_false(self):
+        self.stage._bypass_taglist = False
+        self.stage._tag_exists = AsyncMock(return_value=True)
+        await self.stage._ensure_cosign_tags(["sha256:" + "a" * 64])
+        assert self.stage._cosign_tags == []
+        self.stage._tag_exists.assert_not_called()
+
+    async def test_excludes_filter_candidates(self):
+        digest = "sha256:" + "a" * 64
+        self.stage.remote.excludes = ["sha256-*"]
+
+        self.stage._tag_exists = AsyncMock(return_value=True)
+        await self.stage._ensure_cosign_tags([digest])
+
+        assert self.stage._cosign_tags == []
+        self.stage._tag_exists.assert_not_called()
+
+    async def test_partial_excludes_filter_sig_only(self):
+        digest = "sha256:" + "a" * 64
+        hex_part = "a" * 64
+        self.stage.remote.excludes = ["*.sig"]
+
+        self.stage._tag_exists = AsyncMock(return_value=True)
+        await self.stage._ensure_cosign_tags([digest])
+
+        found = set(self.stage._cosign_tags)
+        assert f"sha256-{hex_part}.sig" not in found
+        assert f"sha256-{hex_part}.att" in found
+        assert f"sha256-{hex_part}.sbom" in found
+        assert f"sha256-{hex_part}" in found
+
+    async def test_child_manifest_digests_probed(self):
+        parent = "sha256:" + "a" * 64
+        child = "sha256:" + "b" * 64
+
+        self.stage._tag_exists = AsyncMock(return_value=True)
+        await self.stage._ensure_cosign_tags([parent, child])
+
+        tags = set(self.stage._cosign_tags)
+        for hex_part in ["a" * 64, "b" * 64]:
+            for suffix in [".sig", ".att", ".sbom", ""]:
+                assert f"sha256-{hex_part}{suffix}" in tags
+
+    async def test_caching_across_parent_and_child(self):
+        parent = "sha256:" + "a" * 64
+        child = "sha256:" + "b" * 64
+
+        self.stage._tag_exists = AsyncMock(return_value=False)
+
+        await self.stage._ensure_cosign_tags([parent])
+        first_call_count = self.stage._tag_exists.await_count
+
+        await self.stage._ensure_cosign_tags([parent, child])
+        second_call_count = self.stage._tag_exists.await_count - first_call_count
+
+        assert second_call_count == 4
+
+    async def test_rate_limit_error_propagates(self):
+        digest = "sha256:" + "a" * 64
+        exc = aiohttp.ClientResponseError(
+            request_info=MagicMock(), history=(), status=429, message="Too Many Requests"
+        )
+        self.stage._tag_exists = AsyncMock(side_effect=exc)
+
+        with self.assertRaises(aiohttp.ClientResponseError):
+            await self.stage._ensure_cosign_tags([digest])
+
+
+class TestTagExistsHeadProbe(unittest.IsolatedAsyncioTestCase):
+    """Test _tag_exists HEAD request behavior."""
+
+    def setUp(self):
+        remote = MagicMock()
+        remote.policy = MagicMock()
+        remote.namespaced_upstream_name = "library/test"
+        remote.url = "https://registry.example/"
+        self.stage = ContainerFirstStage(remote=remote, signed_only=False)
+
+    async def test_returns_true_on_success(self):
+        mock_downloader = AsyncMock()
+        mock_downloader.run = AsyncMock(return_value=MagicMock())
+        self.stage.remote.get_downloader = MagicMock(return_value=mock_downloader)
+
+        result = await self.stage._tag_exists("sha256-aaa.sig")
+        assert result is True
+
+    async def test_returns_false_on_404(self):
+        mock_downloader = AsyncMock()
+        exc = aiohttp.ClientResponseError(
+            request_info=MagicMock(), history=(), status=404, message="Not Found"
+        )
+        mock_downloader.run = AsyncMock(side_effect=exc)
+        self.stage.remote.get_downloader = MagicMock(return_value=mock_downloader)
+
+        result = await self.stage._tag_exists("sha256-nonexistent.sig")
+        assert result is False
+
+    async def test_raises_on_non_404_error(self):
+        mock_downloader = AsyncMock()
+        exc = aiohttp.ClientResponseError(
+            request_info=MagicMock(), history=(), status=401, message="Unauthorized"
+        )
+        mock_downloader.run = AsyncMock(side_effect=exc)
+        self.stage.remote.get_downloader = MagicMock(return_value=mock_downloader)
+
+        with self.assertRaises(aiohttp.ClientResponseError):
+            await self.stage._tag_exists("sha256-aaa.sig")
+
+
 if __name__ == "__main__":
     unittest.main()