Merge pull request #18 from AlleksDev/search

.env.example

@@ -8,10 +8,10 @@ MAIN_API_PLACES_NEARBY_PATH=/api/v1/places/nearby
 MAIN_API_POSTS_SNAPSHOT_PATH=/api/v1/internal/posts/snapshot
 MAIN_API_POSTS_CHANGES_PATH=/api/v1/internal/posts/changes
 MAIN_API_FEED_INTERACTIONS_PATH=/api/v1/internal/feed/interactions/changes
-MAIN_API_USERS_SEARCH_PATH=/api/v1/users/search
-MAIN_API_CLUBS_SEARCH_PATH=/api/v1/clubs/search
-MAIN_API_GROUPS_SEARCH_PATH=/api/v1/groups/search
-MAIN_API_EVENTS_SEARCH_PATH=/api/v1/events/search
+MAIN_API_USERS_SNAPSHOT_PATH=/api/v1/internal/search/users/snapshot
+MAIN_API_CLUBS_SNAPSHOT_PATH=/api/v1/internal/search/clubs/snapshot
+MAIN_API_GROUPS_SNAPSHOT_PATH=/api/v1/internal/search/groups/snapshot
+MAIN_API_EVENTS_SNAPSHOT_PATH=/api/v1/internal/search/events/snapshot
 MAIN_API_INTERNAL_TOKEN=
 MAIN_API_AUTH_TOKEN=
 SEARCH_INTERNAL_TOKEN=
@@ -26,6 +26,12 @@ MAIN_API_PLACES_PAGINATION_MODE=cursor
 MAIN_API_POSTS_PAGINATION_MODE=cursor
 MAIN_API_SEARCH_PAGINATION_MODE=cursor
 GLOBAL_SEARCH_NEARBY_BOOST=0.12
+PUBLIC_GLOBAL_SEARCH_ENABLED=true
+GLOBAL_SEARCH_MIN_SEMANTIC_SCORE=0.30
+GLOBAL_SEARCH_MIN_LEXICAL_SCORE=0.05
+# Overrides opcionales. Cada recurso incluido debe definir ambos valores.
+GLOBAL_SEARCH_RESOURCE_THRESHOLDS_JSON={}
+GLOBAL_SEARCH_THRESHOLD_POLICY_VERSION=global-search-relevance-v1
 
 GROQ_API_KEY=
 GROQ_MODEL=llama-3.1-8b-instant
@@ -58,6 +64,10 @@ SEMANTIC_RELEVANCE_THRESHOLD=0.50
 
 LOG_LEVEL=INFO
 REQUEST_TIMEOUT_SECONDS=10
+# Limites locales por IP y ruta. El gateway puede imponer limites distribuidos menores.
+RATE_LIMIT_REQUESTS_PER_WINDOW=600
+INTERNAL_RATE_LIMIT_REQUESTS_PER_WINDOW=120
+RATE_LIMIT_WINDOW_SECONDS=60
 LLM_TIMEOUT_SECONDS=12
 MAX_LLM_CONCURRENT_REQUESTS=4
 MAX_REQUEST_BODY_BYTES=262144

README.md

@@ -140,12 +140,36 @@ python -m app.jobs.sync_search_embeddings --resource all
 Los jobs calculan un `content_hash` versionado con el contenido, modelo, version y
 dimension. Un cambio de modelo fuerza la regeneracion aunque el texto no haya cambiado.
 
+La sincronizacion de users, clubs, groups y events consume exclusivamente los snapshots
+internos de la API principal:
+
+```text
+/api/v1/internal/search/users/snapshot
+/api/v1/internal/search/clubs/snapshot
+/api/v1/internal/search/groups/snapshot
+/api/v1/internal/search/events/snapshot
+```
+
+Estos clientes requieren `MAIN_API_INTERNAL_TOKEN`; nunca usan un JWT de usuario ni
+hacen fallback a `MAIN_API_AUTH_TOKEN`.
+
 ## Busqueda Global Hibrida
 
-`POST /search` consulta lugares, posts, usuarios, clubes, grupos y eventos en paralelo.
+La app movil debe llamar `POST /api/v1/search` en la API principal con su token de
+sesion. Go valida permisos y filtros de negocio, llama al endpoint interno
+`POST /internal/search/candidates` de NLP, hidrata los IDs y devuelve los recursos
+completos. NLP solo calcula candidatos y puntajes; no es la fuente de verdad ni devuelve
+datos de presentacion.
+
 El query genera un solo embedding FastText. Cada proveedor combina similitud coseno con
-full-text search de PostgreSQL mediante Reciprocal Rank Fusion y el caso de uso devuelve
-un `top_results` diversificado mas una seccion independiente por recurso.
+full-text search de PostgreSQL mediante Reciprocal Rank Fusion. Los resultados que no
+superan el umbral semantico ni el lexical se descartan: no se agregan recursos de relleno.
+Los eventos se descartan antes del ranking cuando
+`start_time + duration_minutes <= as_of`.
+
+`POST /search` permanece como endpoint publico de compatibilidad temporal. Puede
+deshabilitarse con `PUBLIC_GLOBAL_SEARCH_ENABLED=false` cuando la app movil ya use solo
+la API principal.
 
 ```json
 {
@@ -157,12 +181,9 @@ un `top_results` diversificado mas una seccion independiente por recurso.
 }
 ```
 
-`requester_id` debe provenir de la identidad autenticada por la API principal, nunca de
-texto libre enviado directamente por la app. Cuando se envia, la llamada tambien debe
-incluir `Authorization: Bearer <SEARCH_INTERNAL_TOKEN>`. Este header solo se requiere cuando la solicitud incluye
-`requester_id`; las busquedas publicas no requieren autenticacion. Los grupos privados solo son recuperables
-por su creador, miembros agregados o invitados. Clubes privados y eventos no publicos
-permanecen ocultos mientras la API principal no entregue una lista de usuarios autorizados.
+La API principal obtiene la identidad desde el token de login; la app no debe enviar
+`requester_id` como autoridad. El endpoint interno de NLP exige
+`Authorization: Bearer <NLP_SERVICE_TOKEN>` y devuelve exclusivamente IDs y puntajes.
 
 El indice de usuarios excluye correo, fecha de nacimiento, genero y ubicacion actual. Los
 IDs de tags de eventos se conservan como metadatos; para aportar significado semantico la
@@ -197,6 +218,24 @@ sql/verify_post_feed_v2.sql
 `sql/rollback_post_feed_v1.sql` es el rollback destructivo de emergencia. No se debe
 ejecutar ninguna de estas migraciones desde la API ni desde un job.
 
+La busqueda coordinada por la API principal requiere, para una BD existente:
+
+```text
+sql/migrations/20260712_01_global_search_candidate_filters.sql
+sql/verify_global_search_candidate_filters.sql
+```
+
+La migracion reemplaza de forma transaccional `search_resource_embeddings` y agrega dos
+helpers de parseo seguro para timestamps y duraciones. No crea ni modifica tablas,
+columnas o indices. El segundo archivo es de solo lectura, falla con una excepcion si
+detecta deriva y debe ejecutarse despues para comprobar el contrato.
+
+`/ready` exige la firma y los marcadores V2 de search; una funcion heredada ya no puede
+declarar el servicio listo. Las conexiones y consultas pgvector usan
+`REQUEST_TIMEOUT_SECONDS`, y las rutas tienen un limite local configurable con
+`RATE_LIMIT_REQUESTS_PER_WINDOW`, `INTERNAL_RATE_LIMIT_REQUESTS_PER_WINDOW` y
+`RATE_LIMIT_WINDOW_SECONDS`.
+
 Para una BD vacía o una BD existente que ya use `VECTOR(300)`, puede ejecutarse
 `sql/new_pgvector_schema.sql` desde pgAdmin. El archivo es convergente e incluye
 el contrato base, feed V1, correcciones V2, propietarios y permisos. No convierte

app/modules/clubs/infrastructure/main_api_club_source.py

@@ -13,7 +13,7 @@ class MainApiClubsClient(PagedMainApiSearchClient):
     def __init__(self, settings: Settings) -> None:
         super().__init__(
             settings=settings,
-            path=settings.main_api_clubs_search_path,
+            path=settings.main_api_clubs_snapshot_path,
             collection_keys=("clubs",),
             mapper=club_to_source_record,
             page_limit=settings.main_api_search_page_limit,

app/modules/events/infrastructure/main_api_event_source.py

@@ -1,3 +1,4 @@
+from datetime import UTC, datetime
 from typing import Any, AsyncIterator
 
 from app.modules.events.infrastructure.semantic_document import (
@@ -13,7 +14,7 @@ class MainApiEventsClient(PagedMainApiSearchClient):
     def __init__(self, settings: Settings) -> None:
         super().__init__(
             settings=settings,
-            path=settings.main_api_events_search_path,
+            path=settings.main_api_events_snapshot_path,
             collection_keys=("events",),
             mapper=event_to_source_record,
             page_limit=settings.main_api_search_page_limit,
@@ -31,11 +32,16 @@ def event_to_source_record(event: dict[str, Any]) -> SearchSourceRecord | None:
         return None
     title = str(first_present(event, "title", "name", default="")).strip()
     description = str(first_present(event, "description", "summary", default="")).strip()
-    tag_values = first_present(event, "tags", "tag_ids", "tagIds", default=[])
-    tag_ids = as_text_list(tag_values)
+    tag_names = as_text_list(first_present(event, "tags", default=[]))
+    tag_ids = as_text_list(first_present(event, "tag_ids", "tagIds", default=[]))
     # Only textual tag names are semantic. UUID/numeric IDs stay as filters/metadata.
-    tag_names = [tag for tag in tag_ids if not _looks_like_identifier(tag)]
-    is_active = bool(first_present(event, "is_active", "isActive", default=True))
+    tag_names = [tag for tag in tag_names if not _looks_like_identifier(tag)]
+    start_time, duration_minutes = _normalized_temporal_metadata(event)
+    temporal_metadata_valid = start_time is not None and duration_minutes is not None
+    is_active = (
+        bool(first_present(event, "is_active", "isActive", default=True))
+        and temporal_metadata_valid
+    )
     is_public = bool(first_present(event, "is_public", "isPublic", default=True))
     document = build_event_search_document(title, tag_names, description)
     return make_search_record(
@@ -49,16 +55,15 @@ def event_to_source_record(event: dict[str, Any]) -> SearchSourceRecord | None:
             "description": description[:300],
             "club_id": first_present(event, "club_id", "clubId"),
             "place_id": first_present(event, "place_id", "placeId"),
-            "start_time": first_present(event, "start_time", "startTime"),
-            "duration_minutes": first_present(
-                event, "duration_minutes", "durationMinutes"
-            ),
+            "start_time": start_time,
+            "duration_minutes": duration_minutes,
             "max_attendees": first_present(event, "max_attendees", "maxAttendees"),
             "is_public": is_public,
             "cover_image_url": first_present(event, "cover_image_url", "coverImageUrl"),
             "tags": tag_names,
             "tag_ids": tag_ids,
             "is_active": is_active,
+            "temporal_metadata_valid": temporal_metadata_valid,
             "document_version": EVENT_SEARCH_DOCUMENT_VERSION,
         },
     )
@@ -67,3 +72,30 @@ def event_to_source_record(event: dict[str, Any]) -> SearchSourceRecord | None:
 def _looks_like_identifier(value: str) -> bool:
     compact = value.replace("-", "")
     return compact.isdigit() or (len(compact) == 32 and all(c in "0123456789abcdefABCDEF" for c in compact))
+
+
+def _normalized_temporal_metadata(
+    event: dict[str, Any],
+) -> tuple[str | None, int | None]:
+    raw_start = first_present(event, "start_time", "startTime")
+    if not isinstance(raw_start, str) or not raw_start.strip():
+        return None, None
+    try:
+        parsed_start = datetime.fromisoformat(
+            raw_start.strip().replace("Z", "+00:00")
+        )
+    except ValueError:
+        return None, None
+    if parsed_start.utcoffset() is None:
+        return None, None
+
+    raw_duration = first_present(event, "duration_minutes", "durationMinutes")
+    if isinstance(raw_duration, bool):
+        return None, None
+    try:
+        parsed_duration = int(raw_duration)
+    except (TypeError, ValueError):
+        return None, None
+    if parsed_duration <= 0 or parsed_duration > 2_147_483_647:
+        return None, None
+    return parsed_start.astimezone(UTC).isoformat(), parsed_duration

app/modules/groups/infrastructure/main_api_group_source.py

@@ -13,7 +13,7 @@ class MainApiGroupsClient(PagedMainApiSearchClient):
     def __init__(self, settings: Settings) -> None:
         super().__init__(
             settings=settings,
-            path=settings.main_api_groups_search_path,
+            path=settings.main_api_groups_snapshot_path,
             collection_keys=("groups",),
             mapper=group_to_source_record,
             page_limit=settings.main_api_search_page_limit,

app/modules/search/api/dependencies.py

@@ -2,6 +2,9 @@ from functools import lru_cache
 
 from app.modules.search.application.ports.search_provider import SearchProvider
 from app.modules.search.application.use_cases.search_all import SearchAllUseCase
+from app.modules.search.application.use_cases.search_candidates import (
+    SearchCandidatesUseCase,
+)
 from app.modules.search.domain.models import ALL_SEARCH_RESOURCE_TYPES, SearchResourceType
 from app.modules.search.infrastructure.mock_provider import (
     MockHybridSearchProvider,
@@ -11,6 +14,8 @@ from app.modules.search.infrastructure.pgvector_provider import (
     PgvectorHybridSearchProvider,
     PgvectorPlaceSearchProvider,
 )
+from app.modules.search.infrastructure.relevance_policy import build_relevance_policy
+from app.modules.search.domain.relevance import SearchRelevancePolicy
 from app.modules.places.infrastructure.main_api_nearby_place_provider import (
     MainApiNearbyPlaceProvider,
 )
@@ -19,17 +24,27 @@ from app.shared.dependencies import get_embedding_provider
 from app.shared.vector_store.aws_pgvector import AwsPgvectorClient
 
 
+@lru_cache
+def get_search_relevance_policy() -> SearchRelevancePolicy:
+    return build_relevance_policy(get_settings())
+
+
 @lru_cache
 def get_global_search_providers() -> tuple[SearchProvider, ...]:
     settings = get_settings()
     embedding_provider = get_embedding_provider()
+    relevance_policy = get_search_relevance_policy()
     if settings.vector_store_provider == "aws_pgvector":
         vector_client = AwsPgvectorClient(settings, role="reader")
         return tuple(
             (
-                PgvectorPlaceSearchProvider(vector_client)
+                PgvectorPlaceSearchProvider(vector_client, relevance_policy)
                 if resource_type == SearchResourceType.PLACES
-                else PgvectorHybridSearchProvider(resource_type, vector_client)
+                else PgvectorHybridSearchProvider(
+                    resource_type,
+                    vector_client,
+                    relevance_policy,
+                )
             )
             for resource_type in ALL_SEARCH_RESOURCE_TYPES
         )
@@ -40,6 +55,7 @@ def get_global_search_providers() -> tuple[SearchProvider, ...]:
             resource_type=resource_type,
             embedding_provider=embedding_provider,
             documents=documents[resource_type],
+            relevance_policy=relevance_policy,
         )
         for resource_type in ALL_SEARCH_RESOURCE_TYPES
     )
@@ -52,3 +68,14 @@ def get_search_all_use_case() -> SearchAllUseCase:
         providers=list(get_global_search_providers()),
         nearby_place_provider=MainApiNearbyPlaceProvider(get_settings()),
     )
+
+
+@lru_cache
+def get_search_candidates_use_case() -> SearchCandidatesUseCase:
+    policy = get_search_relevance_policy()
+    return SearchCandidatesUseCase(
+        embedding_provider=get_embedding_provider(),
+        providers=list(get_global_search_providers()),
+        policy_version=policy.version,
+        nearby_place_provider=MainApiNearbyPlaceProvider(get_settings()),
+    )

app/modules/search/api/internal_auth.py

@@ -0,0 +1,21 @@
+import hmac
+
+from fastapi import Header, HTTPException, status
+
+from app.shared.config.settings import get_settings
+
+
+def require_search_service_token(
+    authorization: str | None = Header(default=None, alias="Authorization"),
+) -> None:
+    expected = get_settings().post_feed_internal_token
+    if not expected:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="NLP_SERVICE_TOKEN no configurado",
+        )
+    scheme, separator, received = (authorization or "").partition(" ")
+    if not separator or scheme.casefold() != "bearer" or not received.strip():
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
+    if not hmac.compare_digest(received.strip(), expected):
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

app/modules/search/api/internal_router.py

@@ -0,0 +1,74 @@
+import asyncio
+
+from fastapi import APIRouter, Depends, HTTPException
+
+from app.modules.search.api.cursor import (
+    build_search_cursor_context,
+    decode_search_cursor,
+)
+from app.modules.search.api.dependencies import get_search_candidates_use_case
+from app.modules.search.api.internal_auth import require_search_service_token
+from app.modules.search.api.internal_schemas import (
+    InternalSearchCandidatesRequest,
+    InternalSearchCandidatesResponse,
+    candidates_result_to_schema,
+)
+from app.modules.search.application.use_cases.search_candidates import (
+    SearchCandidatesUseCase,
+)
+from app.shared.config.settings import get_settings
+from app.shared.security.rate_limit import rate_limit_placeholder
+
+router = APIRouter(
+    prefix="/internal/search",
+    tags=["internal-search"],
+    dependencies=[
+        Depends(rate_limit_placeholder),
+        Depends(require_search_service_token),
+    ],
+)
+
+
+@router.post(
+    "/candidates",
+    response_model=InternalSearchCandidatesResponse,
+)
+async def search_candidates(
+    payload: InternalSearchCandidatesRequest,
+    use_case: SearchCandidatesUseCase = Depends(get_search_candidates_use_case),
+) -> InternalSearchCandidatesResponse:
+    settings = get_settings()
+    cursor_context = build_search_cursor_context(
+        payload.cursor_context_payload(
+            nearby_boost=settings.global_search_nearby_boost,
+            policy_version=settings.global_search_threshold_policy_version,
+        )
+    )
+    try:
+        offsets = {
+            resource_type: decode_search_cursor(
+                cursor=cursor,
+                resource_type=resource_type,
+                context_fingerprint=cursor_context,
+            )
+            for resource_type, cursor in payload.cursors.items()
+        }
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+    try:
+        async with asyncio.timeout(settings.request_timeout_seconds):
+            result = await use_case.execute(
+                query=payload.query,
+                resource_types=tuple(payload.resource_types),
+                candidate_limit_per_type=payload.candidate_limit_per_type,
+                as_of=payload.as_of,
+                offsets=offsets,
+                criteria=payload.to_domain_criteria(settings.global_search_nearby_boost),
+                cursor_context=cursor_context,
+            )
+    except TimeoutError as exc:
+        raise HTTPException(
+            status_code=503,
+            detail="Internal search timed out",
+        ) from exc
+    return candidates_result_to_schema(result)

