diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 01b0ab92f2f..fdbf1b611b5 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -1434,9 +1434,12 @@ class ToolSourceRecord(Base, Dictifiable, RepresentById):
 
     Deliberately separate from :class:`ToolSource`, whose rows are created
     per executed tool by the job-request path and whose ``source`` column
-    carries the raw source string contract that path deserializes. Store
-    rows are content-addressed by ``hash`` and carry populator metadata;
-    the populator may prune them freely without affecting job records.
+    carries the raw source string contract that path deserializes. The
+    populator keeps one row per ``source_path``; ``hash`` fingerprints the
+    expanded content and is deliberately non-unique — distinct files can
+    expand to identical content (``tools/data_source/upload.xml`` and its
+    ``test/functional/tools`` copy) yet each path must stay resolvable.
+    The populator may prune rows freely without affecting job records.
     """
 
     __tablename__ = "tool_source_record"
@@ -1445,7 +1448,7 @@ class ToolSourceRecord(Base, Dictifiable, RepresentById):
     dict_element_visible_keys = ("id", "hash", "tool_id", "tool_version", "create_time", "update_time")
 
     id: Mapped[int] = mapped_column(primary_key=True)
-    hash: Mapped[str] = mapped_column(String(255), unique=True, index=True, nullable=False)
+    hash: Mapped[str] = mapped_column(String(255), index=True, nullable=False)
     source: Mapped[str] = mapped_column(Text().with_variant(mysql.LONGTEXT(), "mysql"), nullable=False)
     source_class: Mapped[str] = mapped_column(TrimmedString(255))
     tool_id: Mapped[str | None] = mapped_column(String(255), index=True)
diff --git a/lib/galaxy/model/migrations/alembic/versions_gxy/f5a73c8b9d12_add_tool_index_table.py b/lib/galaxy/model/migrations/alembic/versions_gxy/f5a73c8b9d12_add_tool_index_table.py
index b35c2b91247..594351a7b3f 100644
--- a/lib/galaxy/model/migrations/alembic/versions_gxy/f5a73c8b9d12_add_tool_index_table.py
+++ b/lib/galaxy/model/migrations/alembic/versions_gxy/f5a73c8b9d12_add_tool_index_table.py
@@ -52,7 +52,7 @@ def upgrade():
     create_table(
         SOURCE_TABLE_NAME,
         sa.Column("id", sa.Integer, primary_key=True),
-        sa.Column("hash", sa.String(255), nullable=False, unique=True, index=True),
+        sa.Column("hash", sa.String(255), nullable=False, index=True),
         sa.Column("source", sa.Text().with_variant(mysql.LONGTEXT(), "mysql"), nullable=False),
         sa.Column("source_class", sa.String(255)),
         sa.Column("tool_id", sa.String(255), index=True),
diff --git a/lib/galaxy/tool_source_store/database.py b/lib/galaxy/tool_source_store/database.py
index 2ff3b428582..4e42be66398 100644
--- a/lib/galaxy/tool_source_store/database.py
+++ b/lib/galaxy/tool_source_store/database.py
@@ -85,14 +85,45 @@ class DatabaseToolSourceStore(ToolSourceStore):
                 log.debug("read session close raised: %s", e)
 
     def store(self, tool_source: StoredToolSource) -> str:
-        """Store a tool source in the database."""
+        """Store a tool source in the database.
+
+        One row per ``source_path``: distinct files can expand to identical
+        content (same ``hash``), and each path must stay resolvable through
+        :meth:`get_by_source_path` — deduplicating on hash alone would leave
+        the second file's path pointing at nothing. A row whose path already
+        exists is updated in place when its content changed. Path-less
+        sources deduplicate on content hash.
+        """
         session = self._get_session()
 
-        existing = session.execute(
-            select(ToolSourceRecord.id).where(ToolSourceRecord.hash == tool_source.hash)
-        ).scalar_one_or_none()
-        if existing:
-            return tool_source.hash
+        if tool_source.source_path:
+            existing = (
+                session.execute(
+                    select(ToolSourceRecord).where(ToolSourceRecord.source_path == tool_source.source_path).limit(1)
+                )
+                .scalars()
+                .first()
+            )
+            if existing is not None:
+                if existing.hash != tool_source.hash:
+                    existing.hash = tool_source.hash
+                    existing.source = tool_source.raw_source
+                    existing.source_class = tool_source.tool_source_class
+                    existing.tool_id = tool_source.tool_id
+                    existing.tool_version = tool_source.tool_version
+                    existing.tool_dir = tool_source.tool_dir
+                    existing.stored_at = tool_source.stored_at
+                    existing.source_metadata = tool_source.metadata or None
+                    session.flush()
+                return tool_source.hash
+        else:
+            existing_id = (
+                session.execute(select(ToolSourceRecord.id).where(ToolSourceRecord.hash == tool_source.hash).limit(1))
+                .scalars()
+                .first()
+            )
+            if existing_id:
+                return tool_source.hash
 
         model = ToolSourceRecord(
             hash=tool_source.hash,
@@ -111,9 +142,17 @@ class DatabaseToolSourceStore(ToolSourceStore):
         return tool_source.hash
 
     def get(self, hash: str) -> StoredToolSource | None:
-        """Retrieve a tool source by hash."""
+        """Retrieve a tool source by hash.
+
+        Several rows can share a hash (one per source path with identical
+        expanded content); any of them carries the same source.
+        """
         with self._read_session() as session:
-            model = session.execute(select(ToolSourceRecord).where(ToolSourceRecord.hash == hash)).scalar_one_or_none()
+            model = (
+                session.execute(select(ToolSourceRecord).where(ToolSourceRecord.hash == hash).limit(1))
+                .scalars()
+                .first()
+            )
             if not model:
                 return None
             return self._model_to_stored(model)
@@ -135,21 +174,24 @@ class DatabaseToolSourceStore(ToolSourceStore):
     def exists(self, hash: str) -> bool:
         """Check if a tool source exists."""
         with self._read_session() as session:
-            result = session.execute(
-                select(ToolSourceRecord.id).where(ToolSourceRecord.hash == hash)
-            ).scalar_one_or_none()
+            result = (
+                session.execute(select(ToolSourceRecord.id).where(ToolSourceRecord.hash == hash).limit(1))
+                .scalars()
+                .first()
+            )
             return result is not None
 
     def delete(self, hash: str) -> bool:
-        """Delete a tool source by hash."""
+        """Delete all rows carrying this hash (one per source path)."""
         session = self._get_session()
 
-        model = session.execute(select(ToolSourceRecord).where(ToolSourceRecord.hash == hash)).scalar_one_or_none()
+        models = session.execute(select(ToolSourceRecord).where(ToolSourceRecord.hash == hash)).scalars().all()
 
-        if not model:
+        if not models:
             return False
 
-        session.delete(model)
+        for model in models:
+            session.delete(model)
         session.flush()
         return True
 
diff --git a/lib/galaxy/tool_source_store/populator.py b/lib/galaxy/tool_source_store/populator.py
index 5d0bc9e796f..42afb45f3e1 100644
--- a/lib/galaxy/tool_source_store/populator.py
+++ b/lib/galaxy/tool_source_store/populator.py
@@ -713,10 +713,15 @@ def populate_store_inline(
                 stored_at=datetime.now(timezone.utc),
             )
 
-            if incremental and target_store.exists(content_hash):
-                # Source already on disk in the store — skip the write but still
-                # return the parsed source so the index build sees this tool.
-                return ("skipped", d, store_name, stored, tool_source, None)
+            if incremental:
+                # Skip only when this *path* is already stored with this
+                # content. A bare content-hash check is wrong: distinct files
+                # can expand to identical content, and skipping the second
+                # one would leave its path unresolvable (the row keeps the
+                # first writer's ``source_path``).
+                prior = target_store.get_by_source_path(str(path))
+                if prior is not None and prior.hash == content_hash:
+                    return ("skipped", d, store_name, stored, tool_source, None)
 
             if not dry_run:
                 target_store.store(stored)
diff --git a/test/unit/tool_source_store/test_stores.py b/test/unit/tool_source_store/test_stores.py
new file mode 100644
index 00000000000..782ae993a14
--- /dev/null
+++ b/test/unit/tool_source_store/test_stores.py
@@ -0,0 +1,436 @@
+"""Unit tests for tool source storage backends.
+
+Tests verify that the tool source store classes work correctly with
+different backends (database, sqlalchemy). These tests directly
+instantiate the stores to test the backend implementations.
+"""
+
+import pytest
+
+from galaxy.app_unittest_utils.galaxy_mock import MockApp
+from galaxy.tool_source_store import (
+    build_tool_source_store,
+    ConfigurationError,
+    StoredToolSource,
+)
+from galaxy.tool_source_store.database import DatabaseToolSourceStore
+from galaxy.tool_source_store.index import (
+    ToolIndex,
+    ToolIndexEntry,
+)
+from galaxy.tool_source_store.sqlalchemy import SqlAlchemyToolSourceStore
+
+
+class FakeConfig:
+    """Fake config for testing store factory."""
+
+    def __init__(self, **kwargs):
+        for key, value in kwargs.items():
+            setattr(self, key, value)
+
+
+class TestDatabaseBackend:
+    """Unit tests for database backend using MockApp."""
+
+    def test_database_store_basic_operations(self):
+        """Test basic store/get operations with database backend."""
+        app = MockApp()
+        store = DatabaseToolSourceStore(app.model.context)  # type: ignore[arg-type]
+        test_hash = "test_hash_unit_123"
+
+        try:
+            tool_source = StoredToolSource(
+                hash=test_hash,
+                tool_source_class="XmlToolSource",
+                raw_source='<tool id="test" version="1.0"><command>echo</command></tool>',
+                tool_id="test_unit_tool",
+                tool_version="1.0",
+            )
+
+            store.store(tool_source)
+            app.model.context.commit()
+
+            assert store.exists(tool_source.hash)
+
+            retrieved = store.get(tool_source.hash)
+            assert retrieved is not None
+            assert retrieved.tool_id == "test_unit_tool"
+            assert retrieved.tool_version == "1.0"
+            assert "<tool" in retrieved.raw_source
+
+            assert store.delete(tool_source.hash)
+            app.model.context.commit()
+            assert not store.exists(tool_source.hash)
+        finally:
+            if store.exists(test_hash):
+                store.delete(test_hash)
+                app.model.context.commit()
+
+    def test_database_store_index_operations(self):
+        """Test tool index storage with database backend."""
+        app = MockApp()
+        store = DatabaseToolSourceStore(app.model.context)  # type: ignore[arg-type]
+
+        index = ToolIndex()
+        index.entries["test_tool_db"] = ToolIndexEntry(
+            id="test_tool_db",
+            name="Test Tool DB",
+            version="1.0",
+            description="A test tool",
+        )
+
+        store.store_index(index)
+        app.model.context.commit()
+
+        # Clear the cached index to force reload from database
+        store.invalidate_index_cache()
+
+        loaded_index = store.load_index()
+        assert loaded_index is not None
+        assert "test_tool_db" in loaded_index.entries
+        assert loaded_index.entries["test_tool_db"].name == "Test Tool DB"
+
+    def test_database_store_get_by_tool_id(self):
+        """Test retrieving tool sources by tool ID."""
+        app = MockApp()
+        store = DatabaseToolSourceStore(app.model.context)  # type: ignore[arg-type]
+
+        unique_id = "tool_by_id_test_unit"
+        test_hash = f"hash_for_{unique_id}"
+
+        try:
+            tool_source = StoredToolSource(
+                hash=test_hash,
+                tool_source_class="XmlToolSource",
+                raw_source=f'<tool id="{unique_id}" version="1.0"><command>echo</command></tool>',
+                tool_id=unique_id,
+                tool_version="1.0",
+            )
+            store.store(tool_source)
+            app.model.context.commit()
+
+            sources = store.get_by_tool_id(unique_id)
+            assert len(sources) >= 1
+            assert any(s.tool_id == unique_id for s in sources)
+        finally:
+            if store.exists(test_hash):
+                store.delete(test_hash)
+                app.model.context.commit()
+
+    def test_database_store_count(self):
+        """Test counting stored tool sources."""
+        app = MockApp()
+        store = DatabaseToolSourceStore(app.model.context)  # type: ignore[arg-type]
+        test_hash = "count_test_hash_unit"
+
+        try:
+            initial_count = store.count()
+
+            tool_source = StoredToolSource(
+                hash=test_hash,
+                tool_source_class="XmlToolSource",
+                raw_source='<tool id="count_test"><command>echo</command></tool>',
+                tool_id="count_test",
+                tool_version="1.0",
+            )
+            store.store(tool_source)
+            app.model.context.commit()
+
+            assert store.count() == initial_count + 1
+
+            store.delete(test_hash)
+            app.model.context.commit()
+            assert store.count() == initial_count
+        finally:
+            if store.exists(test_hash):
+                store.delete(test_hash)
+                app.model.context.commit()
+
+
+class TestDatabaseBackendPathRows:
+    """One row per source path — identical content must not swallow paths."""
+
+    def _stored(self, hash, path, raw='<tool id="upload1" version="1.1.7"/>'):
+        return StoredToolSource(
+            hash=hash,
+            tool_source_class="XmlToolSource",
+            raw_source=raw,
+            tool_id="upload1",
+            tool_version="1.1.7",
+            source_path=path,
+        )
+
+    def test_identical_content_keeps_row_per_source_path(self):
+        app = MockApp()
+        store = DatabaseToolSourceStore(app.model.context)  # type: ignore[arg-type]
+        twin_hash = "twin_hash_per_path"
+        store.store(self._stored(twin_hash, "/galaxy/tools/data_source/upload.xml"))
+        store.store(self._stored(twin_hash, "/galaxy/test/functional/tools/upload.xml"))
+        app.model.context.commit()
+
+        first = store.get_by_source_path("/galaxy/tools/data_source/upload.xml")
+        second = store.get_by_source_path("/galaxy/test/functional/tools/upload.xml")
+        assert first is not None and first.hash == twin_hash
+        assert second is not None and second.hash == twin_hash
+
+        assert store.delete(twin_hash)
+        app.model.context.commit()
+        assert store.get_by_source_path("/galaxy/tools/data_source/upload.xml") is None
+        assert store.get_by_source_path("/galaxy/test/functional/tools/upload.xml") is None
+
+    def test_changed_content_updates_path_row_in_place(self):
+        app = MockApp()
+        store = DatabaseToolSourceStore(app.model.context)  # type: ignore[arg-type]
+        path = "/galaxy/tools/edited.xml"
+        store.store(self._stored("edited_hash_v1", path, raw="<tool/>"))
+        app.model.context.commit()
+        store.store(self._stored("edited_hash_v2", path, raw="<tool><description/></tool>"))
+        app.model.context.commit()
+
+        row = store.get_by_source_path(path)
+        assert row is not None
+        assert row.hash == "edited_hash_v2"
+        assert not store.exists("edited_hash_v1")
+        store.delete("edited_hash_v2")
+        app.model.context.commit()
+
+    def test_pathless_sources_dedupe_on_hash(self):
+        app = MockApp()
+        store = DatabaseToolSourceStore(app.model.context)  # type: ignore[arg-type]
+        store.store(self._stored("pathless_hash", None))
+        store.store(self._stored("pathless_hash", None))
+        app.model.context.commit()
+        assert store.exists("pathless_hash")
+        assert store.delete("pathless_hash")
+        app.model.context.commit()
+        assert not store.exists("pathless_hash")
+
+
+class TestSqlAlchemyBackend:
+    """Tests for the sqlalchemy/sqlite backend."""
+
+    def test_sqlalchemy_store_basic_operations(self, tmp_path):
+        store = SqlAlchemyToolSourceStore(path=str(tmp_path / "ts.sqlite"))
+
+        tool_source = StoredToolSource(
+            hash="sa_test_hash_123",
+            tool_source_class="XmlToolSource",
+            raw_source='<tool id="sa_test" version="2.0"><command>cat</command></tool>',
+            tool_id="sa_test_tool",
+            tool_version="2.0",
+        )
+        store.store(tool_source)
+
+        assert store.exists("sa_test_hash_123")
+        retrieved = store.get("sa_test_hash_123")
+        assert retrieved is not None
+        assert retrieved.tool_id == "sa_test_tool"
+        assert store.count() >= 1
+        assert store.delete("sa_test_hash_123")
+        assert not store.exists("sa_test_hash_123")
+
+    def test_sqlalchemy_identical_content_keeps_row_per_source_path(self, tmp_path):
+        store = SqlAlchemyToolSourceStore(path=str(tmp_path / "twins.sqlite"))
+        for path in ("/galaxy/tools/a/upload.xml", "/galaxy/tools/b/upload.xml"):
+            store.store(
+                StoredToolSource(
+                    hash="twin_hash",
+                    tool_source_class="XmlToolSource",
+                    raw_source="<tool/>",
+                    tool_id="upload1",
+                    tool_version="1.1.7",
+                    source_path=path,
+                )
+            )
+        assert store.get_by_source_path("/galaxy/tools/a/upload.xml") is not None
+        assert store.get_by_source_path("/galaxy/tools/b/upload.xml") is not None
+        assert store.delete("twin_hash")
+        assert store.get_by_source_path("/galaxy/tools/a/upload.xml") is None
+
+    def test_sqlalchemy_store_persistence(self, tmp_path):
+        path = str(tmp_path / "ts.sqlite")
+        store1 = SqlAlchemyToolSourceStore(path=path)
+        store1.store(
+            StoredToolSource(
+                hash="persist_test_hash",
+                tool_source_class="XmlToolSource",
+                raw_source='<tool id="persist"><command>echo</command></tool>',
+                tool_id="persist_tool",
+                tool_version="1.0",
+            )
+        )
+        store2 = SqlAlchemyToolSourceStore(path=path)
+        assert store2.exists("persist_test_hash")
+        retrieved = store2.get("persist_test_hash")
+        assert retrieved is not None
+        assert retrieved.tool_id == "persist_tool"
+
+
+class TestBuildToolSourceStore:
+    """Tests for the store factory function."""
+
+    def test_build_database_store(self):
+        app = MockApp()
+        store = build_tool_source_store(app.config, app.model.context)  # type: ignore[arg-type]
+        assert isinstance(store, DatabaseToolSourceStore)
+
+    def test_build_sqlalchemy_store(self, tmp_path):
+        config = FakeConfig(
+            tool_source_store="sqlalchemy",
+            tool_source_disk_path=str(tmp_path / "ts.sqlite"),
+            tool_configs=[],
+            tool_source_stores=None,
+            use_lazy_toolbox=False,
+        )
+        store = build_tool_source_store(config, None)  # type: ignore[arg-type]
+        assert isinstance(store, SqlAlchemyToolSourceStore)
+
+    def test_build_sqlalchemy_store_missing_path_raises(self):
+        config = FakeConfig(
+            tool_source_store="sqlalchemy",
+            tool_source_disk_path=None,
+            tool_configs=[],
+            tool_source_stores=None,
+            use_lazy_toolbox=False,
+        )
+        with pytest.raises(ConfigurationError):
+            build_tool_source_store(config, None)  # type: ignore[arg-type]
+
+    def test_build_unknown_backend_raises(self):
+        config = FakeConfig(
+            tool_source_store="not-a-backend",
+            tool_source_disk_path=None,
+            tool_configs=[],
+            tool_source_stores=None,
+            use_lazy_toolbox=False,
+        )
+        with pytest.raises(ConfigurationError):
+            build_tool_source_store(config, None)  # type: ignore[arg-type]
+
+
+class TestPerConfStoreRouting:
+    """Tests for per-conf store routing in build_tool_source_store."""
+
+    def _config(self, tmp_path, **overrides):
+        defaults = dict(
+            tool_source_store="sqlalchemy",
+            tool_source_disk_path=str(tmp_path / "ts.sqlite"),
+            tool_configs=[],
+            tool_source_stores={},
+            use_lazy_toolbox=False,
+        )
+        defaults.update(overrides)
+        return FakeConfig(**defaults)
+
+    def test_lazy_off_ignores_unknown_per_conf_store(self, tmp_path, caplog):
+        conf = tmp_path / "extra_tool_conf.xml"
+        conf.write_text('<?xml version="1.0"?>\n<toolbox store="missing_alias"/>\n')
+        config = self._config(tmp_path, tool_configs=[str(conf)])
+        with caplog.at_level("INFO", logger="galaxy.tool_source_store"):
+            store = build_tool_source_store(config, None)
+        from galaxy.tool_source_store.composite import CompositeToolSourceStore
+
+        assert isinstance(store, SqlAlchemyToolSourceStore)
+        assert not isinstance(store, CompositeToolSourceStore)
+        assert any("missing_alias" in rec.message for rec in caplog.records)
+
+    def test_lazy_unset_also_ignores_per_conf_store(self, tmp_path):
+        conf = tmp_path / "extra_tool_conf.xml"
+        conf.write_text('<?xml version="1.0"?>\n<toolbox store="anything"/>\n')
+        config = self._config(tmp_path, tool_configs=[str(conf)], use_lazy_toolbox=None)
+        store = build_tool_source_store(config, None)
+        assert isinstance(store, SqlAlchemyToolSourceStore)
+
+    def test_lazy_on_with_unknown_store_still_raises(self, tmp_path):
+        conf = tmp_path / "extra_tool_conf.xml"
+        conf.write_text('<?xml version="1.0"?>\n<toolbox store="missing_alias"/>\n')
+        config = self._config(tmp_path, tool_configs=[str(conf)], use_lazy_toolbox=True)
+        with pytest.raises(ConfigurationError):
+            build_tool_source_store(config, None)
+
+
+class TestToolIndex:
+    """Tests for ToolIndex functionality."""
+
+    def test_index_search(self):
+        """Test searching the tool index."""
+        index = ToolIndex()
+        index.entries["filter_tool"] = ToolIndexEntry(
+            id="filter_tool",
+            name="Filter Tool",
+            version="1.0",
+            description="Filters data by column",
+        )
+        index.entries["cat_tool"] = ToolIndexEntry(
+            id="cat_tool",
+            name="Concatenate",
+            version="2.0",
+            description="Concatenates files",
+        )
+
+        results = index.search("Filter", limit=10)
+        assert len(results) >= 1
+        assert any(r.id == "filter_tool" for r in results)
+
+        results = index.search("column", limit=10)
+        assert len(results) >= 1
+        assert any(r.id == "filter_tool" for r in results)
+
+    def test_index_serialization(self):
+        """Test index to_dict/from_dict round trip."""
+        index = ToolIndex()
+        index.entries["test_tool"] = ToolIndexEntry(
+            id="test_tool",
+            name="Test",
+            version="1.0",
+            description="Test tool",
+            labels=["genomics"],
+        )
+        index.by_section["section1"] = ["test_tool"]
+
+        data = index.to_dict()
+
+        restored = ToolIndex.from_dict(data)
+
+        assert "test_tool" in restored.entries
+        assert restored.entries["test_tool"].name == "Test"
+        assert "section1" in restored.by_section
+
+    def test_index_get_tests_summary(self):
+        """Test generating tests summary from index."""
+        index = ToolIndex()
+        index.entries["tool1"] = ToolIndexEntry(
+            id="tool1",
+            name="Tool 1",
+            version="1.0",
+            test_count=3,
+        )
+        index.entries["tool2"] = ToolIndexEntry(
+            id="tool2",
+            name="Tool 2",
+            version="2.0",
+            test_count=0,
+        )
+
+        summary = index.get_tests_summary()
+        # Tools with tests must appear; tools without tests must not.
+        assert "tool1" in summary
+        assert summary["tool1"]["1.0"]["count"] == 3
+        assert summary["tool1"]["1.0"]["tool_name"] == "Tool 1"
+        assert "tool2" not in summary
+
+    def test_index_get_all_requirements(self):
+        """Test aggregating all requirements from index."""
+        index = ToolIndex()
+        index.entries["tool1"] = ToolIndexEntry(
+            id="tool1",
+            name="Tool 1",
+            version="1.0",
+            requirements=[
+                {"name": "samtools", "version": "1.0", "type": "package"},
+            ],
+        )
+
+        requirements = index.get_all_requirements()
+        assert isinstance(requirements, list)
+        assert {"name": "samtools", "version": "1.0", "type": "package"} in requirements