feat: per-cluster target_filter (feat_cluster_target_filter)

backend/app/adapters/elastic.py

@@ -21,6 +21,7 @@
 from __future__ import annotations
 
 import base64
+import fnmatch
 import json
 from datetime import UTC, datetime
 from typing import Any
@@ -355,12 +356,23 @@ def _enforce_min_version(self) -> None:
                 f"{OPENSEARCH_MIN_VERSION[0]}.{OPENSEARCH_MIN_VERSION[1]}"
             )
 
-    async def list_targets(self, *, request_id: str | None = None) -> list[TargetInfo]:
+    async def list_targets(
+        self,
+        *,
+        request_id: str | None = None,
+        target_filter: str | None = None,
+    ) -> list[TargetInfo]:
         """List indices on the cluster via ``_cat/indices?format=json``.
 
         System indices (those whose name starts with ``.``) are filtered out
         so the operator sees only user-facing collections.
 
+        When ``target_filter`` is provided (feat_cluster_target_filter FR-3),
+        the result is further restricted to names where
+        ``fnmatch.fnmatchcase(name, target_filter)`` returns True. The system-
+        index exclusion runs FIRST so operators cannot re-expose ``.kibana*``
+        via a permissive filter.
+
         ``translate_errors=False`` is used so the per-status mapping below can
         distinguish ACL-restricted clusters (401/403 → ``TargetsForbiddenError``)
         from unreachable clusters (5xx / connection failures →
@@ -400,7 +412,9 @@ async def list_targets(self, *, request_id: str | None = None) -> list[TargetInf
         for row in rows:
             name = row.get("index")
             if not name or name.startswith("."):
-                continue
+                continue  # system-index exclusion — runs FIRST
+            if target_filter is not None and not fnmatch.fnmatchcase(name, target_filter):
+                continue  # operator glob filter — runs SECOND
             doc_count_raw = row.get("docs.count")
             doc_count: int | None
             if doc_count_raw is None or doc_count_raw == "":

backend/app/adapters/protocol.py

@@ -128,9 +128,24 @@ async def health_check(self, *, request_id: str | None = None) -> HealthStatus:
         """Probe cluster reachability + engine version. See ``HealthStatus`` for shape."""
         ...
 
-    async def list_targets(self, *, request_id: str | None = None) -> list[TargetInfo]:
+    async def list_targets(
+        self,
+        *,
+        request_id: str | None = None,
+        target_filter: str | None = None,
+    ) -> list[TargetInfo]:
         """List indices/collections on the cluster (excludes engine system indices).
 
+        When ``target_filter`` is provided, the result is further restricted to
+        names where ``fnmatch.fnmatchcase(name, target_filter)`` returns True
+        (feat_cluster_target_filter FR-3). Glob syntax: ``*``, ``?``, ``[seq]``,
+        ``[!seq]`` — no brace expansion (pure Python ``fnmatch``). Case-sensitive
+        via ``fnmatchcase`` (avoids platform-dependent ``os.path.normcase`` in
+        ``fnmatch.fnmatch``).
+
+        Order of operations: system-index ``.`` exclusion → glob filter.
+        Operators cannot re-expose system indices via a permissive filter.
+
         Concrete implementations raise ``TargetsForbiddenError`` when the engine
         denies the listing call due to ACL (401/403), and ``ClusterUnreachableError``
         for connection failures / 5xx. Mirrors ``get_schema``'s pattern of

backend/app/api/v1/clusters.py

@@ -124,6 +124,7 @@ def _summary(cluster: Cluster, health: HealthStatus) -> ClusterSummary:
         environment=cluster.environment,
         base_url=cluster.base_url,
         auth_kind=cluster.auth_kind,
+        target_filter=cluster.target_filter,
         created_at=cluster.created_at,
         health_check=HealthCheckResult.model_validate(health.model_dump()),
     )
@@ -139,6 +140,7 @@ def _detail(cluster: Cluster, health: HealthStatus) -> ClusterDetail:
         auth_kind=cluster.auth_kind,
         engine_config=cluster.engine_config,
         notes=cluster.notes,
+        target_filter=cluster.target_filter,
         created_at=cluster.created_at,
         health_check=HealthCheckResult.model_validate(health.model_dump()),
     )
@@ -173,6 +175,7 @@ async def create_cluster(
             credentials_ref=body.credentials_ref,
             engine_config=body.engine_config,
             notes=body.notes,
+            target_filter=body.target_filter,
         )
     except EngineTypeNotSupported as exc:
         raise _err(400, "ENGINE_NOT_SUPPORTED", str(exc), False) from exc