app/modules/search/api/internal_schemas.py

@@ -0,0 +1,169 @@
+from datetime import UTC, datetime
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+
+from app.modules.search.api.cursor import MAX_SEARCH_OFFSET, encode_search_cursor
+from app.modules.search.api.schemas import (
+    GlobalSearchFiltersSchema,
+    SearchLocationSchema,
+)
+from app.modules.search.domain.filters import SearchCriteria
+from app.modules.search.domain.models import (
+    SearchCandidate,
+    SearchCandidatesResult,
+    SearchResourceType,
+)
+from app.modules.search.domain.query import normalize_search_query
+
+
+class InternalSearchCandidatesRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    query: str = Field(..., min_length=1, max_length=500)
+    resource_types: list[SearchResourceType] = Field(..., min_length=1, max_length=6)
+    candidate_limit_per_type: int = Field(default=20, ge=1, le=500)
+    top_limit: int = Field(default=30, ge=1, le=3000)
+    cursors: dict[SearchResourceType, str] = Field(default_factory=dict)
+    as_of: datetime
+    location: SearchLocationSchema | None = None
+    filters: GlobalSearchFiltersSchema = Field(default_factory=GlobalSearchFiltersSchema)
+
+    @field_validator("query")
+    @classmethod
+    def validate_query(cls, value: str) -> str:
+        if not normalize_search_query(value):
+            raise ValueError("query must not be blank")
+        return value
+
+    @model_validator(mode="after")
+    def validate_context(self) -> "InternalSearchCandidatesRequest":
+        if self.as_of.utcoffset() is None:
+            raise ValueError("as_of must include a timezone")
+        if len(self.resource_types) != len(set(self.resource_types)):
+            raise ValueError("resource_types must not contain duplicates")
+        unexpected = set(self.cursors) - set(self.resource_types)
+        if unexpected:
+            names = ", ".join(sorted(item.value for item in unexpected))
+            raise ValueError(f"cursor resources were not requested: {names}")
+        return self
+
+    def to_domain_criteria(self, nearby_boost: float) -> SearchCriteria:
+        return SearchCriteria(
+            filters=self.filters.to_domain(),
+            location=self.location.to_domain() if self.location else None,
+            nearby_boost=nearby_boost,
+            event_active_at=self.as_of,
+        )
+
+    def cursor_context_payload(
+        self,
+        nearby_boost: float,
+        policy_version: str,
+    ) -> dict[str, object]:
+        filters = self.filters.model_dump(mode="json")
+        for key in ("categories", "price_ranges", "tags", "user_roles"):
+            filters[key] = sorted(
+                {
+                    str(value).strip().casefold()
+                    for value in filters[key]
+                    if str(value).strip()
+                }
+            )
+        for key in ("city", "state"):
+            if filters[key]:
+                filters[key] = str(filters[key]).strip().casefold()
+        for key in ("published_from", "published_to", "event_from", "event_to"):
+            value = getattr(self.filters, key)
+            filters[key] = value.astimezone(UTC).isoformat() if value else None
+        return {
+            "query": normalize_search_query(self.query),
+            "candidate_limit_per_type": self.candidate_limit_per_type,
+            "top_limit": self.top_limit,
+            "as_of": self.as_of.astimezone(UTC).isoformat(),
+            "location": self.location.model_dump(mode="json") if self.location else None,
+            "filters": filters,
+            "nearby_boost": nearby_boost,
+            "policy_version": policy_version,
+        }
+
+
+class InternalSearchCandidateSchema(BaseModel):
+    id: str
+    resource_type: SearchResourceType
+    score: float
+    semantic_score: float | None = None
+    lexical_score: float | None = None
+
+
+class InternalSearchPaginationSchema(BaseModel):
+    returned_count: int
+    has_more: bool
+    next_cursor: str | None = None
+
+
+class InternalSearchMetadataSchema(BaseModel):
+    strategy: str
+    failed_resources: dict[str, str]
+    threshold_policy_version: str
+
+
+class InternalSearchCandidatesResponse(BaseModel):
+    query: str
+    sections: dict[str, list[InternalSearchCandidateSchema]]
+    pagination: dict[str, InternalSearchPaginationSchema]
+    metadata: InternalSearchMetadataSchema
+
+
+def candidates_result_to_schema(
+    result: SearchCandidatesResult,
+) -> InternalSearchCandidatesResponse:
+    return InternalSearchCandidatesResponse(
+        query=result.query,
+        sections={
+            resource_type.value: [_candidate_to_schema(item) for item in items]
+            for resource_type, items in result.sections.items()
+        },
+        pagination={
+            resource_type.value: InternalSearchPaginationSchema(
+                returned_count=page.returned_count,
+                has_more=(
+                    page.has_more
+                    and page.next_offset is not None
+                    and page.next_offset <= MAX_SEARCH_OFFSET
+                ),
+                next_cursor=(
+                    encode_search_cursor(
+                        resource_type,
+                        result.cursor_context,
+                        page.next_offset,
+                    )
+                    if page.next_offset is not None
+                    and page.next_offset <= MAX_SEARCH_OFFSET
+                    else None
+                ),
+            )
+            for resource_type, page in result.pagination.items()
+        },
+        metadata=InternalSearchMetadataSchema(
+            strategy="hybrid_thresholded_candidates_v2",
+            failed_resources={
+                resource_type.value: error
+                for resource_type, error in result.failed_resources.items()
+            },
+            threshold_policy_version=result.policy_version,
+        ),
+    )
+
+
+def _candidate_to_schema(item: SearchCandidate) -> InternalSearchCandidateSchema:
+    return InternalSearchCandidateSchema(
+        id=item.id,
+        resource_type=item.resource_type,
+        score=round(item.score, 6),
+        semantic_score=(
+            round(item.semantic_score, 6) if item.semantic_score is not None else None
+        ),
+        lexical_score=(
+            round(item.lexical_score, 6) if item.lexical_score is not None else None
+        ),
+    )

app/modules/search/api/public_router.py

@@ -0,0 +1,86 @@
+import secrets
+
+from fastapi import APIRouter, Depends, Header, HTTPException, status
+
+from app.modules.search.api.cursor import (
+    build_search_cursor_context,
+    decode_search_cursor,
+)
+from app.modules.search.api.dependencies import get_search_all_use_case
+from app.modules.search.api.schemas import (
+    GlobalSearchRequest,
+    GlobalSearchResponse,
+    result_to_schema,
+)
+from app.modules.search.application.use_cases.search_all import SearchAllUseCase
+from app.modules.search.domain.models import ALL_SEARCH_RESOURCE_TYPES
+from app.shared.config.settings import get_settings
+from app.shared.security.rate_limit import rate_limit_placeholder
+
+router = APIRouter(
+    tags=["search"],
+    dependencies=[Depends(rate_limit_placeholder)],
+)
+
+
+@router.post("/search", response_model=GlobalSearchResponse)
+async def search_all(
+    payload: GlobalSearchRequest,
+    authorization: str | None = Header(default=None, alias="Authorization"),
+    use_case: SearchAllUseCase = Depends(get_search_all_use_case),
+) -> GlobalSearchResponse:
+    settings = get_settings()
+    if not settings.public_global_search_enabled:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
+    if payload.requester_id:
+        expected_token = settings.search_internal_token
+        bearer_token = _extract_bearer_token(authorization)
+        if not expected_token or not bearer_token or not secrets.compare_digest(
+            bearer_token, expected_token
+        ):
+            raise HTTPException(
+                status_code=403,
+                detail="A trusted internal token is required for requester-scoped search",
+            )
+    resource_types = (
+        tuple(dict.fromkeys(payload.resource_types))
+        if payload.resource_types
+        else ALL_SEARCH_RESOURCE_TYPES
+    )
+    cursor_context = build_search_cursor_context(
+        payload.cursor_context_payload(
+            settings.global_search_nearby_boost,
+            settings.global_search_threshold_policy_version,
+        )
+    )
+    try:
+        offsets = {
+            resource_type: decode_search_cursor(
+                cursor=cursor,
+                resource_type=resource_type,
+                context_fingerprint=cursor_context,
+            )
+            for resource_type, cursor in payload.cursors.items()
+        }
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+    result = await use_case.execute(
+        query=payload.query,
+        resource_types=resource_types,
+        per_type_limit=payload.per_type_limit,
+        top_limit=payload.top_limit,
+        requester_id=str(payload.requester_id) if payload.requester_id else None,
+        offsets=offsets,
+        criteria=payload.to_domain_criteria(settings.global_search_nearby_boost),
+        cursor_context=cursor_context,
+    )
+    return result_to_schema(result)
+
+
+def _extract_bearer_token(authorization: str | None) -> str | None:
+    if not authorization:
+        return None
+    scheme, separator, token = authorization.partition(" ")
+    if not separator or scheme.lower() != "bearer" or not token.strip():
+        return None
+    return token.strip()

app/modules/search/api/router.py

@@ -1,81 +1,8 @@
-import secrets
+from fastapi import APIRouter
 
-from fastapi import APIRouter, Depends, Header, HTTPException
+from app.modules.search.api.internal_router import router as internal_router
+from app.modules.search.api.public_router import router as public_router
 
-from app.modules.search.api.dependencies import get_search_all_use_case
-from app.modules.search.api.cursor import (
-    build_search_cursor_context,
-    decode_search_cursor,
-)
-from app.modules.search.api.schemas import (
-    GlobalSearchRequest,
-    GlobalSearchResponse,
-    result_to_schema,
-)
-from app.modules.search.application.use_cases.search_all import SearchAllUseCase
-from app.modules.search.domain.models import ALL_SEARCH_RESOURCE_TYPES
-from app.shared.config.settings import get_settings
-from app.shared.security.rate_limit import rate_limit_placeholder
-
-router = APIRouter(
-    tags=["search"],
-    dependencies=[Depends(rate_limit_placeholder)],
-)
-
-
-@router.post("/search", response_model=GlobalSearchResponse)
-async def search_all(
-    payload: GlobalSearchRequest,
-    authorization: str | None = Header(default=None, alias="Authorization"),
-    use_case: SearchAllUseCase = Depends(get_search_all_use_case),
-) -> GlobalSearchResponse:
-    settings = get_settings()
-    if payload.requester_id:
-        expected_token = settings.search_internal_token
-        bearer_token = _extract_bearer_token(authorization)
-        if not expected_token or not bearer_token or not secrets.compare_digest(
-            bearer_token, expected_token
-        ):
-            raise HTTPException(
-                status_code=403,
-                detail="A trusted internal token is required for requester-scoped search",
-            )
-    resource_types = (
-        tuple(dict.fromkeys(payload.resource_types))
-        if payload.resource_types
-        else ALL_SEARCH_RESOURCE_TYPES
-    )
-    cursor_context = build_search_cursor_context(
-        payload.cursor_context_payload(settings.global_search_nearby_boost)
-    )
-    try:
-        offsets = {
-            resource_type: decode_search_cursor(
-                cursor=cursor,
-                resource_type=resource_type,
-                context_fingerprint=cursor_context,
-            )
-            for resource_type, cursor in payload.cursors.items()
-        }
-    except ValueError as exc:
-        raise HTTPException(status_code=422, detail=str(exc)) from exc
-    result = await use_case.execute(
-        query=payload.query,
-        resource_types=resource_types,
-        per_type_limit=payload.per_type_limit,
-        top_limit=payload.top_limit,
-        requester_id=str(payload.requester_id) if payload.requester_id else None,
-        offsets=offsets,
-        criteria=payload.to_domain_criteria(settings.global_search_nearby_boost),
-        cursor_context=cursor_context,
-    )
-    return result_to_schema(result)
-
-
-def _extract_bearer_token(authorization: str | None) -> str | None:
-    if not authorization:
-        return None
-    scheme, separator, token = authorization.partition(" ")
-    if not separator or scheme.lower() != "bearer" or not token.strip():
-        return None
-    return token.strip()
+router = APIRouter()
+router.include_router(public_router)
+router.include_router(internal_router)

app/modules/search/api/schemas.py

@@ -13,7 +13,7 @@ from app.modules.search.domain.filters import (
     SearchLocation,
 )
 from app.modules.search.domain.models import SearchAllResult, SearchHit, SearchResourceType
-from app.shared.nlp.preprocessing.text import prepare_for_embedding
+from app.modules.search.domain.query import normalize_search_query
 
 
 class SearchLocationSchema(BaseModel):
@@ -121,7 +121,11 @@ class GlobalSearchRequest(BaseModel):
             nearby_boost=nearby_boost,
         )
 
-    def cursor_context_payload(self, nearby_boost: float = 0.12) -> dict[str, Any]:
+    def cursor_context_payload(
+        self,
+        nearby_boost: float = 0.12,
+        policy_version: str = "global-search-relevance-v1",
+    ) -> dict[str, Any]:
         filters = self.filters.model_dump(mode="json")
         for key in ("categories", "price_ranges", "tags", "user_roles"):
             filters[key] = sorted(
@@ -140,7 +144,7 @@ class GlobalSearchRequest(BaseModel):
                 value.astimezone(UTC).isoformat() if value is not None else None
             )
         return {
-            "query": prepare_for_embedding(self.query),
+            "query": normalize_search_query(self.query),
             "per_type_limit": self.per_type_limit,
             "requester_id": str(self.requester_id) if self.requester_id else None,
             "location": (
@@ -148,6 +152,7 @@ class GlobalSearchRequest(BaseModel):
             ),
             "filters": filters,
             "nearby_boost": nearby_boost,
+            "policy_version": policy_version,
         }
 
 

app/modules/search/application/ports/query_embedding_provider.py

@@ -0,0 +1,6 @@
+from typing import Protocol
+
+
+class QueryEmbeddingProvider(Protocol):
+    def embed_text(self, text: str) -> list[float]:
+        """Build one embedding for a normalized search query."""

app/modules/search/application/use_cases/search_all.py

@@ -1,8 +1,13 @@
 import asyncio
+import logging
 import math
+from datetime import UTC, datetime
 
 from app.modules.search.application.ports.search_provider import SearchProvider
 from app.modules.search.application.ports.nearby_place_provider import NearbyPlaceProvider
+from app.modules.search.application.ports.query_embedding_provider import (
+    QueryEmbeddingProvider,
+)
 from app.modules.search.domain.filters import SearchCriteria
 from app.modules.search.domain.models import (
     ALL_SEARCH_RESOURCE_TYPES,
@@ -11,17 +16,15 @@ from app.modules.search.domain.models import (
     SearchResourceType,
     SearchSectionPagination,
 )
-from app.shared.logging.config import get_logger
-from app.shared.nlp.embeddings.base import EmbeddingProvider
-from app.shared.nlp.preprocessing.text import prepare_for_embedding
+from app.modules.search.domain.query import normalize_search_query
 
-logger = get_logger(__name__)
+logger = logging.getLogger(__name__)
 
 
 class SearchAllUseCase:
     def __init__(
         self,
-        embedding_provider: EmbeddingProvider,
+        embedding_provider: QueryEmbeddingProvider,
         providers: list[SearchProvider],
         nearby_place_provider: NearbyPlaceProvider | None = None,
     ) -> None:
@@ -40,10 +43,14 @@ class SearchAllUseCase:
         criteria: SearchCriteria | None = None,
         cursor_context: str = "",
     ) -> SearchAllResult:
-        normalized_query = prepare_for_embedding(query)
+        normalized_query = normalize_search_query(query)
         embedding = self._embedding_provider.embed_text(normalized_query)
         effective_offsets = offsets or {}
         effective_criteria = criteria or SearchCriteria()
+        if effective_criteria.event_active_at is None:
+            effective_criteria = effective_criteria.with_event_active_at(
+                datetime.now(UTC)
+            )
         if effective_criteria.location is not None:
             if self._nearby_place_provider is None:
                 raise RuntimeError("nearby place provider is not configured")

app/modules/search/application/use_cases/search_candidates.py

