diff --git a/docs/CONCEPT.md b/docs/CONCEPT.md
index cfb0d03..1ba0e13 100644
--- a/docs/CONCEPT.md
+++ b/docs/CONCEPT.md
@@ -104,6 +104,13 @@ Consumers should be able to request the governance explanation and dependency
 impact separately from row access. Seeing catalogue metadata must not imply
 permission to read protected data.
 
+When Search is installed, Datasources contributes catalogue entries as a native
+search source. Only the stable catalogue identity, display name, description,
+mode, shape, lifecycle state, classification, publication state, and authority
+mode are indexed. Every result is re-authorized against the current tenant and
+catalogue-read permission. Rows, schemas, connector references, credentials,
+arbitrary metadata, and provenance remain outside the derived search index.
+
 ## Next Providers
 
 Connector providers should cover:
diff --git a/src/govoplan_datasources/backend/manifest.py b/src/govoplan_datasources/backend/manifest.py
index 95e0a6d..fa54b16 100644
--- a/src/govoplan_datasources/backend/manifest.py
+++ b/src/govoplan_datasources/backend/manifest.py
@@ -35,9 +35,13 @@ from govoplan_core.core.provider_governance import (
     ModuleArchitectureDocumentation,
     ModuleMaturityEvidence,
 )
+from govoplan_core.core.search import SearchSourceProviderRegistration
 from govoplan_core.core.views import ViewSurface
 from govoplan_core.db.base import Base
 from govoplan_datasources.backend.db import models as datasource_models