@@ -351,7 +354,9 @@ async def list_cluster_targets(
         raise _err(404, "CLUSTER_NOT_FOUND", f"cluster {cluster_id} not found", False)
     try:
         async with cluster_svc.acquire_adapter(cluster) as adapter:
-            targets = await adapter.list_targets()
+            # feat_cluster_target_filter FR-3: when the cluster has a stored
+            # target_filter, scope list_targets() to matching index names.
+            targets = await adapter.list_targets(target_filter=cluster.target_filter)
             return TargetListResponse(data=targets)
     except TargetsForbiddenError as exc:
         raise _err(403, "TARGETS_FORBIDDEN", str(exc), False) from exc

backend/app/api/v1/schemas.py

@@ -62,6 +62,35 @@ class CreateClusterRequest(BaseModel):
     credentials_ref: str = Field(min_length=1, max_length=128)
     engine_config: dict[str, Any] | None = None
     notes: str | None = Field(default=None, max_length=2000)
+    target_filter: str | None = Field(
+        default=None,
+        min_length=1,
+        max_length=256,
+        description=(
+            "Optional glob pattern (fnmatch.fnmatchcase: *, ?, [seq], [!seq]; "
+            "no brace expansion). Scopes GET /clusters/{id}/targets to "
+            "matching index names. Null = no filter."
+        ),
+    )
+
+    @field_validator("target_filter", mode="before")
+    @classmethod
+    def strip_target_filter(cls, v: Any) -> Any:
+        """Strip whitespace BEFORE min_length/max_length run (feat_cluster_target_filter FR-2).
+
+        Pydantic v2 default validator mode is ``after`` — that would let a
+        padded valid filter like ``"  " + "x"*256`` fail max_length=256 even
+        though the stripped value is exactly 256 chars. ``mode="before"`` runs
+        the strip first; ``min_length=1`` then catches the empty/whitespace-only
+        case and ``max_length=256`` runs on the stripped value.
+
+        Glob syntax is NOT validated — Python ``fnmatch`` is permissive (every
+        non-empty string is a valid glob). A pattern that matches nothing at
+        runtime surfaces via the create-study modal's empty-state, not a 422.
+        """
+        if isinstance(v, str):
+            return v.strip()
+        return v
 
     @field_validator("base_url")
     @classmethod
@@ -103,6 +132,7 @@ class ClusterDetail(BaseModel):
     auth_kind: AuthKind
     engine_config: dict[str, Any] | None = None
     notes: str | None = None
+    target_filter: str | None = None
     created_at: datetime
     health_check: HealthCheckResult
 
@@ -116,6 +146,7 @@ class ClusterSummary(BaseModel):
     environment: Environment
     base_url: str
     auth_kind: AuthKind
+    target_filter: str | None = None
     created_at: datetime
     health_check: HealthCheckResult
 

backend/app/db/models/cluster.py

@@ -75,6 +75,11 @@ class Cluster(Base):
     notes: Mapped[str | None] = mapped_column(String, nullable=True)
     """Operator notes (free-form, max 2000 chars at the API layer)."""
 
+    target_filter: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    """Operator-supplied glob pattern (``fnmatch.fnmatchcase`` syntax) scoping
+    ``list_targets()`` to matching index names. ``NULL`` = no filter (default,
+    backward-compatible). Trimmed at the API layer; stored verbatim otherwise."""
+
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()
     )

backend/app/services/cluster.py