@@ -0,0 +1,138 @@
+import asyncio
+import logging
+from datetime import datetime
+
+from app.modules.search.application.ports.nearby_place_provider import (
+    NearbyPlaceProvider,
+)
+from app.modules.search.application.ports.query_embedding_provider import (
+    QueryEmbeddingProvider,
+)
+from app.modules.search.application.ports.search_provider import SearchProvider
+from app.modules.search.domain.filters import SearchCriteria
+from app.modules.search.domain.models import (
+    SearchCandidate,
+    SearchCandidatesResult,
+    SearchResourceType,
+    SearchSectionPagination,
+)
+from app.modules.search.domain.query import normalize_search_query
+
+logger = logging.getLogger(__name__)
+
+
+class SearchCandidatesUseCase:
+    def __init__(
+        self,
+        embedding_provider: QueryEmbeddingProvider,
+        providers: list[SearchProvider],
+        policy_version: str,
+        nearby_place_provider: NearbyPlaceProvider | None = None,
+    ) -> None:
+        self._embedding_provider = embedding_provider
+        self._providers = {provider.resource_type: provider for provider in providers}
+        self._nearby_place_provider = nearby_place_provider
+        self._policy_version = policy_version
+
+    async def execute(
+        self,
+        query: str,
+        resource_types: tuple[SearchResourceType, ...],
+        candidate_limit_per_type: int,
+        as_of: datetime,
+        offsets: dict[SearchResourceType, int] | None = None,
+        criteria: SearchCriteria | None = None,
+        cursor_context: str = "",
+    ) -> SearchCandidatesResult:
+        normalized_query = normalize_search_query(query)
+        embedding = self._embedding_provider.embed_text(normalized_query)
+        effective_offsets = offsets or {}
+        effective_criteria = (criteria or SearchCriteria()).with_event_active_at(as_of)
+        if effective_criteria.location is not None:
+            if self._nearby_place_provider is None:
+                raise RuntimeError("nearby place provider is not configured")
+            nearby_ids = await self._nearby_place_provider.get_nearby_place_ids(
+                latitude=effective_criteria.location.latitude,
+                longitude=effective_criteria.location.longitude,
+                radius_meters=effective_criteria.location.radius_meters,
+            )
+            effective_criteria = effective_criteria.with_nearby_place_ids(nearby_ids)
+
+        providers = [
+            self._providers[resource_type]
+            for resource_type in resource_types
+            if resource_type in self._providers
+        ]
+        outcomes = await asyncio.gather(
+            *[
+                provider.search(
+                    query=normalized_query,
+                    embedding=embedding,
+                    limit=candidate_limit_per_type + 1,
+                    offset=effective_offsets.get(provider.resource_type, 0),
+                    requester_id=None,
+                    criteria=effective_criteria,
+                )
+                for provider in providers
+            ],
+            return_exceptions=True,
+        )
+
+        sections: dict[SearchResourceType, list[SearchCandidate]] = {
+            resource_type: [] for resource_type in resource_types
+        }
+        pagination: dict[SearchResourceType, SearchSectionPagination] = {}
+        failures: dict[SearchResourceType, str] = {}
+        for provider, outcome in zip(providers, outcomes):
+            if isinstance(outcome, BaseException):
+                logger.error(
+                    "Internal search provider failed resource=%s error=%s",
+                    provider.resource_type,
+                    type(outcome).__name__,
+                )
+                failures[provider.resource_type] = type(outcome).__name__
+                pagination[provider.resource_type] = SearchSectionPagination(
+                    page_size=candidate_limit_per_type,
+                    returned_count=0,
+                    has_more=False,
+                )
+                continue
+            ranked = list(outcome)
+            page_hits = ranked[:candidate_limit_per_type]
+            has_more = len(ranked) > candidate_limit_per_type
+            current_offset = effective_offsets.get(provider.resource_type, 0)
+            sections[provider.resource_type] = [
+                SearchCandidate(
+                    id=hit.id,
+                    resource_type=hit.resource_type,
+                    score=hit.score,
+                    semantic_score=hit.semantic_score,
+                    lexical_score=hit.lexical_score,
+                )
+                for hit in page_hits
+            ]
+            pagination[provider.resource_type] = SearchSectionPagination(
+                page_size=candidate_limit_per_type,
+                returned_count=len(page_hits),
+                has_more=has_more,
+                next_offset=(current_offset + len(page_hits) if has_more else None),
+            )
+
+        for resource_type in resource_types:
+            pagination.setdefault(
+                resource_type,
+                SearchSectionPagination(
+                    page_size=candidate_limit_per_type,
+                    returned_count=0,
+                    has_more=False,
+                ),
+            )
+        return SearchCandidatesResult(
+            query=query,
+            normalized_query=normalized_query,
+            sections=sections,
+            pagination=pagination,
+            failed_resources=failures,
+            cursor_context=cursor_context,
+            policy_version=self._policy_version,
+        )

app/modules/search/domain/filters.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass, field, replace
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from enum import StrEnum
 from typing import Any, Sequence
 
@@ -62,10 +62,14 @@ class SearchCriteria:
     location: SearchLocation | None = None
     nearby_place_ids: frozenset[str] = frozenset()
     nearby_boost: float = 0.12
+    event_active_at: datetime | None = None
 
     def with_nearby_place_ids(self, place_ids: set[str]) -> "SearchCriteria":
         return replace(self, nearby_place_ids=frozenset(place_ids))
 
+    def with_event_active_at(self, value: datetime) -> "SearchCriteria":
+        return replace(self, event_active_at=value)
+
     def requires_extended_candidates(self, resource_type: SearchResourceType) -> bool:
         if not self.filters.is_empty():
             return True
@@ -86,6 +90,12 @@ def filter_and_rank_hits(
     for hit in hits:
         if not _matches_filters(resource_type, hit.metadata, criteria.filters):
             continue
+        if (
+            resource_type == SearchResourceType.EVENTS
+            and criteria.event_active_at is not None
+            and not _event_is_active(hit.metadata, criteria.event_active_at)
+        ):
+            continue
         is_nearby = _is_nearby(resource_type, hit, criteria.nearby_place_ids)
         location_applicable = resource_type in _LOCATION_AWARE_RESOURCES
         if (
@@ -178,6 +188,22 @@ def _matches_filters(
     return True
 
 
+def _event_is_active(metadata: dict[str, Any], active_at: datetime) -> bool:
+    start_time = _parse_datetime(metadata.get("start_time"))
+    if start_time is None:
+        return False
+    raw_duration = metadata.get("duration_minutes")
+    if isinstance(raw_duration, bool):
+        return False
+    try:
+        duration_minutes = int(raw_duration)
+    except (TypeError, ValueError):
+        return False
+    if duration_minutes <= 0:
+        return False
+    return start_time + timedelta(minutes=duration_minutes) > _normalized_datetime(active_at)
+
+
 def _is_nearby(
     resource_type: SearchResourceType,
     hit: SearchHit,

app/modules/search/domain/models.py

@@ -38,6 +38,26 @@ class SearchSectionPagination:
     next_offset: int | None = None
 
 
+@dataclass(frozen=True)
+class SearchCandidate:
+    id: str
+    resource_type: SearchResourceType
+    score: float
+    semantic_score: float | None = None
+    lexical_score: float | None = None
+
+
+@dataclass(frozen=True)
+class SearchCandidatesResult:
+    query: str
+    normalized_query: str
+    sections: dict[SearchResourceType, list[SearchCandidate]]
+    pagination: dict[SearchResourceType, SearchSectionPagination]
+    failed_resources: dict[SearchResourceType, str]
+    cursor_context: str
+    policy_version: str
+
+
 @dataclass(frozen=True)
 class SearchAllResult:
     query: str

app/modules/search/domain/query.py

@@ -0,0 +1,12 @@
+import re
+import unicodedata
+
+
+def normalize_search_query(value: str) -> str:
+    normalized = unicodedata.normalize("NFKD", (value or "").casefold())
+    without_accents = "".join(
+        character
+        for character in normalized
+        if not unicodedata.combining(character)
+    )
+    return re.sub(r"\s+", " ", without_accents).strip()

app/modules/search/domain/relevance.py

@@ -0,0 +1,69 @@
+import math
+from dataclasses import dataclass
+
+from app.modules.search.domain.models import SearchHit, SearchResourceType
+
+
+@dataclass(frozen=True)
+class SearchRelevanceThreshold:
+    semantic_min: float
+    lexical_min: float
+
+    def __post_init__(self) -> None:
+        if not math.isfinite(self.semantic_min) or not -1.0 <= self.semantic_min <= 1.0:
+            raise ValueError("semantic_min must be finite and between -1 and 1")
+        if not math.isfinite(self.lexical_min) or self.lexical_min < 0.0:
+            raise ValueError("lexical_min must be finite and non-negative")
+
+    def accepts(self, hit: SearchHit) -> bool:
+        semantic_match = (
+            hit.semantic_score is not None
+            and math.isfinite(hit.semantic_score)
+            and hit.semantic_score >= self.semantic_min
+        )
+        lexical_match = (
+            hit.lexical_score is not None
+            and math.isfinite(hit.lexical_score)
+            and hit.lexical_score >= self.lexical_min
+        )
+        return semantic_match or lexical_match
+
+
+@dataclass(frozen=True)
+class SearchRelevancePolicy:
+    thresholds: dict[SearchResourceType, SearchRelevanceThreshold]
+    version: str
+
+    def __post_init__(self) -> None:
+        missing = set(SearchResourceType) - set(self.thresholds)
+        extra = set(self.thresholds) - set(SearchResourceType)
+        if missing or extra:
+            raise ValueError("relevance policy must define every search resource")
+        if not self.version.strip() or len(self.version) > 64:
+            raise ValueError("relevance policy version is invalid")
+
+    def threshold_for(
+        self, resource_type: SearchResourceType
+    ) -> SearchRelevanceThreshold:
+        return self.thresholds[resource_type]
+
+    def accepts(self, hit: SearchHit) -> bool:
+        return self.threshold_for(hit.resource_type).accepts(hit)
+
+    @classmethod
+    def uniform(
+        cls,
+        semantic_min: float = 0.30,
+        lexical_min: float = 0.05,
+        version: str = "global-search-relevance-v1",
+    ) -> "SearchRelevancePolicy":
+        return cls(
+            thresholds={
+                resource_type: SearchRelevanceThreshold(
+                    semantic_min=semantic_min,
+                    lexical_min=lexical_min,
+                )
+                for resource_type in SearchResourceType
+            },
+            version=version,
+        )

app/modules/search/infrastructure/mock_provider.py

@@ -7,6 +7,7 @@ from app.modules.posts.infrastructure.mock_post_repository import SAMPLE_POSTS
 from app.modules.search.application.ports.search_provider import SearchProvider
 from app.modules.search.domain.filters import SearchCriteria, filter_and_rank_hits
 from app.modules.search.domain.models import SearchHit, SearchResourceType
+from app.modules.search.domain.relevance import SearchRelevancePolicy
 from app.shared.nlp.embeddings.base import EmbeddingProvider
 from app.shared.nlp.preprocessing.text import prepare_for_embedding
 
@@ -25,8 +26,10 @@ class MockHybridSearchProvider(SearchProvider):
         resource_type: SearchResourceType,
         embedding_provider: EmbeddingProvider,
         documents: list[MockSearchDocument],
+        relevance_policy: SearchRelevancePolicy | None = None,
     ) -> None:
         self.resource_type = resource_type
+        self._relevance_policy = relevance_policy or SearchRelevancePolicy.uniform()
         self._records = [
             (document, embedding_provider.embed_text(document.document))
             for document in documents
@@ -70,7 +73,8 @@ class MockHybridSearchProvider(SearchProvider):
                     metadata=metadata,
                 )
             )
-        ranked = filter_and_rank_hits(self.resource_type, hits, criteria)
+        relevant_hits = [hit for hit in hits if self._relevance_policy.accepts(hit)]
+        ranked = filter_and_rank_hits(self.resource_type, relevant_hits, criteria)
         return ranked[offset : offset + limit]
 
 
@@ -134,7 +138,11 @@ def get_mock_search_documents() -> dict[SearchResourceType, list[MockSearchDocum
                 id="event_birthday",
                 title="Fiesta de Cumpleanos",
                 document="fiesta cumpleanos evento independiente",
-                metadata={"is_public": True, "duration_minutes": 300},
+                metadata={
+                    "is_public": True,
+                    "start_time": "2099-07-12T20:00:00Z",
+                    "duration_minutes": 300,
+                },
             )
         ],
     }

app/modules/search/infrastructure/pgvector_provider.py

@@ -7,6 +7,7 @@ from app.modules.search.domain.filters import (
     filter_and_rank_hits,
 )
 from app.modules.search.domain.models import SearchHit, SearchResourceType
+from app.modules.search.domain.relevance import SearchRelevancePolicy
 from app.shared.vector_store.aws_pgvector import AwsPgvectorClient
 from app.shared.vector_store.models import VectorMatch
 
@@ -18,8 +19,13 @@ class PgvectorPlaceSearchProvider(SearchProvider):
 
     resource_type = SearchResourceType.PLACES
 
-    def __init__(self, vector_client: AwsPgvectorClient) -> None:
+    def __init__(
+        self,
+        vector_client: AwsPgvectorClient,
+        relevance_policy: SearchRelevancePolicy | None = None,
+    ) -> None:
         self._vector_client = vector_client
+        self._relevance_policy = relevance_policy or SearchRelevancePolicy.uniform()
 
     async def search(
         self,
@@ -48,7 +54,8 @@ class PgvectorPlaceSearchProvider(SearchProvider):
             _to_search_hit(self.resource_type, match)
             for match in matches
         ]
-        ranked = filter_and_rank_hits(self.resource_type, hits, criteria)
+        relevant_hits = [hit for hit in hits if self._relevance_policy.accepts(hit)]
+        ranked = filter_and_rank_hits(self.resource_type, relevant_hits, criteria)
         return ranked[offset : offset + limit]
 
 
@@ -57,9 +64,11 @@ class PgvectorHybridSearchProvider(SearchProvider):
         self,
         resource_type: SearchResourceType,
         vector_client: AwsPgvectorClient,
+        relevance_policy: SearchRelevancePolicy | None = None,
     ) -> None:
         self.resource_type = resource_type
         self._vector_client = vector_client
+        self._relevance_policy = relevance_policy or SearchRelevancePolicy.uniform()
 
     async def search(
         self,
@@ -78,6 +87,14 @@ class PgvectorHybridSearchProvider(SearchProvider):
         ):
             return []
         filters = {"is_active": True}
+        threshold = self._relevance_policy.threshold_for(self.resource_type)
+        filters["min_semantic_score"] = threshold.semantic_min
+        filters["min_lexical_score"] = threshold.lexical_min
+        if (
+            self.resource_type == SearchResourceType.EVENTS
+            and criteria.event_active_at is not None
+        ):
+            filters["event_active_at"] = criteria.event_active_at.isoformat()
         if requester_id:
             filters["requester_id"] = requester_id
         filters.update(_hybrid_metadata_filters(criteria, self.resource_type))
@@ -93,7 +110,8 @@ class PgvectorHybridSearchProvider(SearchProvider):
             _to_search_hit(self.resource_type, match)
             for match in matches
         ]
-        ranked = filter_and_rank_hits(self.resource_type, hits, criteria)
+        relevant_hits = [hit for hit in hits if self._relevance_policy.accepts(hit)]
+        ranked = filter_and_rank_hits(self.resource_type, relevant_hits, criteria)
         return ranked[offset : offset + limit]
 
 

app/modules/search/infrastructure/relevance_policy.py

@@ -0,0 +1,33 @@
+from app.modules.search.domain.models import SearchResourceType
+from app.modules.search.domain.relevance import (
+    SearchRelevancePolicy,
+    SearchRelevanceThreshold,
+)
+from app.shared.config.settings import Settings
+
+
+def build_relevance_policy(settings: Settings) -> SearchRelevancePolicy:
+    thresholds: dict[SearchResourceType, SearchRelevanceThreshold] = {}
+    for resource_type in SearchResourceType:
+        override = settings.global_search_resource_thresholds.get(
+            resource_type.value,
+            {},
+        )
+        thresholds[resource_type] = SearchRelevanceThreshold(
+            semantic_min=float(
+                override.get(
+                    "semantic_min",
+                    settings.global_search_min_semantic_score,
+                )
+            ),
+            lexical_min=float(
+                override.get(
+                    "lexical_min",
+                    settings.global_search_min_lexical_score,
+                )
+            ),
+        )
+    return SearchRelevancePolicy(
+        thresholds=thresholds,
+        version=settings.global_search_threshold_policy_version,
+    )

app/modules/users/infrastructure/main_api_user_source.py

@@ -13,7 +13,7 @@ class MainApiUsersClient(PagedMainApiSearchClient):
     def __init__(self, settings: Settings) -> None:
         super().__init__(
             settings=settings,
-            path=settings.main_api_users_search_path,
+            path=settings.main_api_users_snapshot_path,
             collection_keys=("users",),
             mapper=user_to_source_record,
             page_limit=settings.main_api_search_page_limit,

app/shared/config/settings.py

@@ -1,4 +1,5 @@
 from functools import lru_cache
+import math
 
 from pydantic import AliasChoices, Field, model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -39,17 +40,21 @@ class Settings(BaseSettings):
         default="/api/v1/internal/feed/interactions/changes",
         alias="MAIN_API_FEED_INTERACTIONS_PATH",
     )
-    main_api_users_search_path: str = Field(
-        default="/api/v1/users/search", alias="MAIN_API_USERS_SEARCH_PATH"
+    main_api_users_snapshot_path: str = Field(
+        default="/api/v1/internal/search/users/snapshot",
+        alias="MAIN_API_USERS_SNAPSHOT_PATH",
     )
-    main_api_clubs_search_path: str = Field(
-        default="/api/v1/clubs/search", alias="MAIN_API_CLUBS_SEARCH_PATH"
+    main_api_clubs_snapshot_path: str = Field(
+        default="/api/v1/internal/search/clubs/snapshot",
+        alias="MAIN_API_CLUBS_SNAPSHOT_PATH",
     )
-    main_api_groups_search_path: str = Field(
-        default="/api/v1/groups/search", alias="MAIN_API_GROUPS_SEARCH_PATH"
+    main_api_groups_snapshot_path: str = Field(
+        default="/api/v1/internal/search/groups/snapshot",
+        alias="MAIN_API_GROUPS_SNAPSHOT_PATH",
     )
-    main_api_events_search_path: str = Field(
-        default="/api/v1/events/search", alias="MAIN_API_EVENTS_SEARCH_PATH"
+    main_api_events_snapshot_path: str = Field(
+        default="/api/v1/internal/search/events/snapshot",
+        alias="MAIN_API_EVENTS_SNAPSHOT_PATH",
     )
     main_api_internal_token: str | None = Field(
         default=None,
@@ -85,6 +90,31 @@ class Settings(BaseSettings):
         le=1.0,
         alias="GLOBAL_SEARCH_NEARBY_BOOST",
     )
+    public_global_search_enabled: bool = Field(
+        default=True,
+        alias="PUBLIC_GLOBAL_SEARCH_ENABLED",
+    )
+    global_search_min_semantic_score: float = Field(
+        default=0.30,
+        ge=-1.0,
+        le=1.0,
+        alias="GLOBAL_SEARCH_MIN_SEMANTIC_SCORE",
+    )
+    global_search_min_lexical_score: float = Field(
+        default=0.05,
+        ge=0.0,
+        alias="GLOBAL_SEARCH_MIN_LEXICAL_SCORE",
+    )
+    global_search_resource_thresholds: dict[str, dict[str, float]] = Field(
+        default_factory=dict,
+        alias="GLOBAL_SEARCH_RESOURCE_THRESHOLDS_JSON",
+    )
+    global_search_threshold_policy_version: str = Field(
+        default="global-search-relevance-v1",
+        min_length=1,
+        max_length=64,
+        alias="GLOBAL_SEARCH_THRESHOLD_POLICY_VERSION",
+    )
 
     groq_api_key: str | None = Field(default=None, alias="GROQ_API_KEY")
     groq_model: str = Field(default="llama-3.1-8b-instant", alias="GROQ_MODEL")
@@ -154,7 +184,26 @@ class Settings(BaseSettings):
     )
 
     log_level: str = Field(default="INFO", alias="LOG_LEVEL")
-    request_timeout_seconds: int = Field(default=10, alias="REQUEST_TIMEOUT_SECONDS")
+    request_timeout_seconds: int = Field(
+        default=10,
+        gt=0,
+        alias="REQUEST_TIMEOUT_SECONDS",
+    )
+    rate_limit_requests_per_window: int = Field(
+        default=600,
+        ge=1,
+        alias="RATE_LIMIT_REQUESTS_PER_WINDOW",
+    )
+    internal_rate_limit_requests_per_window: int = Field(
+        default=120,
+        ge=1,
+        alias="INTERNAL_RATE_LIMIT_REQUESTS_PER_WINDOW",
+    )
+    rate_limit_window_seconds: int = Field(
+        default=60,
+        ge=1,
+        alias="RATE_LIMIT_WINDOW_SECONDS",
+    )
     llm_timeout_seconds: int = Field(default=12, alias="LLM_TIMEOUT_SECONDS")
     max_llm_concurrent_requests: int = Field(
         default=4,
@@ -194,6 +243,30 @@ class Settings(BaseSettings):
     @model_validator(mode="after")
     def validate_post_feed_security(self) -> "Settings":
         self.vector_store_provider = self.vector_store_provider.strip().lower()
+        allowed_resources = {"places", "posts", "users", "clubs", "groups", "events"}
+        for resource_type, thresholds in self.global_search_resource_thresholds.items():
+            if resource_type not in allowed_resources:
+                raise ValueError(
+                    f"GLOBAL_SEARCH_RESOURCE_THRESHOLDS_JSON contiene un recurso invalido: {resource_type}"
+                )
+            if set(thresholds) != {"semantic_min", "lexical_min"}:
+                raise ValueError(
+                    "cada override de GLOBAL_SEARCH_RESOURCE_THRESHOLDS_JSON debe "
+                    "contener semantic_min y lexical_min"
+                )
+            semantic_min = float(thresholds["semantic_min"])
+            lexical_min = float(thresholds["lexical_min"])
+            if not math.isfinite(semantic_min) or not -1.0 <= semantic_min <= 1.0:
+                raise ValueError("semantic_min debe ser finito y estar entre -1 y 1")
+            if not math.isfinite(lexical_min) or lexical_min < 0.0:
+                raise ValueError("lexical_min debe ser finito y no negativo")
+        if not math.isfinite(self.global_search_min_semantic_score):
+            raise ValueError("GLOBAL_SEARCH_MIN_SEMANTIC_SCORE debe ser finito")
+        if not math.isfinite(self.global_search_min_lexical_score):
+            raise ValueError("GLOBAL_SEARCH_MIN_LEXICAL_SCORE debe ser finito")
+        self.global_search_threshold_policy_version = (
+            self.global_search_threshold_policy_version.strip()
+        )
         if self.env.lower() != "local":
             missing: list[str] = []
             if not self.post_feed_internal_token:

app/shared/search/source.py

@@ -34,6 +34,8 @@ class PagedMainApiSearchClient:
         self._settings = settings
         self._base_url = settings.main_api_base_url.rstrip("/") + "/"
         self._path = path.lstrip("/")
+        if not self._path.startswith("api/v1/internal/search/"):
+            raise ValueError("search sync source must use an internal snapshot endpoint")
         self._collection_keys = collection_keys
         self._mapper = mapper
         self._page_limit = page_limit
@@ -88,8 +90,12 @@ class PagedMainApiSearchClient:
                 offset += limit
 
     def _build_headers(self) -> dict[str, str]:
-        token = self._settings.main_api_internal_token or self._settings.main_api_auth_token
-        return {"Authorization": f"Bearer {token}"} if token else {}
+        token = (self._settings.main_api_internal_token or "").strip()
+        if not token:
+            raise RuntimeError(
+                "MAIN_API_INTERNAL_TOKEN is required for search snapshot synchronization"
+            )
+        return {"Authorization": f"Bearer {token}"}
 
     def _pagination_params(
         self,

app/shared/security/rate_limit.py

@@ -1,6 +1,65 @@
-from fastapi import Request
+from collections import deque
+from threading import Lock
+from time import monotonic
+
+from fastapi import HTTPException, Request, status
+
+from app.shared.config.settings import get_settings
+
+
+_requests: dict[str, deque[float]] = {}
+_lock = Lock()
+_MAX_BUCKETS = 10_000
 
 
 async def rate_limit_placeholder(request: Request) -> None:
-    """Placeholder for future per-client or per-service rate limiting."""
-    return None
+    """Apply a small in-process fixed-window guard at the HTTP boundary.
+
+    The production gateway may impose an additional distributed limit. This guard is
+    intentionally local to each worker so a missing gateway rule cannot leave the
+    service completely unbounded.
+    """
+    settings = get_settings()
+    window_seconds = settings.rate_limit_window_seconds
+    limit = (
+        settings.internal_rate_limit_requests_per_window
+        if request.url.path.startswith("/internal/")
+        else settings.rate_limit_requests_per_window
+    )
+    client_host = request.client.host if request.client else "unknown"
+    key = f"{client_host}:{request.url.path}"
+    now = monotonic()
+    cutoff = now - window_seconds
+
+    with _lock:
+        if key not in _requests and len(_requests) >= _MAX_BUCKETS:
+            stale_keys = [
+                bucket_key
+                for bucket_key, bucket in _requests.items()
+                if not bucket or bucket[-1] <= cutoff
+            ]
+            for stale_key in stale_keys:
+                _requests.pop(stale_key, None)
+            if len(_requests) >= _MAX_BUCKETS:
+                raise HTTPException(
+                    status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+                    detail="Rate limit capacity exceeded",
+                    headers={"Retry-After": str(window_seconds)},
+                )
+        timestamps = _requests.setdefault(key, deque())
+        while timestamps and timestamps[0] <= cutoff:
+            timestamps.popleft()
+        if len(timestamps) >= limit:
+            retry_after = max(1, int(window_seconds - (now - timestamps[0])))
+            raise HTTPException(
+                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+                detail="Rate limit exceeded",
+                headers={"Retry-After": str(retry_after)},
+            )
+        timestamps.append(now)
+
+
+def reset_rate_limit_state() -> None:
+    """Clear process-local buckets for deterministic tests."""
+    with _lock:
+        _requests.clear()

app/shared/vector_store/aws_pgvector.py

@@ -21,6 +21,13 @@ READ_CONTRACT_SIGNATURES = {
     "get_post_feed_features": "get_post_feed_features(text, text[])",
 }
 
+SEARCH_CANDIDATE_V2_MARKERS = (
+    "event_active_at",
+    "min_semantic_score",
+    "min_lexical_score",
+    "try_parse_timestamptz",
+)
+
 SEARCH_RESOURCE_TYPES = frozenset({"places", "posts", "users", "clubs", "groups", "events"})
 
 
@@ -134,6 +141,20 @@ class AwsPgvectorClient:
                                 "executable": bool(row["executable"]) if row else False,
                             }
                         )
+                        if function_name == "search_resource_embeddings":
+                            definition_row = await connection.fetchrow(
+                                "SELECT pg_get_functiondef(to_regprocedure($1)) AS body",
+                                signature,
+                            )
+                            definition = (
+                                str(definition_row["body"] or "")
+                                if definition_row
+                                else ""
+                            )
+                            functions[function_name]["candidate_filters_v2"] = all(
+                                marker in definition
+                                for marker in SEARCH_CANDIDATE_V2_MARKERS
+                            )
         except Exception as exc:
             return {
                 "ready": False,
@@ -141,10 +162,7 @@ class AwsPgvectorClient:
                 "message": str(exc),
             }
 
-        ready = vector_available and all(
-            details["exists"] and details["executable"]
-            for details in functions.values()
-        )
+        ready = _read_contract_is_ready(vector_available, functions)
         return {
             "ready": ready,
             "vector_extension": vector_available,
@@ -387,9 +405,26 @@ def _build_connection_kwargs(settings: Settings, role: str) -> dict[str, Any]:
         "database": settings.pgvector_database,
         "user": user,
         "password": password,
+        "timeout": settings.request_timeout_seconds,
+        "command_timeout": settings.request_timeout_seconds,
     }
 
 
+def _read_contract_is_ready(
+    vector_available: bool,
+    functions: dict[str, dict[str, Any]],
+) -> bool:
+    return vector_available and all(
+        details.get("exists") is True
+        and details.get("executable") is True
+        and (
+            function_name != "search_resource_embeddings"
+            or details.get("candidate_filters_v2") is True
+        )
+        for function_name, details in functions.items()
+    )
+
+
 def _credentials_for_role(settings: Settings, role: str) -> tuple[str | None, str | None]:
     if role == "writer":
         return (

docs/api_endpoints.md

@@ -42,6 +42,7 @@ Todas las peticiones y respuestas con body utilizan `application/json`.
 | `GET` | `/internal/posts/clusters/runs/{id}` | Detalle de un run de clusters | Servicio |
 | `POST` | `/internal/posts/clusters/runs/{id}/activate` | Activar/rollback de run | Servicio |
 | `POST` | `/search` | Busqueda global sobre todos los recursos | Condicional |
+| `POST` | `/internal/search/candidates` | Candidatos tecnicos para la API principal | Servicio |
 
 La autenticacion de `/search` solo es obligatoria cuando el body contiene
 `requester_id`. En ese caso se envia:
@@ -96,6 +97,7 @@ Comprueba que el servicio puede utilizar su almacenamiento vectorial. Cuando
 - existencia y permiso de ejecucion de `match_places(...)`;
 - existencia y permiso de ejecucion de `match_posts(...)`;
 - existencia y permiso de ejecucion de `search_resource_embeddings(...)`.
+- presencia de los filtros V2 (`event_active_at`, umbrales y parseo temporal seguro).
 
 Respuesta correcta: `200 OK` con `status: "ready"`.
 
@@ -759,19 +761,72 @@ Ejemplo de error de validacion `422`:
 }
 ```
 
-## 6. Variables relacionadas
+## 6. Candidatos internos de busqueda
+
+### `POST /internal/search/candidates`
+
+Endpoint de servicio consumido exclusivamente por la API principal. La app movil no
+debe llamarlo directamente.
+
+```http
+Authorization: Bearer <NLP_SERVICE_TOKEN>
+Content-Type: application/json
+```
+
+Ejemplo minimo:
+
+```json
+{
+  "query": "club universitario de ajedrez",
+  "resource_types": ["clubs", "groups", "events", "posts"],
+  "candidate_limit_per_type": 50,
+  "top_limit": 100,
+  "as_of": "2026-07-12T14:00:00Z"
+}
+```
+
+La respuesta contiene secciones independientes con `id`, `resource_type`, `score`,
+`semantic_score` y `lexical_score`. No incluye metadata de UI ni registros hidratados.
+La API principal aplica permisos y reglas de negocio, carga los recursos vigentes y
+conserva el orden del ranking.
+
+NLP aplica antes del `LIMIT` los umbrales configurados. Un candidato se acepta cuando
+supera el umbral semantico **o** el lexical; si ninguno lo supera se descarta y no se
+rellena la respuesta. Para eventos tambien exige que
+`start_time + duration_minutes > as_of`, evitando calcular ranking sobre eventos ya
+finalizados. Los cursores quedan ligados a query, recursos, filtros, ubicacion,
+`as_of` y version de la politica de umbrales.
+
+## 7. Variables relacionadas
 
 | Variable | Uso |
 | --- | --- |
 | `SEARCH_INTERNAL_TOKEN` | Valida `Authorization: Bearer ...` cuando `/search` recibe `requester_id` |
-| `NLP_SERVICE_TOKEN` | Autentica todos los endpoints `/internal/posts/*`; debe coincidir con Go |
+| `NLP_SERVICE_TOKEN` | Autentica `/internal/posts/*` y `/internal/search/candidates`; debe coincidir con `NLP_SERVICE_TOKEN` en Go |
 | `MAIN_API_INTERNAL_TOKEN` | Autentica los jobs NLP al leer endpoints internos de Go |
 | `MAX_REQUEST_BODY_BYTES` | Debe ser al menos `131072`; valor recomendado `262144` para 500 candidatos |
+| `REQUEST_TIMEOUT_SECONDS` | Presupuesto maximo para search interno y timeout de conexion/consulta pgvector |
+| `RATE_LIMIT_REQUESTS_PER_WINDOW` | Limite local por IP/ruta para endpoints publicos |
+| `INTERNAL_RATE_LIMIT_REQUESTS_PER_WINDOW` | Limite local por IP/ruta para endpoints internos |
+| `RATE_LIMIT_WINDOW_SECONDS` | Ventana del limitador local; el gateway puede agregar un limite distribuido |
 | `VECTOR_STORE_PROVIDER` | Selecciona `aws_pgvector` o el proveedor mock |
 | `PGVECTOR_*` | Conexion y roles de PostgreSQL/pgvector |
 | `EMBEDDING_PROVIDER` | Proveedor de embeddings; produccion utiliza `fasttext` |
 | `EMBEDDING_DIMENSION` | Dimension vectorial; FastText utiliza `300` |
 | `GROQ_API_KEY` | Habilita Groq/Llama para redactar respuestas conversacionales |
 | `MAIN_API_BASE_URL` | API principal usada para fuentes y filtros geograficos |
+| `MAIN_API_USERS_SNAPSHOT_PATH` | Snapshot interno paginado de usuarios para search |
+| `MAIN_API_CLUBS_SNAPSHOT_PATH` | Snapshot interno paginado de clubs para search |
+| `MAIN_API_GROUPS_SNAPSHOT_PATH` | Snapshot interno paginado de grupos para search |
+| `MAIN_API_EVENTS_SNAPSHOT_PATH` | Snapshot interno paginado de eventos para search |
 | `MAIN_API_PLACES_NEARBY_PATH` | Endpoint que resuelve los IDs dentro del radio solicitado |
 | `GLOBAL_SEARCH_NEARBY_BOOST` | Peso de priorizacion geografica; default `0.12` |
+| `PUBLIC_GLOBAL_SEARCH_ENABLED` | Mantiene o deshabilita el endpoint publico heredado `POST /search` |
+| `GLOBAL_SEARCH_MIN_SEMANTIC_SCORE` | Umbral semantico global; default `0.30` |
+| `GLOBAL_SEARCH_MIN_LEXICAL_SCORE` | Umbral lexical global; default `0.05` |
+| `GLOBAL_SEARCH_RESOURCE_THRESHOLDS_JSON` | Overrides opcionales por recurso con `semantic_min` y `lexical_min` |
+| `GLOBAL_SEARCH_THRESHOLD_POLICY_VERSION` | Version estable incluida en cursores para invalidarlos al cambiar la politica |
+
+`top_limit` se conserva en el request y en la huella del cursor para mantener el
+contexto coordinado con Go. NLP devuelve candidatos por seccion; la composicion final de
+`top_results` pertenece a la API principal.

docs/global_search.md

@@ -1,54 +1,142 @@
 # Busqueda global
 
-## Responsabilidades
-
-- Los modulos `users`, `clubs`, `groups` y `events` construyen su documento semantico y
-  adaptan el contrato de la API principal.
-- `modules/search` genera un embedding por consulta, ejecuta proveedores en paralelo y
-  compone `top_results`, resultados paginados y cursores por seccion.
-- La API principal conserva los recursos originales y la identidad autenticada.
-- La API principal tambien conserva la verdad geografica. El servicio NLP obtiene un
-  allowlist de `place_ids` desde `/api/v1/places/nearby`; no duplica coordenadas.
-- PostgreSQL + pgvector contiene exclusivamente indices derivados reconstruibles.
-
-## Ubicacion y filtros
-
-- `location.mode=prioritize` agrega una senal de ranking separada para lugares cercanos.
-  El score semantico original no se modifica.
-- `location.mode=strict` limita places, clubs y events a los `place_ids` resueltos por la
-  API principal.
-- Los filtros son tipados y se aplican por recurso: ciudad/estado, categorias, precios,
-  tags, fechas de posts, fechas de eventos, modalidad del club y roles de usuarios.
-- Los filtros duros se ejecutan en PostgreSQL antes del ranking y se vuelven a validar en
-  la aplicacion como defensa del contrato.
-- Cada recurso entrega su propio cursor. El cursor queda vinculado a consulta, filtros,
-  ubicacion, identidad y tamano de pagina.
-
-## Pesos semanticos
-
-| Recurso | Campos |
-| --- | --- |
-| users | username x10, full_name x8, bio x2 |
-| clubs | name x8, category x5, description x3 |
-| groups | name x8, description x3 |
-| events | title x8, nombres de tags x5, description x3 |
-
-## Despliegue
-
-1. Ejecutar `sql/aws_pgvector_contract.sql` con el rol propietario.
-2. Conceder las funciones nuevas a `nlp_reader` y `nlp_writer` usando las sentencias
-   comentadas al final del archivo SQL.
-3. Configurar los cuatro paths `MAIN_API_*_SEARCH_PATH`.
-4. Probar una pagina sin escribir:
-   `python -m app.jobs.sync_search_embeddings --resource all --dry-run --max-pages 1`.
-5. Ejecutar la carga: `python -m app.jobs.sync_search_embeddings --resource all`.
-6. Verificar dimensiones, normas y funciones con
-   `psql <cadena-de-conexion> -f sql/verify_global_search_embeddings.sql`.
-7. Volver a ejecutar `python -m app.jobs.initial_load_post_embeddings` para incorporar
-   `published_at` a la metadata de posts ya indexados.
-8. Volver a ejecutar `python -m app.jobs.sync_search_embeddings --resource all` para
-   refrescar filtros de recursos; `user_roles` requiere que el snapshot interno de
-   usuarios incluya `role`.
-
-El contrato SQL agrega `textsearch` a lugares y posts sin regenerar sus vectores. Las
-tablas nuevas usan HNSW para FastText y GIN para texto/metadatos.
+## Fuente de verdad y flujo
+
+```text
+App movil + JWT
+  -> API principal Go: POST /api/v1/search
+  -> NLP: POST /internal/search/candidates + NLP_SERVICE_TOKEN
+  -> pgvector/FTS: IDs y scores sobre umbral
+  -> Go: permisos, reglas de negocio e hidratacion PostgreSQL
+  -> App movil: recursos completos
+```
+
+- La API principal conserva identidad, permisos y objetos originales.
+- NLP solo calcula embeddings, recuperacion, umbrales, ranking y cursores internos.
+- PostgreSQL pgvector contiene indices derivados reconstruibles.
+- La respuesta interna excluye metadata privada y campos para presentacion.
+- `POST /search` es compatibilidad temporal y puede apagarse con
+  `PUBLIC_GLOBAL_SEARCH_ENABLED=false`.
+
+## Relevancia sin relleno
+
+Cada recurso posee `semantic_min` y `lexical_min`. Un candidato se acepta cuando cumple
+al menos una de estas condiciones:
+
+```text
+semantic_score >= semantic_min
+OR lexical_score >= lexical_min
+```
+
+RRF ordena candidatos aceptados, pero no decide por si solo si existe coincidencia. El
+limite solicitado es un maximo: cero coincidencias devuelve `[]`; una coincidencia
+devuelve una. La diversificacion nunca recupera elementos bajo el umbral.
+
+Variables:
+
+```dotenv
+GLOBAL_SEARCH_MIN_SEMANTIC_SCORE=0.30
+GLOBAL_SEARCH_MIN_LEXICAL_SCORE=0.05
+GLOBAL_SEARCH_RESOURCE_THRESHOLDS_JSON={}
+GLOBAL_SEARCH_THRESHOLD_POLICY_VERSION=global-search-relevance-v1
+```
+
+Ejemplo de override completo:
+
+```dotenv
+GLOBAL_SEARCH_RESOURCE_THRESHOLDS_JSON={"posts":{"semantic_min":0.35,"lexical_min":0.05},"events":{"semantic_min":0.38,"lexical_min":0.06}}
+```
+
+Los valores iniciales deben recalibrarse con consultas positivas y negativas reales. El
+umbral lexical corresponde a `ts_rank_cd`; no se debe copiar
+`BM25_RELEVANCE_THRESHOLD`.
+
+El repositorio incluye un calibrador reproducible que solo consume etiquetas y scores,
+sin texto ni metadata privada. Cada linea JSONL debe usar este contrato:
+
+```json
+{"resource_type":"events","relevant":true,"semantic_score":0.62,"lexical_score":0.08}
+```
+
+Ejecucion:
+
+```powershell
+python scripts/calibrate_global_search_thresholds.py .\global_search_labels.jsonl `
+  --policy-version global-search-relevance-v2 `
+  --target-precision 0.95 `
+  --output .\global_search_calibration.json
+```
+
+El resultado contiene `resource_thresholds` listo para copiar a
+`GLOBAL_SEARCH_RESOURCE_THRESHOLDS_JSON` y metricas por recurso. La activacion sigue
+bloqueada hasta alimentar el script con resultados reales de los seis recursos.
+
+## Eventos
+
+El endpoint interno recibe `as_of` desde Go y pgvector aplica en el CTE `eligible`, antes
+del ranking:
+
+```text
+start_time + duration_minutes > as_of
+```
+
+Esto conserva eventos futuros y en curso, y elimina eventos terminados o con metadata
+temporal invalida. Go repite la validacion con la tabla principal; NLP solo optimiza.
+
+## Ubicacion, filtros y cursores
+
+- `prioritize` agrega una senal separada de cercania.
+- `strict` limita places, clubs y events a los `place_ids` resueltos por la API principal.
+- Los filtros duros se ejecutan en SQL y se validan nuevamente en Python.
+- Cada recurso tiene cursor independiente ligado a query, filtros, ubicacion, `as_of`,
+  limite y version de la politica.
+- Go encapsula el cursor NLP dentro de su cursor HMAC; la app nunca recibe el cursor
+  interno sin esa proteccion.
+
+## Migracion de una BD existente
+
+Ejecutar con el propietario de `search_resource_embeddings(...)`:
+
+1. `sql/migrations/20260712_01_global_search_candidate_filters.sql`.
+2. `sql/verify_global_search_candidate_filters.sql`.
+
+El primer archivo es el unico incremental de esta funcionalidad. No crea tablas,
+columnas ni indices; usa `CREATE OR REPLACE FUNCTION`, conserva la firma, agrega helpers
+de parseo seguro y vuelve a otorgar `EXECUTE` a `nlp_reader`. Sus postchecks funcionales
+ocurren antes del `COMMIT`. El verificador posterior abre una transaccion de solo lectura
+y aborta si encuentra contrato viejo, consultas bajo umbral con resultados, eventos
+terminados o eventos activos con metadata temporal invalida.
+
+No ejecutar `sql/new_pgvector_schema.sql` en produccion: es el esquema convergente para
+inicializar desde cero. `sql/aws_pgvector_contract.sql` contiene el contrato completo de
+referencia.
+
+## Sincronizacion
+
+Las fuentes para users, clubs, groups y events son endpoints internos paginados de Go,
+autenticados con `MAIN_API_INTERNAL_TOKEN`:
+
+```dotenv
+MAIN_API_USERS_SNAPSHOT_PATH=/api/v1/internal/search/users/snapshot
+MAIN_API_CLUBS_SNAPSHOT_PATH=/api/v1/internal/search/clubs/snapshot
+MAIN_API_GROUPS_SNAPSHOT_PATH=/api/v1/internal/search/groups/snapshot
+MAIN_API_EVENTS_SNAPSHOT_PATH=/api/v1/internal/search/events/snapshot
+```
+
+Los jobs de search no aceptan `MAIN_API_AUTH_TOKEN` como reemplazo. La identidad de un
+usuario final no debe controlar la construccion del indice derivado.
+
+Antes de activar la busqueda:
+
+1. Ejecutar dry-run de `python -m app.jobs.sync_search_embeddings --resource all`.
+2. Sincronizar todos los recursos.
+3. Confirmar que events incluye `start_time` y `duration_minutes` validos.
+4. Ejecutar `sql/verify_global_search_embeddings.sql`.
+5. Ejecutar el verificador V2 indicado arriba.
+6. Desplegar NLP y probar `/health`, `/ready` e `/internal/search/candidates`.
+7. Activar `GLOBAL_SEARCH_ENABLED=true` en Go.
+
+`/ready` devuelve `503` si la firma existe pero la definicion instalada no contiene el
+contrato V2. Search interno tambien posee timeout total y las consultas pgvector usan el
+mismo presupuesto. El rate limit local es una ultima defensa por proceso; en despliegues
+con multiples replicas debe complementarse con el gateway.

scripts/calibrate_global_search_thresholds.py

@@ -0,0 +1,215 @@
+"""Calibrate per-resource global-search thresholds from labeled score samples.
+
+Input is JSONL with: resource_type, relevant, semantic_score and lexical_score.
+At least one score must be present in every row. The script never reads source text or
+private metadata.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Iterable
+
+
+RESOURCE_TYPES = ("places", "posts", "users", "clubs", "groups", "events")
+
+
+@dataclass(frozen=True)
+class LabeledScore:
+    resource_type: str
+    relevant: bool
+    semantic_score: float | None
+    lexical_score: float | None
+
+
+def calibrate_resource(
+    samples: list[LabeledScore],
+    target_precision: float = 0.95,
+    default_semantic: float = 0.30,
+    default_lexical: float = 0.05,
+) -> dict[str, float | int]:
+    positives = sum(sample.relevant for sample in samples)
+    negatives = len(samples) - positives
+    if positives == 0 or negatives == 0:
+        raise ValueError("each resource requires positive and negative labeled samples")
+
+    semantic_candidates = _candidate_values(
+        (sample.semantic_score for sample in samples),
+        default=default_semantic,
+        minimum=-1.0,
+        maximum=1.0,
+    )
+    lexical_candidates = _candidate_values(
+        (sample.lexical_score for sample in samples),
+        default=default_lexical,
+        minimum=0.0,
+        maximum=None,
+    )
+
+    evaluated: list[dict[str, float | int]] = []
+    for semantic_min in semantic_candidates:
+        for lexical_min in lexical_candidates:
+            tp = fp = fn = 0
+            for sample in samples:
+                accepted = (
+                    sample.semantic_score is not None
+                    and sample.semantic_score >= semantic_min
+                ) or (
+                    sample.lexical_score is not None
+                    and sample.lexical_score >= lexical_min
+                )
+                if accepted and sample.relevant:
+                    tp += 1
+                elif accepted:
+                    fp += 1
+                elif sample.relevant:
+                    fn += 1
+            precision = tp / (tp + fp) if tp + fp else 1.0
+            recall = tp / (tp + fn) if tp + fn else 0.0
+            f1 = (
+                2.0 * precision * recall / (precision + recall)
+                if precision + recall
+                else 0.0
+            )
+            evaluated.append(
+                {
+                    "semantic_min": semantic_min,
+                    "lexical_min": lexical_min,
+                    "precision": precision,
+                    "recall": recall,
+                    "f1": f1,
+                    "true_positives": tp,
+                    "false_positives": fp,
+                    "false_negatives": fn,
+                }
+            )
+
+    precision_qualified = [
+        result for result in evaluated if result["precision"] >= target_precision
+    ]
+    pool = precision_qualified or evaluated
+    return max(
+        pool,
+        key=lambda result: (
+            result["recall"] if precision_qualified else result["f1"],
+            result["precision"],
+            result["f1"],
+            -result["semantic_min"],
+            -result["lexical_min"],
+        ),
+    )
+
+
+def load_samples(path: Path) -> list[LabeledScore]:
+    samples: list[LabeledScore] = []
+    for line_number, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
+        if not raw_line.strip():
+            continue
+        try:
+            payload = json.loads(raw_line)
+            resource_type = str(payload["resource_type"])
+            relevant = payload["relevant"]
+            if resource_type not in RESOURCE_TYPES or not isinstance(relevant, bool):
+                raise ValueError("invalid resource_type or relevant label")
+            semantic = _optional_finite_float(payload.get("semantic_score"))
+            lexical = _optional_finite_float(payload.get("lexical_score"))
+            if semantic is None and lexical is None:
+                raise ValueError("at least one score is required")
+            if semantic is not None and not -1.0 <= semantic <= 1.0:
+                raise ValueError("semantic_score must be between -1 and 1")
+            if lexical is not None and lexical < 0.0:
+                raise ValueError("lexical_score must be non-negative")
+        except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
+            raise ValueError(f"invalid calibration row {line_number}: {exc}") from exc
+        samples.append(LabeledScore(resource_type, relevant, semantic, lexical))
+    return samples
+
+
+def _optional_finite_float(value: object) -> float | None:
+    if value is None:
+        return None
+    parsed = float(value)
+    if not math.isfinite(parsed):
+        raise ValueError("scores must be finite")
+    return parsed
+
+
+def _candidate_values(
+    values: Iterable[float | None],
+    *,
+    default: float,
+    minimum: float,
+    maximum: float | None,
+) -> list[float]:
+    observed = sorted({value for value in values if value is not None})
+    if len(observed) > 200:
+        observed = [
+            observed[round(index * (len(observed) - 1) / 199)]
+            for index in range(200)
+        ]
+    candidates = {default, minimum, *observed}
+    if observed:
+        above_max = math.nextafter(observed[-1], math.inf)
+        if maximum is None or above_max <= maximum:
+            candidates.add(above_max)
+    if maximum is not None:
+        candidates.add(maximum)
+    return sorted(
+        value
+        for value in candidates
+        if value >= minimum and (maximum is None or value <= maximum)
+    )
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("input", type=Path)
+    parser.add_argument("--policy-version", required=True)
+    parser.add_argument("--target-precision", type=float, default=0.95)
+    parser.add_argument("--allow-partial", action="store_true")
+    parser.add_argument("--output", type=Path)
+    args = parser.parse_args()
+    if not 0.0 < args.target_precision <= 1.0:
+        raise SystemExit("--target-precision must be in (0, 1]")
+
+    samples = load_samples(args.input)
+    grouped = {
+        resource_type: [
+            sample for sample in samples if sample.resource_type == resource_type
+        ]
+        for resource_type in RESOURCE_TYPES
+    }
+    missing = [resource for resource, items in grouped.items() if not items]
+    if missing and not args.allow_partial:
+        raise SystemExit(f"missing labeled resources: {', '.join(missing)}")
+
+    results = {
+        resource: calibrate_resource(items, target_precision=args.target_precision)
+        for resource, items in grouped.items()
+        if items
+    }
+    output = {
+        "policy_version": args.policy_version,
+        "target_precision": args.target_precision,
+        "resource_thresholds": {
+            resource: {
+                "semantic_min": result["semantic_min"],
+                "lexical_min": result["lexical_min"],
+            }
+            for resource, result in results.items()
+        },
+        "metrics": results,
+    }
+    rendered = json.dumps(output, ensure_ascii=False, indent=2, sort_keys=True)
+    if args.output:
+        args.output.write_text(rendered + "\n", encoding="utf-8")
+    else:
+        print(rendered)
+
+
+if __name__ == "__main__":
+    main()

sql/aws_pgvector_contract.sql

@@ -393,6 +393,37 @@ AS $$
     WHERE p.external_id = ANY(p_external_ids);
 $$;
 
+CREATE OR REPLACE FUNCTION try_parse_timestamptz(p_value TEXT)
+RETURNS TIMESTAMPTZ
+LANGUAGE plpgsql
+IMMUTABLE
+PARALLEL SAFE
+SET search_path = pg_catalog
+AS $$
+BEGIN
+    RETURN NULLIF(btrim(p_value), '')::timestamptz;
+EXCEPTION WHEN OTHERS THEN
+    RETURN NULL;
+END;
+$$;
+
+CREATE OR REPLACE FUNCTION try_parse_positive_integer(p_value TEXT)
+RETURNS INTEGER
+LANGUAGE plpgsql
+IMMUTABLE
+PARALLEL SAFE
+SET search_path = pg_catalog
+AS $$
+DECLARE
+    parsed INTEGER;
+BEGIN
+    parsed := NULLIF(btrim(p_value), '')::integer;
+    RETURN CASE WHEN parsed > 0 THEN parsed ELSE NULL END;
+EXCEPTION WHEN OTHERS THEN
+    RETURN NULL;
+END;
+$$;
+
 CREATE OR REPLACE FUNCTION search_resource_embeddings(
     p_resource_type TEXT,
     p_query_text TEXT,
@@ -462,24 +493,40 @@ BEGIN
               )
               AND (
                     ($4 ? 'published_from') IS FALSE
-                    OR NULLIF(e.metadata->>'published_at', '')::timestamptz
+                    OR public.try_parse_timestamptz(e.metadata->>'published_at')
                        >= ($4->>'published_from')::timestamptz
               )
               AND (
                     ($4 ? 'published_to') IS FALSE
-                    OR NULLIF(e.metadata->>'published_at', '')::timestamptz
+                    OR public.try_parse_timestamptz(e.metadata->>'published_at')
                        <= ($4->>'published_to')::timestamptz
               )
               AND (
                     ($4 ? 'event_from') IS FALSE
-                    OR NULLIF(e.metadata->>'start_time', '')::timestamptz
+                    OR public.try_parse_timestamptz(e.metadata->>'start_time')
                        >= ($4->>'event_from')::timestamptz
               )
               AND (
                     ($4 ? 'event_to') IS FALSE
-                    OR NULLIF(e.metadata->>'start_time', '')::timestamptz
+                    OR public.try_parse_timestamptz(e.metadata->>'start_time')
                        <= ($4->>'event_to')::timestamptz
               )
+              AND (
+                    ($4 ? 'event_active_at') IS FALSE
+                    OR (
+                        public.try_parse_timestamptz(e.metadata->>'start_time') IS NOT NULL
+                        AND public.try_parse_positive_integer(
+                            e.metadata->>'duration_minutes'
+                        ) IS NOT NULL
+                        AND public.try_parse_timestamptz(e.metadata->>'start_time')
+                            + make_interval(
+                                mins => public.try_parse_positive_integer(
+                                    e.metadata->>'duration_minutes'
+                                )
+                              )
+                            > ($4->>'event_active_at')::timestamptz
+                    )
+              )
               AND (
                     ($4 ? 'is_online') IS FALSE
                     OR COALESCE((e.metadata->>'is_online')::boolean, false)
@@ -570,6 +617,26 @@ BEGIN
             f.lexical_score::double precision
         FROM fused f
         JOIN eligible e USING (external_id)
+        WHERE (
+            (
+                ($4 ? 'min_semantic_score') IS FALSE
+                AND ($4 ? 'min_lexical_score') IS FALSE
+            )
+            OR (
+                ($4 ? 'min_semantic_score')
+                AND COALESCE(
+                    f.semantic_score >= ($4->>'min_semantic_score')::double precision,
+                    FALSE
+                )
+            )
+            OR (
+                ($4 ? 'min_lexical_score')
+                AND COALESCE(
+                    f.lexical_score >= ($4->>'min_lexical_score')::double precision,
+                    FALSE
+                )
+            )
+        )
         ORDER BY score DESC, f.semantic_score DESC NULLS LAST, e.external_id ASC
         LIMIT GREATEST($3, 1)
     $query$, target_table)
@@ -577,6 +644,11 @@ BEGIN
 END;
 $$;
 
+REVOKE ALL ON FUNCTION public.try_parse_timestamptz(text) FROM PUBLIC;
+REVOKE ALL ON FUNCTION public.try_parse_positive_integer(text) FROM PUBLIC;
+GRANT EXECUTE ON FUNCTION public.try_parse_timestamptz(text) TO nlp_reader;
+GRANT EXECUTE ON FUNCTION public.try_parse_positive_integer(text) TO nlp_reader;
+
 CREATE OR REPLACE FUNCTION get_resource_content_hashes(
     p_resource_type TEXT,
     p_external_ids TEXT[]

sql/migrations/20260712_01_global_search_candidate_filters.sql

@@ -0,0 +1,413 @@
+-- Busqueda global V2: vigencia de eventos y umbrales de relevancia.
+--
+-- Ejecutar UNA sola vez en la BD NLP desplegada con el propietario de la
+-- funcion public.search_resource_embeddings. No crea tablas, columnas ni
+-- indices; conserva la firma usada por la API y agrega dos helpers de parseo
+-- seguro para que metadata temporal corrupta se descarte sin abortar la busqueda.
+
+BEGIN;
+
+SET LOCAL lock_timeout = '10s';
+SET LOCAL statement_timeout = '10min';
+SET LOCAL idle_in_transaction_session_timeout = '5min';
+SET LOCAL search_path = public;
+
+SELECT pg_advisory_xact_lock(
+    hashtextextended('public:20260712_01_global_search_candidate_filters', 0)
+);
+
+DO $preflight$
+DECLARE
+    required_table TEXT;
+    current_definition TEXT;
+BEGIN
+    IF NOT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'vector') THEN
+        RAISE EXCEPTION 'Preflight search V2: falta la extension vector';
+    END IF;
+
+    IF to_regprocedure(
+        'public.search_resource_embeddings(text,text,vector,integer,jsonb)'
+    ) IS NULL THEN
+        RAISE EXCEPTION 'Preflight search V2: falta search_resource_embeddings(text,text,vector,integer,jsonb)';
+    END IF;
+
+    FOREACH required_table IN ARRAY ARRAY[
+        'place_embeddings',
+        'post_embeddings',
+        'user_embeddings',
+        'club_embeddings',
+        'group_embeddings',
+        'event_embeddings'
+    ] LOOP
+        IF to_regclass('public.' || required_table) IS NULL THEN
+            RAISE EXCEPTION 'Preflight search V2: falta public.%', required_table;
+        END IF;
+    END LOOP;
+
+    IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'nlp_reader') THEN
+        RAISE EXCEPTION 'Preflight search V2: falta el rol nlp_reader';
+    END IF;
+
+    SELECT pg_get_functiondef(
+        'public.search_resource_embeddings(text,text,vector,integer,jsonb)'::regprocedure
+    ) INTO current_definition;
+
+    IF current_definition LIKE '%event_active_at%'
+       OR current_definition LIKE '%min_semantic_score%'
+       OR current_definition LIKE '%min_lexical_score%' THEN
+        RAISE EXCEPTION USING
+            MESSAGE = 'Preflight search V2: la funcion ya contiene cambios V2 o una instalacion parcial',
+            HINT = 'No vuelva a ejecutar la migracion. Ejecute sql/verify_global_search_candidate_filters.sql.';
+    END IF;
+END;
+$preflight$;
+
+CREATE OR REPLACE FUNCTION public.try_parse_timestamptz(p_value TEXT)
+RETURNS TIMESTAMPTZ
+LANGUAGE plpgsql
+IMMUTABLE
+PARALLEL SAFE
+SET search_path = pg_catalog
+AS $$
+BEGIN
+    RETURN NULLIF(btrim(p_value), '')::timestamptz;
+EXCEPTION WHEN OTHERS THEN
+    RETURN NULL;
+END;
+$$;
+
+CREATE OR REPLACE FUNCTION public.try_parse_positive_integer(p_value TEXT)
+RETURNS INTEGER
+LANGUAGE plpgsql
+IMMUTABLE
+PARALLEL SAFE
+SET search_path = pg_catalog
+AS $$
+DECLARE
+    parsed INTEGER;
+BEGIN
+    parsed := NULLIF(btrim(p_value), '')::integer;
+    RETURN CASE WHEN parsed > 0 THEN parsed ELSE NULL END;
+EXCEPTION WHEN OTHERS THEN
+    RETURN NULL;
+END;
+$$;
+
+CREATE OR REPLACE FUNCTION public.search_resource_embeddings(
+    p_resource_type TEXT,
+    p_query_text TEXT,
+    p_query_embedding VECTOR(300),
+    p_match_count INTEGER,
+    p_filters JSONB DEFAULT '{}'::jsonb
+)
+RETURNS TABLE (
+    external_id TEXT,
+    document TEXT,
+    metadata JSONB,
+    score DOUBLE PRECISION,
+    semantic_score DOUBLE PRECISION,
+    lexical_score DOUBLE PRECISION
+)
+LANGUAGE plpgsql
+STABLE
+SECURITY DEFINER
+SET search_path = public
+AS $$
+DECLARE
+    target_table TEXT;
+BEGIN
+    target_table := CASE p_resource_type
+        WHEN 'places' THEN 'place_embeddings'
+        WHEN 'posts' THEN 'post_embeddings'
+        WHEN 'users' THEN 'user_embeddings'
+        WHEN 'clubs' THEN 'club_embeddings'
+        WHEN 'groups' THEN 'group_embeddings'
+        WHEN 'events' THEN 'event_embeddings'
+        ELSE NULL
+    END;
+    IF target_table IS NULL THEN
+        RAISE EXCEPTION 'Unsupported search resource type: %', p_resource_type;
+    END IF;
+
+    RETURN QUERY EXECUTE format($query$
+        WITH eligible AS NOT MATERIALIZED (
+            SELECT e.*
+            FROM public.%I e
+            WHERE e.is_active = true
+              AND COALESCE(($4->>'is_active')::boolean, true) = true
+              AND (
+                    ($4 ? 'city') IS FALSE
+                    OR lower(e.metadata->>'city') = lower($4->>'city')
+              )
+              AND (
+                    ($4 ? 'state') IS FALSE
+                    OR lower(e.metadata->>'state') = lower($4->>'state')
+              )
+              AND (
+                    ($4 ? 'categories') IS FALSE
+                    OR EXISTS (
+                        SELECT 1
+                        FROM jsonb_array_elements_text($4->'categories') AS allowed(value)
+                        WHERE lower(e.metadata->>'category') = lower(allowed.value)
+                    )
+              )
+              AND (
+                    ($4 ? 'tags') IS FALSE
+                    OR EXISTS (
+                        SELECT 1
+                        FROM jsonb_array_elements_text($4->'tags') AS allowed(value)
+                        WHERE lower(COALESCE(e.metadata->>'tags', ''))
+                              LIKE ('%%' || lower(allowed.value) || '%%')
+                    )
+              )
+              AND (
+                    ($4 ? 'published_from') IS FALSE
+                    OR public.try_parse_timestamptz(e.metadata->>'published_at')
+                       >= ($4->>'published_from')::timestamptz
+              )
+              AND (
+                    ($4 ? 'published_to') IS FALSE
+                    OR public.try_parse_timestamptz(e.metadata->>'published_at')
+                       <= ($4->>'published_to')::timestamptz
+              )
+              AND (
+                    ($4 ? 'event_from') IS FALSE
+                    OR public.try_parse_timestamptz(e.metadata->>'start_time')
+                       >= ($4->>'event_from')::timestamptz
+              )
+              AND (
+                    ($4 ? 'event_to') IS FALSE
+                    OR public.try_parse_timestamptz(e.metadata->>'start_time')
+                       <= ($4->>'event_to')::timestamptz
+              )
+              AND (
+                    ($4 ? 'event_active_at') IS FALSE
+                    OR (
+                        public.try_parse_timestamptz(e.metadata->>'start_time') IS NOT NULL
+                        AND public.try_parse_positive_integer(
+                            e.metadata->>'duration_minutes'
+                        ) IS NOT NULL
+                        AND public.try_parse_timestamptz(e.metadata->>'start_time')
+                            + make_interval(
+                                mins => public.try_parse_positive_integer(
+                                    e.metadata->>'duration_minutes'
+                                )
+                              )
+                            > ($4->>'event_active_at')::timestamptz
+                    )
+              )
+              AND (
+                    ($4 ? 'is_online') IS FALSE
+                    OR COALESCE((e.metadata->>'is_online')::boolean, false)
+                       = ($4->>'is_online')::boolean
+              )
+              AND (
+                    ($4 ? 'user_roles') IS FALSE
+                    OR EXISTS (
+                        SELECT 1
+                        FROM jsonb_array_elements_text($4->'user_roles') AS allowed(value)
+                        WHERE lower(e.metadata->>'role') = lower(allowed.value)
+                    )
+              )
+              AND (
+                    ($4 ? 'place_ids') IS FALSE
+                    OR e.metadata->>'place_id' IN (
+                        SELECT jsonb_array_elements_text($4->'place_ids')
+                    )
+              )
+              AND (
+                    (
+                        COALESCE((e.metadata->>'is_private')::boolean, false) = false
+                        AND COALESCE((e.metadata->>'is_public')::boolean, true) = true
+                    )
+                    OR (
+                        $4 ? 'requester_id'
+                        AND (
+                            e.metadata->>'creator_id' = $4->>'requester_id'
+                            OR COALESCE(e.metadata->'authorized_user_ids', '[]'::jsonb)
+                               ? ($4->>'requester_id')
+                        )
+                    )
+              )
+        ),
+        vector_results AS (
+            SELECT
+                e.external_id,
+                1 - (e.embedding <=> $1) AS semantic_score,
+                row_number() OVER (
+                    ORDER BY e.embedding <=> $1, e.external_id ASC
+                ) AS vector_rank
+            FROM eligible e
+            ORDER BY e.embedding <=> $1, e.external_id ASC
+            LIMIT GREATEST($3 * 3, 30)
+        ),
+        lexical_results AS (
+            SELECT
+                e.external_id,
+                ts_rank_cd(e.textsearch, query.value) AS lexical_score,
+                row_number() OVER (
+                    ORDER BY ts_rank_cd(e.textsearch, query.value) DESC, e.external_id ASC
+                ) AS lexical_rank
+            FROM eligible e
+            CROSS JOIN websearch_to_tsquery('simple', $2) AS query(value)
+            WHERE e.textsearch @@ query.value
+            ORDER BY lexical_score DESC, e.external_id ASC
+            LIMIT GREATEST($3 * 3, 30)
+        ),
+        fused AS (
+            SELECT
+                COALESCE(v.external_id, l.external_id) AS external_id,
+                v.semantic_score,
+                l.lexical_score,
+                COALESCE(1.0 / (60.0 + v.vector_rank), 0.0)
+                    + COALESCE(1.0 / (60.0 + l.lexical_rank), 0.0) AS rrf_score
+            FROM vector_results v
+            FULL OUTER JOIN lexical_results l USING (external_id)
+        )
+        SELECT
+            e.external_id,
+            e.document,
+            e.metadata,
+            LEAST(
+                1.0,
+                f.rrf_score / (2.0 / 61.0)
+                + CASE
+                    WHEN lower(COALESCE(
+                        e.metadata->>'search_title',
+                        e.metadata->>'name',
+                        e.metadata->>'title',
+                        e.metadata->>'username',
+                        ''
+                    )) = lower($2) THEN 0.15
+                    ELSE 0.0
+                  END
+            )::double precision AS score,
+            f.semantic_score::double precision,
+            f.lexical_score::double precision
+        FROM fused f
+        JOIN eligible e USING (external_id)
+        WHERE (
+            (
+                ($4 ? 'min_semantic_score') IS FALSE
+                AND ($4 ? 'min_lexical_score') IS FALSE
+            )
+            OR (
+                ($4 ? 'min_semantic_score')
+                AND COALESCE(
+                    f.semantic_score >= ($4->>'min_semantic_score')::double precision,
+                    FALSE
+                )
+            )
+            OR (
+                ($4 ? 'min_lexical_score')
+                AND COALESCE(
+                    f.lexical_score >= ($4->>'min_lexical_score')::double precision,
+                    FALSE
+                )
+            )
+        )
+        ORDER BY score DESC, f.semantic_score DESC NULLS LAST, e.external_id ASC
+        LIMIT GREATEST($3, 1)
+    $query$, target_table)
+    USING p_query_embedding, p_query_text, p_match_count, p_filters;
+END;
+$$;
+
+REVOKE ALL ON FUNCTION public.search_resource_embeddings(
+    text, text, vector, integer, jsonb
+) FROM PUBLIC;
+REVOKE ALL ON FUNCTION public.try_parse_timestamptz(text) FROM PUBLIC;
+REVOKE ALL ON FUNCTION public.try_parse_positive_integer(text) FROM PUBLIC;
+GRANT EXECUTE ON FUNCTION public.search_resource_embeddings(
+    text, text, vector, integer, jsonb
+) TO nlp_reader;
+GRANT EXECUTE ON FUNCTION public.try_parse_timestamptz(text) TO nlp_reader;
+GRANT EXECUTE ON FUNCTION public.try_parse_positive_integer(text) TO nlp_reader;
+
+DO $postcheck$
+DECLARE
+    installed_definition TEXT;
+BEGIN
+    SELECT pg_get_functiondef(
+        'public.search_resource_embeddings(text,text,vector,integer,jsonb)'::regprocedure
+    ) INTO installed_definition;
+
+    IF installed_definition NOT LIKE '%event_active_at%'
+       OR installed_definition NOT LIKE '%min_semantic_score%'
+       OR installed_definition NOT LIKE '%min_lexical_score%'
+       OR installed_definition NOT LIKE '%try_parse_timestamptz%'
+       OR installed_definition NOT LIKE '%try_parse_positive_integer%' THEN
+        RAISE EXCEPTION 'Postcheck search V2: la definicion instalada esta incompleta';
+    END IF;
+
+    IF NOT has_function_privilege(
+        'nlp_reader',
+        'public.search_resource_embeddings(text,text,vector,integer,jsonb)',
+        'EXECUTE'
+    ) THEN
+        RAISE EXCEPTION 'Postcheck search V2: nlp_reader no conserva EXECUTE';
+    END IF;
+
+    IF NOT has_function_privilege(
+        'nlp_reader', 'public.try_parse_timestamptz(text)', 'EXECUTE'
+    ) OR NOT has_function_privilege(
+        'nlp_reader', 'public.try_parse_positive_integer(text)', 'EXECUTE'
+    ) THEN
+        RAISE EXCEPTION 'Postcheck search V2: nlp_reader no puede ejecutar los helpers seguros';
+    END IF;
+END;
+$postcheck$;
+
+DO $functional_postcheck$
+DECLARE
+    zero_embedding VECTOR(300);
+    invalid_count BIGINT;
+BEGIN
+    zero_embedding := (
+        '[' || rtrim(repeat('0,', 300), ',') || ']'
+    )::vector;
+
+    SELECT COUNT(*) INTO invalid_count
+    FROM public.search_resource_embeddings(
+        'users',
+        '__frimeet_verification_no_match_9f7c2__',
+        zero_embedding,
+        100,
+        jsonb_build_object(
+            'is_active', TRUE,
+            'min_semantic_score', 1.000001,
+            'min_lexical_score', 1000000.0
+        )
+    );
+    IF invalid_count <> 0 THEN
+        RAISE EXCEPTION 'Postcheck search V2: una consulta bajo umbrales imposibles devolvio % filas', invalid_count;
+    END IF;
+
+    SELECT COUNT(*) INTO invalid_count
+    FROM public.search_resource_embeddings(
+        'events',
+        '__frimeet_event_filter_verification__',
+        zero_embedding,
+        1000,
+        jsonb_build_object(
+            'is_active', TRUE,
+            'event_active_at', CURRENT_TIMESTAMP
+        )
+    ) result
+    WHERE public.try_parse_timestamptz(result.metadata->>'start_time') IS NULL
+       OR public.try_parse_positive_integer(
+            result.metadata->>'duration_minutes'
+          ) IS NULL
+       OR public.try_parse_timestamptz(result.metadata->>'start_time')
+          + make_interval(
+                mins => public.try_parse_positive_integer(
+                    result.metadata->>'duration_minutes'
+                )
+            ) <= CURRENT_TIMESTAMP;
+    IF invalid_count <> 0 THEN
+        RAISE EXCEPTION 'Postcheck search V2: se devolvieron % eventos terminados o invalidos', invalid_count;
+    END IF;
+END;
+$functional_postcheck$;
+
+COMMIT;

sql/new_pgvector_schema.sql

@@ -424,6 +424,37 @@ AS $$
     WHERE p.external_id = ANY(p_external_ids);
 $$;
 
+CREATE OR REPLACE FUNCTION try_parse_timestamptz(p_value TEXT)
+RETURNS TIMESTAMPTZ
+LANGUAGE plpgsql
+IMMUTABLE
+PARALLEL SAFE
+SET search_path = pg_catalog
+AS $$
+BEGIN
+    RETURN NULLIF(btrim(p_value), '')::timestamptz;
+EXCEPTION WHEN OTHERS THEN
+    RETURN NULL;
+END;
+$$;
+
+CREATE OR REPLACE FUNCTION try_parse_positive_integer(p_value TEXT)
+RETURNS INTEGER
+LANGUAGE plpgsql
+IMMUTABLE
+PARALLEL SAFE
+SET search_path = pg_catalog
+AS $$
+DECLARE
+    parsed INTEGER;
+BEGIN
+    parsed := NULLIF(btrim(p_value), '')::integer;
+    RETURN CASE WHEN parsed > 0 THEN parsed ELSE NULL END;
+EXCEPTION WHEN OTHERS THEN
+    RETURN NULL;
+END;
+$$;
+
 CREATE OR REPLACE FUNCTION search_resource_embeddings(
     p_resource_type TEXT,
     p_query_text TEXT,
@@ -493,24 +524,40 @@ BEGIN
               )
               AND (
                     ($4 ? 'published_from') IS FALSE
-                    OR NULLIF(e.metadata->>'published_at', '')::timestamptz
+                    OR public.try_parse_timestamptz(e.metadata->>'published_at')
                        >= ($4->>'published_from')::timestamptz
               )
               AND (
                     ($4 ? 'published_to') IS FALSE
-                    OR NULLIF(e.metadata->>'published_at', '')::timestamptz
+                    OR public.try_parse_timestamptz(e.metadata->>'published_at')
                        <= ($4->>'published_to')::timestamptz
               )
               AND (
                     ($4 ? 'event_from') IS FALSE
-                    OR NULLIF(e.metadata->>'start_time', '')::timestamptz
+                    OR public.try_parse_timestamptz(e.metadata->>'start_time')
                        >= ($4->>'event_from')::timestamptz
               )
               AND (
                     ($4 ? 'event_to') IS FALSE
-                    OR NULLIF(e.metadata->>'start_time', '')::timestamptz
+                    OR public.try_parse_timestamptz(e.metadata->>'start_time')
                        <= ($4->>'event_to')::timestamptz
               )
+              AND (
+                    ($4 ? 'event_active_at') IS FALSE
+                    OR (
+                        public.try_parse_timestamptz(e.metadata->>'start_time') IS NOT NULL
+                        AND public.try_parse_positive_integer(
+                            e.metadata->>'duration_minutes'
+                        ) IS NOT NULL
+                        AND public.try_parse_timestamptz(e.metadata->>'start_time')
+                            + make_interval(
+                                mins => public.try_parse_positive_integer(
+                                    e.metadata->>'duration_minutes'
+                                )
+                              )
+                            > ($4->>'event_active_at')::timestamptz
+                    )
+              )
               AND (
                     ($4 ? 'is_online') IS FALSE
                     OR COALESCE((e.metadata->>'is_online')::boolean, false)
@@ -601,6 +648,26 @@ BEGIN
             f.lexical_score::double precision
         FROM fused f
         JOIN eligible e USING (external_id)
+        WHERE (
+            (
+                ($4 ? 'min_semantic_score') IS FALSE
+                AND ($4 ? 'min_lexical_score') IS FALSE
+            )
+            OR (
+                ($4 ? 'min_semantic_score')
+                AND COALESCE(
+                    f.semantic_score >= ($4->>'min_semantic_score')::double precision,
+                    FALSE
+                )
+            )
+            OR (
+                ($4 ? 'min_lexical_score')
+                AND COALESCE(
+                    f.lexical_score >= ($4->>'min_lexical_score')::double precision,
+                    FALSE
+                )
+            )
+        )
         ORDER BY score DESC, f.semantic_score DESC NULLS LAST, e.external_id ASC
         LIMIT GREATEST($3, 1)
     $query$, target_table)
@@ -608,6 +675,11 @@ BEGIN
 END;
 $$;
 
+REVOKE ALL ON FUNCTION public.try_parse_timestamptz(text) FROM PUBLIC;
+REVOKE ALL ON FUNCTION public.try_parse_positive_integer(text) FROM PUBLIC;
+GRANT EXECUTE ON FUNCTION public.try_parse_timestamptz(text) TO nlp_reader;
+GRANT EXECUTE ON FUNCTION public.try_parse_positive_integer(text) TO nlp_reader;
+
 CREATE OR REPLACE FUNCTION get_resource_content_hashes(
     p_resource_type TEXT,
     p_external_ids TEXT[]

sql/verify_global_search_candidate_filters.sql

@@ -0,0 +1,186 @@
+-- Verificador de solo lectura para Busqueda Global V2.
+-- Ejecutar despues de sql/migrations/20260712_01_global_search_candidate_filters.sql.
+
+BEGIN;
+SET TRANSACTION READ ONLY;
+
+WITH function_contract AS (
+    SELECT
+        to_regprocedure(
+            'public.search_resource_embeddings(text,text,vector,integer,jsonb)'
+        ) AS function_oid
+), definition AS (
+    SELECT pg_get_functiondef(function_oid) AS body
+    FROM function_contract
+    WHERE function_oid IS NOT NULL
+)
+SELECT
+    (SELECT function_oid IS NOT NULL FROM function_contract) AS function_exists,
+    COALESCE((SELECT body LIKE '%event_active_at%' FROM definition), FALSE)
+        AS has_event_active_filter,
+    COALESCE((SELECT body LIKE '%min_semantic_score%' FROM definition), FALSE)
+        AS has_semantic_threshold,
+    COALESCE((SELECT body LIKE '%min_lexical_score%' FROM definition), FALSE)
+        AS has_lexical_threshold,
+    COALESCE((SELECT body LIKE '%try_parse_timestamptz%' FROM definition), FALSE)
+        AS has_safe_timestamp_parser,
+    to_regprocedure('public.try_parse_positive_integer(text)') IS NOT NULL
+        AS has_safe_duration_parser,
+    has_function_privilege(
+        'nlp_reader',
+        'public.search_resource_embeddings(text,text,vector,integer,jsonb)',
+        'EXECUTE'
+    ) AS reader_can_execute;
+
+WITH zero_embedding AS (
+    SELECT (
+        '[' || rtrim(repeat('0,', 300), ',') || ']'
+    )::vector AS value
+), negative_results AS (
+    SELECT result.external_id
+    FROM zero_embedding embedding
+    CROSS JOIN LATERAL public.search_resource_embeddings(
+        'users',
+        '__frimeet_verification_no_match_9f7c2__',
+        embedding.value,
+        100,
+        jsonb_build_object(
+            'is_active', TRUE,
+            'min_semantic_score', 1.000001,
+            'min_lexical_score', 1000000.0
+        )
+    ) result
+)
+SELECT COUNT(*) AS negative_query_results
+FROM negative_results;
+
+WITH zero_embedding AS (
+    SELECT (
+        '[' || rtrim(repeat('0,', 300), ',') || ']'
+    )::vector AS value
+), event_results AS (
+    SELECT result.metadata
+    FROM zero_embedding embedding
+    CROSS JOIN LATERAL public.search_resource_embeddings(
+        'events',
+        '__frimeet_event_filter_verification__',
+        embedding.value,
+        1000,
+        jsonb_build_object(
+            'is_active', TRUE,
+            'event_active_at', CURRENT_TIMESTAMP
+        )
+    ) result
+)
+SELECT COUNT(*) AS ended_or_invalid_events_returned
+FROM event_results
+WHERE public.try_parse_timestamptz(metadata->>'start_time') IS NULL
+   OR public.try_parse_positive_integer(metadata->>'duration_minutes') IS NULL
+   OR public.try_parse_timestamptz(metadata->>'start_time')
+      + make_interval(
+            mins => public.try_parse_positive_integer(
+                metadata->>'duration_minutes'
+            )
+        )
+      <= CURRENT_TIMESTAMP;
+
+DO $assertions$
+DECLARE
+    function_body TEXT;
+    zero_embedding VECTOR(300);
+    invalid_count BIGINT;
+BEGIN
+    IF to_regprocedure(
+        'public.search_resource_embeddings(text,text,vector,integer,jsonb)'
+    ) IS NULL THEN
+        RAISE EXCEPTION 'Verify search V2: falta search_resource_embeddings';
+    END IF;
+    IF to_regprocedure('public.try_parse_timestamptz(text)') IS NULL
+       OR to_regprocedure('public.try_parse_positive_integer(text)') IS NULL THEN
+        RAISE EXCEPTION 'Verify search V2: faltan helpers de parseo seguro';
+    END IF;
+
+    SELECT pg_get_functiondef(
+        'public.search_resource_embeddings(text,text,vector,integer,jsonb)'::regprocedure
+    ) INTO function_body;
+    IF function_body NOT LIKE '%event_active_at%'
+       OR function_body NOT LIKE '%min_semantic_score%'
+       OR function_body NOT LIKE '%min_lexical_score%'
+       OR function_body NOT LIKE '%try_parse_timestamptz%'
+       OR function_body NOT LIKE '%try_parse_positive_integer%' THEN
+        RAISE EXCEPTION 'Verify search V2: contrato funcional incompleto';
+    END IF;
+
+    IF NOT has_function_privilege(
+        'nlp_reader',
+        'public.search_resource_embeddings(text,text,vector,integer,jsonb)',
+        'EXECUTE'
+    ) THEN
+        RAISE EXCEPTION 'Verify search V2: nlp_reader no puede ejecutar la busqueda';
+    END IF;
+
+    SELECT COUNT(*) INTO invalid_count
+    FROM public.event_embeddings
+    WHERE is_active = TRUE
+      AND (
+            public.try_parse_timestamptz(metadata->>'start_time') IS NULL
+            OR public.try_parse_positive_integer(
+                metadata->>'duration_minutes'
+            ) IS NULL
+      );
+    IF invalid_count <> 0 THEN
+        RAISE EXCEPTION USING
+            MESSAGE = format(
+                'Verify search V2: existen %s eventos activos con metadata temporal invalida',
+                invalid_count
+            ),
+            HINT = 'Ejecute nuevamente sync_search_embeddings para events antes de activar search.';
+    END IF;
+
+    zero_embedding := (
+        '[' || rtrim(repeat('0,', 300), ',') || ']'
+    )::vector;
+    SELECT COUNT(*) INTO invalid_count
+    FROM public.search_resource_embeddings(
+        'users',
+        '__frimeet_verification_no_match_9f7c2__',
+        zero_embedding,
+        100,
+        jsonb_build_object(
+            'is_active', TRUE,
+            'min_semantic_score', 1.000001,
+            'min_lexical_score', 1000000.0
+        )
+    );
+    IF invalid_count <> 0 THEN
+        RAISE EXCEPTION 'Verify search V2: consulta negativa devolvio % filas', invalid_count;
+    END IF;
+
+    SELECT COUNT(*) INTO invalid_count
+    FROM public.search_resource_embeddings(
+        'events',
+        '__frimeet_event_filter_verification__',
+        zero_embedding,
+        1000,
+        jsonb_build_object(
+            'is_active', TRUE,
+            'event_active_at', CURRENT_TIMESTAMP
+        )
+    ) result
+    WHERE public.try_parse_timestamptz(result.metadata->>'start_time') IS NULL
+       OR public.try_parse_positive_integer(
+            result.metadata->>'duration_minutes'
+          ) IS NULL
+       OR public.try_parse_timestamptz(result.metadata->>'start_time')
+          + make_interval(
+                mins => public.try_parse_positive_integer(
+                    result.metadata->>'duration_minutes'
+                )
+            ) <= CURRENT_TIMESTAMP;
+    IF invalid_count <> 0 THEN
+        RAISE EXCEPTION 'Verify search V2: se devolvieron % eventos invalidos', invalid_count;
+    END IF;
+END;
+$assertions$;
+
+COMMIT;

tests/conftest.py

@@ -7,3 +7,8 @@ os.environ["EMBEDDING_MODEL"] = "mock-embedding"
 os.environ["EMBEDDING_VERSION"] = "test-v1"
 os.environ["GROQ_API_KEY"] = ""
 os.environ["SEARCH_INTERNAL_TOKEN"] = "test-search-token"
+os.environ["NLP_SERVICE_TOKEN"] = "test-nlp-service-token"
+os.environ["PUBLIC_GLOBAL_SEARCH_ENABLED"] = "true"
+os.environ["GLOBAL_SEARCH_MIN_SEMANTIC_SCORE"] = "0.30"
+os.environ["GLOBAL_SEARCH_MIN_LEXICAL_SCORE"] = "0.05"
+os.environ["GLOBAL_SEARCH_THRESHOLD_POLICY_VERSION"] = "test-search-policy-v1"

tests/test_feed_settings.py

@@ -22,3 +22,37 @@ def test_production_rejects_mock_vector_store_for_feed() -> None:
             MAIN_API_INTERNAL_TOKEN="main-token",
             VECTOR_STORE_PROVIDER="mock",
         )
+
+
+def test_search_threshold_overrides_accept_complete_resource_policy() -> None:
+    settings = Settings(
+        _env_file=None,
+        ENV="local",
+        NLP_SERVICE_TOKEN="nlp-token",
+        GLOBAL_SEARCH_RESOURCE_THRESHOLDS_JSON={
+            "posts": {"semantic_min": 0.42, "lexical_min": 0.08}
+        },
+    )
+
+    assert settings.global_search_resource_thresholds["posts"] == {
+        "semantic_min": 0.42,
+        "lexical_min": 0.08,
+    }
+
+
+@pytest.mark.parametrize(
+    "overrides",
+    [
+        {"posts": {"semantic_min": 0.42}},
+        {"unknown": {"semantic_min": 0.42, "lexical_min": 0.08}},
+        {"posts": {"semantic_min": 1.1, "lexical_min": 0.08}},
+    ],
+)
+def test_search_threshold_overrides_reject_invalid_policy(overrides: dict) -> None:
+    with pytest.raises(ValidationError):
+        Settings(
+            _env_file=None,
+            ENV="local",
+            NLP_SERVICE_TOKEN="nlp-token",
+            GLOBAL_SEARCH_RESOURCE_THRESHOLDS_JSON=overrides,
+        )

tests/test_internal_search_candidates.py

@@ -0,0 +1,115 @@
+import asyncio
+
+from fastapi.testclient import TestClient
+
+from app.main import create_app
+from app.modules.search.api.dependencies import get_search_candidates_use_case
+from app.shared.config.settings import get_settings
+
+
+def _payload() -> dict[str, object]:
+    return {
+        "query": "tester1",
+        "resource_types": ["users"],
+        "candidate_limit_per_type": 5,
+        "top_limit": 5,
+        "cursors": {},
+        "as_of": "2026-07-12T12:00:00Z",
+        "filters": {},
+    }
+
+
+def test_internal_candidates_requires_nlp_service_token() -> None:
+    client = TestClient(create_app())
+
+    missing = client.post("/internal/search/candidates", json=_payload())
+    wrong = client.post(
+        "/internal/search/candidates",
+        headers={"Authorization": "Bearer wrong"},
+        json=_payload(),
+    )
+
+    assert missing.status_code == 401
+    assert wrong.status_code == 401
+
+
+def test_internal_candidates_returns_only_ids_and_scores() -> None:
+    client = TestClient(create_app())
+
+    response = client.post(
+        "/internal/search/candidates",
+        headers={"Authorization": "Bearer test-nlp-service-token"},
+        json=_payload(),
+    )
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert set(payload) == {"query", "sections", "pagination", "metadata"}
+    assert payload["metadata"] == {
+        "strategy": "hybrid_thresholded_candidates_v2",
+        "failed_resources": {},
+        "threshold_policy_version": "test-search-policy-v1",
+    }
+    candidate = payload["sections"]["users"][0]
+    assert set(candidate) == {
+        "id",
+        "resource_type",
+        "score",
+        "semantic_score",
+        "lexical_score",
+    }
+    assert "title" not in candidate
+    assert "metadata" not in candidate
+
+
+def test_internal_cursor_is_bound_to_as_of_and_policy_context() -> None:
+    client = TestClient(create_app())
+    headers = {"Authorization": "Bearer test-nlp-service-token"}
+    first_payload = {
+        "query": "lugar tranquilo",
+        "resource_types": ["places"],
+        "candidate_limit_per_type": 1,
+        "top_limit": 1,
+        "cursors": {},
+        "as_of": "2026-07-12T12:00:00Z",
+        "filters": {},
+    }
+    first = client.post(
+        "/internal/search/candidates", headers=headers, json=first_payload
+    )
+    assert first.status_code == 200
+    cursor = first.json()["pagination"]["places"]["next_cursor"]
+    assert cursor
+
+    changed = dict(first_payload)
+    changed["as_of"] = "2026-07-12T12:01:00Z"
+    changed["cursors"] = {"places": cursor}
+    response = client.post(
+        "/internal/search/candidates", headers=headers, json=changed
+    )
+
+    assert response.status_code == 422
+    assert response.json()["detail"] == "cursor belongs to another search context"
+
+
+def test_internal_candidates_returns_503_when_request_budget_expires() -> None:
+    class SlowUseCase:
+        async def execute(self, **kwargs: object) -> None:
+            await asyncio.sleep(0.05)
+
+    app = create_app()
+    app.dependency_overrides[get_search_candidates_use_case] = lambda: SlowUseCase()
+    settings = get_settings()
+    original_timeout = settings.request_timeout_seconds
+    settings.request_timeout_seconds = 0.01  # type: ignore[assignment]
+    try:
+        response = TestClient(app).post(
+            "/internal/search/candidates",
+            headers={"Authorization": "Bearer test-nlp-service-token"},
+            json=_payload(),
+        )
+    finally:
+        settings.request_timeout_seconds = original_timeout
+
+    assert response.status_code == 503
+    assert response.json()["detail"] == "Internal search timed out"

tests/test_pgvector_readiness.py

@@ -0,0 +1,22 @@
+from app.shared.vector_store.aws_pgvector import _read_contract_is_ready
+
+
+def _functions(search_v2: bool) -> dict[str, dict[str, object]]:
+    return {
+        "match_places": {"exists": True, "executable": True},
+        "match_posts": {"exists": True, "executable": True},
+        "search_resource_embeddings": {
+            "exists": True,
+            "executable": True,
+            "candidate_filters_v2": search_v2,
+        },
+        "get_post_feed_features": {"exists": True, "executable": True},
+    }
+
+
+def test_readiness_rejects_legacy_search_function() -> None:
+    assert _read_contract_is_ready(True, _functions(search_v2=False)) is False
+
+
+def test_readiness_accepts_search_candidate_v2_contract() -> None:
+    assert _read_contract_is_ready(True, _functions(search_v2=True)) is True

tests/test_pgvector_search_provider.py

@@ -1,10 +1,14 @@
+from datetime import UTC, datetime
+
 import pytest
 
 from app.modules.search.domain.filters import SearchCriteria
 from app.modules.search.domain.models import SearchResourceType
 from app.modules.search.infrastructure.pgvector_provider import (
+    PgvectorHybridSearchProvider,
     PgvectorPlaceSearchProvider,
 )
+from app.modules.search.domain.relevance import SearchRelevancePolicy
 from app.shared.vector_store.models import VectorMatch
 
 
@@ -66,3 +70,63 @@ async def test_places_use_same_cosine_match_function_as_recommendations() -> Non
     assert hits[0].score == 0.75
     assert hits[0].semantic_score == 0.75
     assert hits[0].lexical_score is None
+
+
+class RecordingHybridVectorClient:
+    def __init__(self) -> None:
+        self.calls: list[dict[str, object]] = []
+
+    async def search_resource_embeddings(self, **kwargs: object) -> list[VectorMatch]:
+        self.calls.append(dict(kwargs))
+        return [
+            VectorMatch(
+                id="rejected",
+                score=0.9,
+                semantic_score=0.39,
+                lexical_score=0.09,
+                metadata={
+                    "start_time": "2026-07-12T11:30:00Z",
+                    "duration_minutes": 120,
+                },
+            ),
+            VectorMatch(
+                id="lexical",
+                score=0.8,
+                semantic_score=0.20,
+                lexical_score=0.10,
+                metadata={
+                    "start_time": "2026-07-12T11:30:00Z",
+                    "duration_minutes": 120,
+                },
+            ),
+        ]
+
+
+@pytest.mark.asyncio
+async def test_hybrid_provider_pushes_event_and_threshold_filters_to_sql() -> None:
+    vector_client = RecordingHybridVectorClient()
+    policy = SearchRelevancePolicy.uniform(semantic_min=0.40, lexical_min=0.10)
+    provider = PgvectorHybridSearchProvider(  # type: ignore[arg-type]
+        SearchResourceType.EVENTS,
+        vector_client,
+        policy,
+    )
+    active_at = datetime(2026, 7, 12, 12, 0, tzinfo=UTC)
+
+    hits = await provider.search(
+        query="evento",
+        embedding=[0.1, 0.2, 0.3],
+        limit=5,
+        offset=0,
+        requester_id=None,
+        criteria=SearchCriteria(event_active_at=active_at),
+    )
+
+    assert [hit.id for hit in hits] == ["lexical"]
+    filters = vector_client.calls[0]["filters"]
+    assert filters == {
+        "is_active": True,
+        "min_semantic_score": 0.40,
+        "min_lexical_score": 0.10,
+        "event_active_at": active_at.isoformat(),
+    }

tests/test_rate_limit.py

@@ -0,0 +1,43 @@
+from types import SimpleNamespace
+
+import pytest
+from fastapi import HTTPException
+from starlette.requests import Request
+
+from app.shared.security import rate_limit
+
+
+def _request(path: str = "/internal/search/candidates") -> Request:
+    return Request(
+        {
+            "type": "http",
+            "http_version": "1.1",
+            "method": "POST",
+            "scheme": "http",
+            "path": path,
+            "raw_path": path.encode(),
+            "query_string": b"",
+            "headers": [],
+            "client": ("127.0.0.1", 12345),
+            "server": ("testserver", 80),
+        }
+    )
+
+
+@pytest.mark.asyncio
+async def test_internal_rate_limit_rejects_requests_over_the_window(monkeypatch) -> None:
+    settings = SimpleNamespace(
+        rate_limit_window_seconds=60,
+        rate_limit_requests_per_window=10,
+        internal_rate_limit_requests_per_window=1,
+    )
+    monkeypatch.setattr(rate_limit, "get_settings", lambda: settings)
+    rate_limit.reset_rate_limit_state()
+
+    await rate_limit.rate_limit_placeholder(_request())
+    with pytest.raises(HTTPException) as captured:
+        await rate_limit.rate_limit_placeholder(_request())
+
+    assert captured.value.status_code == 429
+    assert captured.value.headers["Retry-After"]
+    rate_limit.reset_rate_limit_state()

tests/test_search_architecture.py

@@ -0,0 +1,34 @@
+import ast
+from pathlib import Path
+
+
+def _imports(path: Path) -> list[str]:
+    tree = ast.parse(path.read_text(encoding="utf-8"))
+    result: list[str] = []
+    for node in ast.walk(tree):
+        if isinstance(node, ast.ImportFrom) and node.module:
+            result.append(node.module)
+        elif isinstance(node, ast.Import):
+            result.extend(alias.name for alias in node.names)
+    return result
+
+
+def test_search_domain_and_application_keep_clean_boundaries() -> None:
+    root = Path("app/modules/search")
+    violations: list[str] = []
+    for path in (root / "domain").rglob("*.py"):
+        for imported in _imports(path):
+            if imported.startswith("app.") and not imported.startswith(
+                "app.modules.search.domain"
+            ):
+                violations.append(f"{path}: {imported}")
+    for path in (root / "application").rglob("*.py"):
+        for imported in _imports(path):
+            if imported.startswith("app.") and not imported.startswith(
+                (
+                    "app.modules.search.domain",
+                    "app.modules.search.application",
+                )
+            ):
+                violations.append(f"{path}: {imported}")
+    assert violations == []

tests/test_search_candidates_use_case.py

@@ -0,0 +1,110 @@
+from datetime import UTC, datetime
+
+import pytest
+
+from app.modules.search.application.use_cases.search_candidates import (
+    SearchCandidatesUseCase,
+)
+from app.modules.search.domain.models import SearchHit, SearchResourceType
+
+
+class StubEmbeddingProvider:
+    def embed_text(self, text: str) -> list[float]:
+        assert text
+        return [1.0, 0.0]
+
+
+class StubSearchProvider:
+    def __init__(
+        self,
+        resource_type: SearchResourceType,
+        hits: list[SearchHit] | None = None,
+        error: Exception | None = None,
+    ) -> None:
+        self.resource_type = resource_type
+        self._hits = hits or []
+        self._error = error
+
+    async def search(self, **kwargs: object) -> list[SearchHit]:
+        if self._error is not None:
+            raise self._error
+        offset = int(kwargs["offset"])
+        limit = int(kwargs["limit"])
+        return self._hits[offset : offset + limit]
+
+
+def _hit(identifier: str, score: float) -> SearchHit:
+    return SearchHit(
+        id=identifier,
+        resource_type=SearchResourceType.USERS,
+        title=identifier,
+        score=score,
+        semantic_score=score,
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("hits", "limit", "expected_ids", "has_more"),
+    [
+        ([], 2, [], False),
+        ([_hit("one", 0.9)], 2, ["one"], False),
+        (
+            [_hit("one", 0.9), _hit("two", 0.8), _hit("three", 0.7)],
+            2,
+            ["one", "two"],
+            True,
+        ),
+    ],
+)
+async def test_candidates_returns_zero_one_or_bounded_results_without_fill(
+    hits: list[SearchHit],
+    limit: int,
+    expected_ids: list[str],
+    has_more: bool,
+) -> None:
+    use_case = SearchCandidatesUseCase(
+        embedding_provider=StubEmbeddingProvider(),
+        providers=[StubSearchProvider(SearchResourceType.USERS, hits)],
+        policy_version="test-v1",
+    )
+
+    result = await use_case.execute(
+        query="consulta",
+        resource_types=(SearchResourceType.USERS,),
+        candidate_limit_per_type=limit,
+        as_of=datetime(2026, 7, 12, tzinfo=UTC),
+    )
+
+    assert [item.id for item in result.sections[SearchResourceType.USERS]] == expected_ids
+    page = result.pagination[SearchResourceType.USERS]
+    assert page.returned_count == len(expected_ids)
+    assert page.has_more is has_more
+    assert page.next_offset == (len(expected_ids) if has_more else None)
+
+
+@pytest.mark.asyncio
+async def test_candidates_preserves_successful_sections_when_one_provider_fails() -> None:
+    use_case = SearchCandidatesUseCase(
+        embedding_provider=StubEmbeddingProvider(),
+        providers=[
+            StubSearchProvider(SearchResourceType.USERS, [_hit("user", 0.9)]),
+            StubSearchProvider(
+                SearchResourceType.EVENTS,
+                error=TimeoutError("database timeout"),
+            ),
+        ],
+        policy_version="test-v1",
+    )
+
+    result = await use_case.execute(
+        query="consulta",
+        resource_types=(SearchResourceType.USERS, SearchResourceType.EVENTS),
+        candidate_limit_per_type=2,
+        as_of=datetime(2026, 7, 12, tzinfo=UTC),
+    )
+
+    assert [item.id for item in result.sections[SearchResourceType.USERS]] == ["user"]
+    assert result.sections[SearchResourceType.EVENTS] == []
+    assert result.failed_resources == {SearchResourceType.EVENTS: "TimeoutError"}
+    assert result.pagination[SearchResourceType.EVENTS].has_more is False

tests/test_search_relevance.py

@@ -0,0 +1,80 @@
+from datetime import UTC, datetime
+
+from app.modules.search.domain.filters import SearchCriteria, filter_and_rank_hits
+from app.modules.search.domain.models import SearchHit, SearchResourceType
+from app.modules.search.domain.relevance import SearchRelevanceThreshold
+
+
+def _hit(
+    semantic_score: float | None,
+    lexical_score: float | None,
+) -> SearchHit:
+    return SearchHit(
+        id="candidate",
+        resource_type=SearchResourceType.POSTS,
+        title="candidate",
+        score=0.9,
+        semantic_score=semantic_score,
+        lexical_score=lexical_score,
+    )
+
+
+def test_relevance_threshold_accepts_either_absolute_signal() -> None:
+    threshold = SearchRelevanceThreshold(semantic_min=0.40, lexical_min=0.10)
+
+    assert threshold.accepts(_hit(0.40, None))
+    assert threshold.accepts(_hit(0.20, 0.10))
+    assert not threshold.accepts(_hit(0.399, 0.099))
+    assert not threshold.accepts(_hit(None, None))
+
+
+def test_event_active_filter_keeps_in_progress_and_future_only() -> None:
+    active_at = datetime(2026, 7, 12, 12, 0, tzinfo=UTC)
+    criteria = SearchCriteria(event_active_at=active_at)
+    hits = [
+        SearchHit(
+            id="ended",
+            resource_type=SearchResourceType.EVENTS,
+            title="ended",
+            score=0.9,
+            semantic_score=0.9,
+            metadata={
+                "start_time": "2026-07-12T10:00:00Z",
+                "duration_minutes": 60,
+            },
+        ),
+        SearchHit(
+            id="in-progress",
+            resource_type=SearchResourceType.EVENTS,
+            title="in-progress",
+            score=0.8,
+            semantic_score=0.8,
+            metadata={
+                "start_time": "2026-07-12T11:30:00Z",
+                "duration_minutes": 90,
+            },
+        ),
+        SearchHit(
+            id="future",
+            resource_type=SearchResourceType.EVENTS,
+            title="future",
+            score=0.7,
+            semantic_score=0.7,
+            metadata={
+                "start_time": "2026-07-13T10:00:00Z",
+                "duration_minutes": 60,
+            },
+        ),
+        SearchHit(
+            id="invalid",
+            resource_type=SearchResourceType.EVENTS,
+            title="invalid",
+            score=1.0,
+            semantic_score=1.0,
+            metadata={"start_time": "invalid", "duration_minutes": 60},
+        ),
+    ]
+
+    ranked = filter_and_rank_hits(SearchResourceType.EVENTS, hits, criteria)
+
+    assert [item.id for item in ranked] == ["in-progress", "future"]

tests/test_search_resource_documents.py

@@ -90,14 +90,50 @@ def test_event_identifier_tags_are_not_embedded_as_semantics() -> None:
             "title": "Fiesta de Cumpleanos",
             "description": "Evento totalmente independiente",
             "is_public": True,
-            "tags": ["7", "b83ab97e-91a4-4f69-b102-b27c6092e9cb"],
+            "tags": ["musica", "b83ab97e-91a4-4f69-b102-b27c6092e9cb"],
+            "tag_ids": ["7", "b83ab97e-91a4-4f69-b102-b27c6092e9cb"],
         }
     )
     assert record is not None
     assert "b83ab97e" not in record.document
+    assert "musica" in record.document
     assert record.metadata["tag_ids"] == ["7", "b83ab97e-91a4-4f69-b102-b27c6092e9cb"]
 
 
+def test_event_sync_normalizes_valid_temporal_metadata() -> None:
+    record = event_to_source_record(
+        {
+            "id": "event-valid",
+            "title": "Evento futuro",
+            "start_time": "2026-07-12T14:00:00-06:00",
+            "duration_minutes": "90",
+        }
+    )
+
+    assert record is not None
+    assert record.is_active is True
+    assert record.metadata["start_time"] == "2026-07-12T20:00:00+00:00"
+    assert record.metadata["duration_minutes"] == 90
+    assert record.metadata["temporal_metadata_valid"] is True
+
+
+def test_event_sync_deactivates_invalid_temporal_metadata() -> None:
+    record = event_to_source_record(
+        {
+            "id": "event-invalid",
+            "title": "Evento invalido",
+            "start_time": "not-a-date",
+            "duration_minutes": 60,
+        }
+    )
+
+    assert record is not None
+    assert record.is_active is False
+    assert record.metadata["temporal_metadata_valid"] is False
+    assert "start_time" not in record.metadata
+    assert "duration_minutes" not in record.metadata
+
+
 def test_group_builds_private_search_acl_from_members_and_invites() -> None:
     record = group_to_source_record(
         {

tests/test_search_snapshot_sources.py

@@ -0,0 +1,52 @@
+import pytest
+
+from app.modules.clubs.infrastructure.main_api_club_source import MainApiClubsClient
+from app.modules.events.infrastructure.main_api_event_source import MainApiEventsClient
+from app.modules.groups.infrastructure.main_api_group_source import MainApiGroupsClient
+from app.modules.users.infrastructure.main_api_user_source import MainApiUsersClient
+from app.shared.config.settings import Settings
+
+
+def test_search_sources_use_internal_snapshot_paths() -> None:
+    settings = Settings(_env_file=None, ENV="local", MAIN_API_INTERNAL_TOKEN="internal")
+
+    assert MainApiUsersClient(settings)._path == "api/v1/internal/search/users/snapshot"
+    assert MainApiClubsClient(settings)._path == "api/v1/internal/search/clubs/snapshot"
+    assert MainApiGroupsClient(settings)._path == "api/v1/internal/search/groups/snapshot"
+    assert MainApiEventsClient(settings)._path == "api/v1/internal/search/events/snapshot"
+
+
+def test_search_source_uses_only_internal_service_token() -> None:
+    settings = Settings(
+        _env_file=None,
+        ENV="local",
+        MAIN_API_INTERNAL_TOKEN="service-token",
+        MAIN_API_AUTH_TOKEN="user-jwt-must-not-be-used",
+    )
+
+    assert MainApiUsersClient(settings)._build_headers() == {
+        "Authorization": "Bearer service-token"
+    }
+
+
+def test_search_source_rejects_missing_internal_token_even_with_user_jwt() -> None:
+    settings = Settings(
+        _env_file=None,
+        ENV="local",
+        MAIN_API_AUTH_TOKEN="user-jwt-must-not-be-used",
+    )
+
+    with pytest.raises(RuntimeError, match="MAIN_API_INTERNAL_TOKEN"):
+        MainApiUsersClient(settings)._build_headers()
+
+
+def test_search_source_rejects_public_search_path() -> None:
+    settings = Settings(
+        _env_file=None,
+        ENV="local",
+        MAIN_API_INTERNAL_TOKEN="service-token",
+        MAIN_API_USERS_SNAPSHOT_PATH="/api/v1/users/search",
+    )
+
+    with pytest.raises(ValueError, match="internal snapshot"):
+        MainApiUsersClient(settings)

tests/test_search_threshold_calibration.py

@@ -0,0 +1,16 @@
+from scripts.calibrate_global_search_thresholds import LabeledScore, calibrate_resource
+
+
+def test_calibration_selects_thresholds_that_reject_negative_samples() -> None:
+    samples = [
+        LabeledScore("users", True, 0.80, None),
+        LabeledScore("users", True, 0.70, 0.20),
+        LabeledScore("users", False, 0.10, 0.01),
+        LabeledScore("users", False, 0.20, 0.02),
+    ]
+
+    result = calibrate_resource(samples, target_precision=1.0)
+
+    assert result["precision"] == 1.0
+    assert result["recall"] == 1.0
+    assert result["false_positives"] == 0

tests/test_sql_contract.py

@@ -58,3 +58,48 @@ def test_post_feed_v2_and_convergent_schema_are_present() -> None:
     assert "CREATE TABLE IF NOT EXISTS post_embeddings" in convergent
     assert "CREATE OR REPLACE FUNCTION public.get_post_feed_features" in convergent
     assert "migrate_post_feed_v2.sql" not in convergent  # contenido inline, sin includes
+
+
+def test_global_search_candidate_filters_are_convergent_and_incremental() -> None:
+    contract = Path("sql/aws_pgvector_contract.sql").read_text(encoding="utf-8")
+    convergent = Path("sql/new_pgvector_schema.sql").read_text(encoding="utf-8")
+    migration = Path(
+        "sql/migrations/20260712_01_global_search_candidate_filters.sql"
+    ).read_text(encoding="utf-8")
+    verifier = Path("sql/verify_global_search_candidate_filters.sql")
+
+    for token in (
+        "event_active_at",
+        "min_semantic_score",
+        "min_lexical_score",
+        "make_interval",
+        "try_parse_timestamptz",
+        "try_parse_positive_integer",
+    ):
+        assert token in contract
+        assert token in convergent
+        assert token in migration
+    upper_migration = migration.upper()
+    assert "BEGIN;" in upper_migration
+    assert "COMMIT;" in upper_migration
+    assert "CREATE OR REPLACE FUNCTION PUBLIC.SEARCH_RESOURCE_EMBEDDINGS" in upper_migration
+    assert "CREATE TABLE" not in upper_migration
+    assert "ALTER TABLE" not in upper_migration
+    assert "CREATE INDEX" not in upper_migration
+    assert "DROP FUNCTION" not in upper_migration
+    assert "TO NLP_READER" in upper_migration
+    assert verifier.exists()
+    verifier_sql = verifier.read_text(encoding="utf-8")
+    assert "RAISE EXCEPTION 'Verify search V2" in verifier_sql
+    assert "SET TRANSACTION READ ONLY" in verifier_sql
+
+
+def test_global_search_sql_does_not_directly_cast_untrusted_event_metadata() -> None:
+    for path in (
+        Path("sql/aws_pgvector_contract.sql"),
+        Path("sql/new_pgvector_schema.sql"),
+        Path("sql/migrations/20260712_01_global_search_candidate_filters.sql"),
+    ):
+        sql = path.read_text(encoding="utf-8")
+        assert "NULLIF(e.metadata->>'start_time', '')::timestamptz" not in sql
+        assert "(e.metadata->>'duration_minutes')::integer" not in sql