+from govoplan_datasources.backend.search_source import (
+    create_datasources_search_source,
+)
 from govoplan_datasources.backend.service import (
     ADMIN_SCOPE,
     CATALOGUE_READ_SCOPE,
@@ -219,6 +223,7 @@ manifest = ModuleManifest(
         "files",
         "notifications",
         "policy",
+        "search",
     ),
     optional_capabilities=(
         CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
@@ -254,6 +259,12 @@ manifest = ModuleManifest(
             version_max_exclusive="1.0.0",
             optional=True,
         ),
+        ModuleInterfaceRequirement(
+            name="search.source",
+            version_min="1.0.0",
+            version_max_exclusive="2.0.0",
+            optional=True,
+        ),
     ),
     permissions=PERMISSIONS,
     role_templates=ROLE_TEMPLATES,
@@ -302,6 +313,12 @@ manifest = ModuleManifest(
         CAPABILITY_DATASOURCE_PUBLICATION: _provider,
     },
     tenant_summary_providers=(_tenant_summary,),
+    search_sources=(
+        SearchSourceProviderRegistration(
+            id="datasources.catalogue",
+            factory=create_datasources_search_source,
+        ),
+    ),
     migration_spec=MigrationSpec(
         module_id=MODULE_ID,
         metadata=Base.metadata,
@@ -350,7 +367,11 @@ manifest = ModuleManifest(
                 "reports, controls, and decisions. It preserves origin source mode, "
                 "structured health, declared pushdown, and effective row, byte, and time "
                 "limits for live previews. Governance metadata visibility does not "
-                "grant access to protected rows."
+                "grant access to protected rows. When Search is enabled, the module "
+                "indexes only bounded catalogue labels and governance-safe facets, "
+                "then rechecks the current catalogue permission before returning a "
+                "result. Rows, schemas, connector references, credentials, arbitrary "
+                "metadata, and provenance are never copied into the search index."
             ),
             layer="available",
             documentation_types=("admin", "user"),
@@ -362,6 +383,7 @@ manifest = ModuleManifest(
                 "files",
                 "reporting",
                 "risk_compliance",
+                "search",
             ),
             order=70,
             metadata={
diff --git a/src/govoplan_datasources/backend/search_source.py b/src/govoplan_datasources/backend/search_source.py
new file mode 100644
index 0000000..d32f1aa
--- /dev/null
+++ b/src/govoplan_datasources/backend/search_source.py
@@ -0,0 +1,179 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from urllib.parse import quote
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.modules import ModuleContext
+from govoplan_core.core.search import (
+    SearchAuthorizationRequest,
+    SearchBackfillPage,
+    SearchBackfillRequest,
+    SearchDocument,
+    SearchResourceType,
+)
+from govoplan_datasources.backend.db.models import DatasourceRecord
+from govoplan_datasources.backend.service import ADMIN_SCOPE, CATALOGUE_READ_SCOPE
+
+
+PROVIDER_ID = "datasources.catalogue"
+RESOURCE_TYPE = "datasource"
+
+
+class DatasourcesSearchSource:
+    def resource_types(self) -> Sequence[SearchResourceType]:
+        return (
+            SearchResourceType(
+                provider_id=PROVIDER_ID,
+                module_id="datasources",
+                resource_type=RESOURCE_TYPE,
+                label="Datasources",
+                requires_authorization_recheck=True,
+            ),
+        )
+
+    def backfill(
+        self,
+        session: object,
+        *,
+        request: SearchBackfillRequest,
+    ) -> SearchBackfillPage:
+        _assert_source(request.provider_id, request.resource_type)
+        db = _session(session)
+        statement = select(DatasourceRecord).where(
+            DatasourceRecord.tenant_id == request.tenant_id,
+            DatasourceRecord.deleted_at.is_(None),
+        )
+        if request.cursor:
+            statement = statement.where(DatasourceRecord.id > request.cursor)
+        rows = list(
+            db.scalars(
+                statement.order_by(DatasourceRecord.id).limit(request.limit + 1)
+            ).all()
+        )
+        has_more = len(rows) > request.limit
+        selected = rows[: request.limit]
+        high_watermark = db.scalar(
+            select(func.max(DatasourceRecord.updated_at)).where(
+                DatasourceRecord.tenant_id == request.tenant_id,
+                DatasourceRecord.deleted_at.is_(None),
+            )
+        )
+        return SearchBackfillPage(
+            documents=tuple(_document(row) for row in selected),
+            next_cursor=selected[-1].id if has_more and selected else None,
+            complete=not has_more,
+            high_watermark=high_watermark.isoformat() if high_watermark else None,
+        )
+
+    def authorize(
+        self,
+        session: object,
+        principal: object,
+        *,
+        requests: Sequence[SearchAuthorizationRequest],
+    ) -> Mapping[str, bool]:
+        decisions = {item.reference.key: False for item in requests}
+        if not isinstance(principal, ApiPrincipal) or not (
+            principal.has(CATALOGUE_READ_SCOPE) or principal.has(ADMIN_SCOPE)
+        ):
+            return decisions
+        db = _session(session)
+        eligible = [
+            request
+            for request in requests
+            if request.reference.tenant_id == principal.tenant_id
+            and request.reference.module_id == "datasources"
+            and request.reference.resource_type == RESOURCE_TYPE
+        ]
+        resource_ids = {request.reference.resource_id for request in eligible}
+        available_ids = (
+            set(
+                db.scalars(
+                    select(DatasourceRecord.id).where(
+                        DatasourceRecord.tenant_id == principal.tenant_id,
+                        DatasourceRecord.id.in_(resource_ids),
+                        DatasourceRecord.deleted_at.is_(None),
+                    )
+                ).all()
+            )
+            if resource_ids
+            else set()
+        )
+        for request in eligible:
+            decisions[request.reference.key] = (
+                request.reference.resource_id in available_ids
+            )
+        return decisions
+
+
+def create_datasources_search_source(
+    _context: ModuleContext,
+) -> DatasourcesSearchSource:
+    return DatasourcesSearchSource()
+
+
+def _document(row: DatasourceRecord) -> SearchDocument:
+    datasource_ref = quote(f"datasource:{row.id}", safe="")
+    return SearchDocument(
+        tenant_id=row.tenant_id,
+        module_id="datasources",
+        provider_id=PROVIDER_ID,
+        resource_type=RESOURCE_TYPE,
+        resource_id=row.id,
+        title=row.name,
+        url=f"/datasources?datasource={datasource_ref}",
+        summary=(row.description or row.source_name)[:4000],
+        keywords=tuple(
+            value[:200]
+            for value in (
+                row.source_name,
+                row.kind,
+                row.mode,
+                row.shape,
+                row.status,
+                row.classification,
+                row.publication_state,
+            )
+            if value
+        ),
+        visibility="restricted",
+        acl_tokens=(
+            f"scope:{CATALOGUE_READ_SCOPE}",
+            f"scope:{ADMIN_SCOPE}",
+        ),
+        metadata={
+            "kind": row.kind,
+            "mode": row.mode,
+            "shape": row.shape,
+            "status": row.status,
+            "classification": row.classification,
+            "publication_state": row.publication_state,
+            "authority_mode": row.authority_mode,
+        },
+        source_revision=f"{row.schema_version}:{row.updated_at.isoformat()}",
+        source_updated_at=row.updated_at,
+        requires_authorization_recheck=True,
+    )
+
+
+def _assert_source(provider_id: str, resource_type: str) -> None:
+    if provider_id != PROVIDER_ID or resource_type != RESOURCE_TYPE:
+        raise ValueError("Unsupported Datasources search source.")
+
+
+def _session(value: object) -> Session:
+    if not isinstance(value, Session):
+        raise TypeError("Datasources search requires a SQLAlchemy session.")
+    return value
+
+
+__all__ = [
+    "DatasourcesSearchSource",
+    "PROVIDER_ID",
+    "RESOURCE_TYPE",
+    "create_datasources_search_source",
+]
diff --git a/tests/test_search_source.py b/tests/test_search_source.py
new file mode 100644
index 0000000..1099ba8
--- /dev/null
+++ b/tests/test_search_source.py
@@ -0,0 +1,162 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.search import (
+    SearchAuthorizationRequest,
+    SearchBackfillRequest,
+    SearchResourceReference,
+)
+from govoplan_core.db.base import Base
+from govoplan_datasources.backend.db.models import DatasourceRecord
+from govoplan_datasources.backend.manifest import get_manifest
+from govoplan_datasources.backend.search_source import (
+    DatasourcesSearchSource,
+    PROVIDER_ID,
+    RESOURCE_TYPE,
+)
+from govoplan_datasources.backend.service import ADMIN_SCOPE, CATALOGUE_READ_SCOPE
+
+
+class DatasourcesSearchSourceTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite://")
+        Base.metadata.create_all(
+            self.engine,
+            tables=(DatasourceRecord.__table__,),
+        )
+        self.session = Session(self.engine)
+        self.session.add_all(
+            (
+                _datasource("source-1", "tenant-1", "Monthly source"),
+                _datasource("source-other", "tenant-2", "Other tenant"),
+            )
+        )
+        self.session.commit()
+        self.source = DatasourcesSearchSource()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_manifest_registers_optional_search_source(self) -> None:
+        manifest = get_manifest()
+
+        self.assertIn("search", manifest.optional_dependencies)
+        self.assertIn(
+            PROVIDER_ID,
+            {registration.id for registration in manifest.search_sources},
+        )
+
+    def test_backfill_is_tenant_bound_and_excludes_protected_payloads(self) -> None:
+        page = self.source.backfill(
+            self.session,
+            request=SearchBackfillRequest(
+                tenant_id="tenant-1",
+                provider_id=PROVIDER_ID,
+                resource_type=RESOURCE_TYPE,
+                rebuild_id="rebuild-1",
+            ),
+        )
+
+        self.assertEqual(("source-1",), tuple(doc.resource_id for doc in page.documents))
+        document = page.documents[0]
+        serialized = repr(document)
+        self.assertNotIn("must-not-be-indexed", serialized)
+        self.assertNotIn("secret_field", serialized)
+        self.assertNotIn("credential-1", serialized)
+        self.assertTrue(document.requires_authorization_recheck)
+        self.assertEqual(
+            "/datasources?datasource=datasource%3Asource-1",
+            document.url,
+        )
+
+    def test_authorization_rechecks_scope_tenant_and_current_existence(self) -> None:
+        reference = SearchResourceReference(
+            tenant_id="tenant-1",
+            module_id="datasources",
+            resource_type=RESOURCE_TYPE,
+            resource_id="source-1",
+        )
+        request = SearchAuthorizationRequest(reference=reference, source_revision="1")
+
+        self.assertTrue(
+            self.source.authorize(
+                self.session,
+                _principal({CATALOGUE_READ_SCOPE}),
+                requests=(request,),
+            )[reference.key]
+        )
+        self.assertTrue(
+            self.source.authorize(
+                self.session,
+                _principal({ADMIN_SCOPE}),
+                requests=(request,),
+            )[reference.key]
+        )
+        self.assertFalse(
+            self.source.authorize(
+                self.session,
+                _principal(set()),
+                requests=(request,),
+            )[reference.key]
+        )
+        other_reference = SearchResourceReference(
+            tenant_id="tenant-2",
+            module_id="datasources",
+            resource_type=RESOURCE_TYPE,
+            resource_id="source-other",
+        )
+        other_request = SearchAuthorizationRequest(
+            reference=other_reference,
+            source_revision="1",
+        )
+        self.assertFalse(
+            self.source.authorize(
+                self.session,
+                _principal({CATALOGUE_READ_SCOPE}),
+                requests=(other_request,),
+            )[other_reference.key]
+        )
+
+
+def _datasource(identifier: str, tenant_id: str, name: str) -> DatasourceRecord:
+    return DatasourceRecord(
+        id=identifier,
+        tenant_id=tenant_id,
+        source_name=identifier.replace("-", "_"),
+        name=name,
+        description="Safe catalogue description",
+        kind="database",
+        mode="cached",
+        shape="tabular",
+        status="active",
+        provider="connectors.sql",
+        provider_ref="credential-1",
+        schema_=[{"name": "secret_field", "type": "string"}],
+        provenance_={"query": "must-not-be-indexed"},
+        metadata_={"password": "must-not-be-indexed"},
+    )
+
+
+def _principal(scopes: set[str]) -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id="account-1",
+            membership_id="user-1",
+            tenant_id="tenant-1",
+            scopes=frozenset(scopes),
+        ),
+        account=SimpleNamespace(id="account-1"),
+        user=SimpleNamespace(id="user-1"),
+    )
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/webui/src/features/datasources/DatasourcesPage.tsx b/webui/src/features/datasources/DatasourcesPage.tsx
index 14cd153..128e14d 100644
--- a/webui/src/features/datasources/DatasourcesPage.tsx
+++ b/webui/src/features/datasources/DatasourcesPage.tsx
@@ -83,7 +83,9 @@ export default function DatasourcesPage({
   const [stages, setStages] = useState<DatasourceStage[]>([]);
   const [origins, setOrigins] = useState<DatasourceOrigin[]>([]);
   const [originsAvailable, setOriginsAvailable] = useState(false);
-  const [selectedDatasourceRef, setSelectedDatasourceRef] = useState("");
+  const [selectedDatasourceRef, setSelectedDatasourceRef] = useState(
+    initialDatasourceRef
+  );
   const [selectedStageRef, setSelectedStageRef] = useState("");
   const [selectedOriginRef, setSelectedOriginRef] = useState("");
   const [preview, setPreview] = useState<DatasourcePreview | null>(null);
@@ -1501,6 +1503,11 @@ function filterItems<T>(
     : items;
 }
 
+function initialDatasourceRef(): string {
+  if (typeof window === "undefined") return "";
+  return new URLSearchParams(window.location.search).get("datasource") ?? "";
+}
+
 function parseRows(text: string): Record<string, unknown>[] {
   const value: unknown = JSON.parse(text);
   if (!Array.isArray(value) || value.some((row) => !isRecord(row))) {
diff --git a/webui/src/module.ts b/webui/src/module.ts
index 0f37298..547bf1b 100644
--- a/webui/src/module.ts
+++ b/webui/src/module.ts
@@ -10,14 +10,15 @@ const readScopes = ["datasources:catalogue:read", "datasources:source:admin"];
 export const datasourcesModule: PlatformWebModule = {
   id: "datasources",
   label: "i18n:govoplan-datasources.datasources",
-  version: "0.1.14",
+  version: "0.1.18",
   optionalDependencies: [
     "access",
     "audit",
     "connectors",
     "files",
     "notifications",
-    "policy"
+    "policy",
+    "search"
   ],
   translations: generatedTranslations,
   viewSurfaces: [