@@ -92,6 +92,7 @@ async def register_cluster(
     credentials_ref: str,
     engine_config: dict[str, Any] | None,
     notes: str | None,
+    target_filter: str | None = None,
 ) -> tuple[Cluster, HealthStatus]:
     """Probe → insert (or revive) → cache. Reject if the cluster is unreachable.
 
@@ -167,6 +168,7 @@ async def register_cluster(
             credentials_ref=credentials_ref,
             engine_config=cfg or None,
             notes=notes,
+            target_filter=target_filter,
         )
     else:
         cluster = await repo.create_cluster(
@@ -180,6 +182,7 @@ async def register_cluster(
             credentials_ref=credentials_ref,
             engine_config=cfg or None,
             notes=notes,
+            target_filter=target_filter,
         )
     await db.commit()
     await write_cached_health(redis, cluster.id, health)

backend/tests/contract/test_clusters_api_contract.py

@@ -75,6 +75,102 @@ def test_create_cluster_request_validates_scheme() -> None:
         )
 
 
+def _make_cluster_request(**overrides: object) -> CreateClusterRequest:
+    """Build a minimal valid CreateClusterRequest for validator-focused tests."""
+    fields: dict[str, object] = {
+        "name": "cluster",
+        "engine_type": "elasticsearch",
+        "environment": "dev",
+        "base_url": "http://elasticsearch:9200",
+        "auth_kind": "es_basic",
+        "credentials_ref": "ref",
+    }
+    fields.update(overrides)
+    return CreateClusterRequest(**fields)
+
+
+def test_target_filter_present_on_cluster_pydantic_schemas() -> None:
+    """feat_cluster_target_filter plan §3.3: the new ``target_filter`` field
+    must appear in the JSON schemas of CreateClusterRequest, ClusterDetail,
+    and ClusterSummary. Validator-only tests (below) prove runtime behavior;
+    this test locks the public schema surface so a future regression that
+    drops the field from one of the response models fails the contract
+    suite, not just the downstream UI types-regen step.
+
+    Reads ``model_json_schema()`` directly — equivalent to the OpenAPI shape
+    FastAPI emits for these Pydantic models, without needing to instantiate
+    the FastAPI app (which would require Settings + DATABASE_URL_FILE).
+    """
+    req = CreateClusterRequest.model_json_schema()["properties"]["target_filter"]
+    # Nullable str — Pydantic v2 emits anyOf[{type: string}, {type: null}]
+    assert any(
+        opt.get("type") == "string" and opt.get("maxLength") == 256 for opt in req["anyOf"]
+    ), f"CreateClusterRequest.target_filter missing string/maxLength=256: {req}"
+    assert any(opt.get("type") == "null" for opt in req["anyOf"]), (
+        f"CreateClusterRequest.target_filter not nullable: {req}"
+    )
+
+    for cls, name in ((ClusterDetail, "ClusterDetail"), (ClusterSummary, "ClusterSummary")):
+        props = cls.model_json_schema()["properties"]
+        assert "target_filter" in props, f"{name} missing target_filter property"
+        opts = props["target_filter"]["anyOf"]
+        assert any(opt.get("type") == "string" for opt in opts), (
+            f"{name}.target_filter missing string type: {opts}"
+        )
+        assert any(opt.get("type") == "null" for opt in opts), (
+            f"{name}.target_filter not nullable: {opts}"
+        )
+
+
+class TestTargetFilterValidator:
+    """feat_cluster_target_filter Story B3 — request validator (FR-2 + Finding #5)."""
+
+    def test_omitted_defaults_to_none(self) -> None:
+        req = _make_cluster_request()
+        assert req.target_filter is None
+
+    def test_valid_pattern_accepted(self) -> None:
+        req = _make_cluster_request(target_filter="products*")
+        assert req.target_filter == "products*"
+
+    def test_padded_string_strips_to_canonical(self) -> None:
+        """Finding #5: mode='before' validator strips before min/max_length run."""
+        req = _make_cluster_request(target_filter="  products*  ")
+        assert req.target_filter == "products*"
+
+    def test_padded_at_max_length_boundary_passes(self) -> None:
+        """Padded 256-char filter MUST pass — proves max_length sees stripped value.
+
+        Without mode='before' the leading/trailing whitespace would push the
+        wire length past 256 and falsely 422.
+        """
+        padded = "  " + ("x" * 256) + "  "
+        req = _make_cluster_request(target_filter=padded)
+        assert req.target_filter == "x" * 256
+
+    def test_empty_string_rejected(self) -> None:
+        import pytest
+        from pydantic import ValidationError
+
+        with pytest.raises(ValidationError):
+            _make_cluster_request(target_filter="")
+
+    def test_whitespace_only_rejected(self) -> None:
+        """Whitespace strips to empty, then min_length=1 catches it."""
+        import pytest
+        from pydantic import ValidationError
+
+        with pytest.raises(ValidationError):
+            _make_cluster_request(target_filter="   ")
+
+    def test_over_256_chars_rejected(self) -> None:
+        import pytest
+        from pydantic import ValidationError
+
+        with pytest.raises(ValidationError):
+            _make_cluster_request(target_filter="x" * 257)
+
+
 def test_run_query_request_caps_top_k_at_1000() -> None:
     """``top_k`` is bounded at 1000 (spec FR-6 / Decision Log 2026-05-09)."""
     import pytest

backend/tests/integration/fixtures/stub_adapter.py

@@ -54,7 +54,15 @@ async def health_check(self, *, request_id: str | None = None) -> HealthStatus:
             checked_at=datetime.now(UTC).isoformat(),
         )
 
-    async def list_targets(self, *, request_id: str | None = None) -> list[TargetInfo]:
+    async def list_targets(
+        self,
+        *,
+        request_id: str | None = None,
+        target_filter: str | None = None,
+    ) -> list[TargetInfo]:
+        # `target_filter` accepted to match Protocol signature (feat_cluster_target_filter
+        # FR-3); stub returns hardcoded data so the kwarg is intentionally unused.
+        del target_filter  # silence unused-arg lint
         return [TargetInfo(name="stub-index", doc_count=100)]
 
     async def get_schema(self, target: str, *, request_id: str | None = None) -> Schema: