Feat/indexing callback url

CLAUDE.md

@@ -248,6 +248,70 @@ Optional web search augmentation via the Staan API, allowing the LLM to combine
 - `openrag/services/orchestrators/query_service.py` — `_prepare_for_web_only()`, web search logic in `_prepare_for_chat_completion()`
 - `openrag/api/routers/user/chat.py` — `__prepare_sources()` merges document and web sources
 
+### Indexing Status Callbacks
+
+Indexing is asynchronous, so a client can either poll the task-status URL or
+hand OpenRag a URL to notify once the task settles.
+
+**Request fields** (multipart form, on `POST` and `PUT /indexer/partition/{partition}/file/{file_id}`):
+- `callback_url` — POSTed once the task reaches a terminal state
+- `callback_token` — sent as `Authorization: Bearer <token>` on that POST
+
+**Body:** `{"partition", "file_id", "status": "success"|"error", "metadata"}`. `metadata` is
+the upload metadata echoed back **minus** `UPLOAD_METADATA_SERVER_KEYS` (`core/utils/consts.py`:
+`source`, `filename`, `original_filename`, `file_size`, `file_id`, `content_sha256`) — an exclusion,
+not a fixed field list, so any caller-supplied field (cozy-stack's revision marker is `doc_rev`)
+travels through unrecomputed and under whatever name the caller gave it; a key the caller never sent
+is simply absent, not echoed back as `null`. The exclusion exists because the target is a
+caller-supplied URL and `_build_metadata` merges those server-computed keys into the same dict —
+`source` (the server's on-disk path) is the load-bearing one. `test_build_metadata_only_adds_keys_in_upload_metadata_server_keys`
+(`tests/unit/services/orchestrators/test_indexing_service.py`) fails the build if `_build_metadata`
+ever injects a key the constant doesn't cover.
+
+**Guarantees:** one attempt, no retries, a 5 s deadline on the request, never raises — a failed
+callback is logged and cannot change the indexing outcome. No `callback_url` → strict no-op. A
+user-cancelled task sends nothing (only a real failure notifies `"error"`), which is why the sender
+keys off the return value of `set_failed_if_not_cancelled` and the pool's pre-flight handler catches
+`Exception`, not `BaseException`.
+
+**Not guaranteed:** delivery. The send is awaited on the worker's slot, so a blackholing target costs
+up to 5 s of indexing throughput per file, and an actor lost before the task settles notifies
+nothing at all. Clients keep a timeout and fall back to polling the task-status URL.
+
+`callback_url` is checked against the SSRF guard (`is_safe_url` — scheme, loopback/private/link-local,
+decimal/hex/octal/short-form IPv4 literals, all normalized through `socket.inet_aton` so a legacy
+numeric spelling can't overflow `ipaddress.ip_address(int(...))` into a false-negative IPv6 address)
+but not resolved: neither a DNS lookup nor an https requirement is enforced beyond that literal
+check, deliberately — the URL is caller-supplied, so picking a safe target is the caller's call, not
+OpenRag's to police. Checked twice — in the router (immediate `400`) and again in the sender (a
+direct caller bypasses the router) — so accepting a hostname the sender would refuse never happens.
+`INDEXING_CALLBACK_ALLOW_PRIVATE_URLS=true` / `indexing_callback.allow_private_urls` lifts the
+*address* half of the guard for dev stacks whose target is a local instance; the scheme check always
+applies. Keep it off in production: with it on, any user allowed to upload can make the server POST
+to an internal address.
+
+**Rolling deploys:** the worker actors are named, detached and `get_if_exists`, so changing
+`process_file`'s remote contract requires bumping `_INDEXER_ACTOR_PROTOCOL_VERSION` in
+`indexer_pool.py` (v3 → v4 for `callback_url`/`callback_token`; v5 added worker-ref-registration wait
+and TSM `set_state` fencing; v7 folds in a second, independent v6 lineage — STT-preset-aware registry
+hydration plus the `_active_indexation_config` contextvar — that landed on `develop` under the same
+version string while this branch's own v6 was in flight). Without the bump, new replicas attach to the
+previous release's actors and every submit raises `TypeError`. Old generations are retired with
+`services/workers/retire_indexer_generation.py`.
+
+**Key files:**
+- `openrag/services/workers/indexing_callback.py` — `send_indexing_callback()` (was `webhook.py`; the
+  target is a normal authenticated route now, not an unauthenticated webhook trigger)
+- `openrag/core/utils/url_safety.py` — `is_safe_url(url, *, allow_private_hosts=False)`, shared with
+  the MCP `index_url` tool (which keeps the strict default). The web-search content fetcher
+  (`services/websearch/content_fetcher.py`) does **not** import this — it has its own older,
+  un-synced `_is_safe_url`, so hardening this module does not automatically harden that one.
+- `openrag/core/config/indexation.py` — `IndexingCallbackConfig`
+
+Both fields travel the same chain as the rest of an indexing job: `api/routers/admin/indexing.py` →
+`services/orchestrators/indexing_service.py` → `core/indexing/dispatcher.py` (port) →
+`services/workers/dispatcher.py` → `indexer_pool.py` → `indexer_actor.py` → `indexing_callback.py`.
+
 ### File Quota System
 
 Per-user file quota enforcement tracked via the `file_count` and `file_quota` columns on `users`, and `created_by` on `files`.

conf/config.yaml

@@ -279,6 +279,13 @@ loader:
     concurrency_limit: 20
     enable_thinking: null
 
+# --- Indexing callback ---
+# Env: INDEXING_CALLBACK_ALLOW_PRIVATE_URLS
+indexing_callback:
+  # Dev only. Keep false in production: it disables the SSRF guard on a
+  # caller-supplied URL.
+  allow_private_urls: false
+
 # --- Ray ---
 ray:
   indexer:

docs/content/docs/documentation/API.mdx

@@ -85,11 +85,50 @@ Upload a new file to a specific partition for indexing.
 **Request Body (form-data):**
 - `file` (binary): File to upload
 - `metadata` (JSON string): File metadata (e.g., `{"owner": "user1"}`)
+- `workspace_ids` (JSON array, optional): Workspaces to add the file to
+- `callback_url` (string, optional): URL notified once indexing reaches a terminal state
+- `callback_token` (string, optional): Bearer token for that notification
 
 **Responses:**
 - `201 Created`: Returns task status URL
+- `400 Bad Request`: `callback_url` is malformed or not a public `http(s)` URL
 - `409 Conflict`: File already exists in partition
 
+##### Indexing status callback
+
+Indexing is asynchronous. Pass a `callback_url` and OpenRAG POSTs the outcome
+once the task settles, so a client can wait instead of polling:
+
+```json
+{
+  "partition": "alice.example.org",
+  "file_id": "file-123",
+  "status": "success",
+  "metadata": {"doc_rev": "<revision>", "datetime": "...", "doctype": "..."}
+}
+```
+
+`status` is `"success"` or `"error"`; a user-cancelled task sends nothing.
+`metadata` is whatever you sent at upload, echoed back unchanged — including
+a revision-tracking field under any name you like, if that's how your receiver
+orders these callbacks (the example above uses `doc_rev`). A field you didn't
+send is simply absent, not sent back as `null`. Only server-computed keys are
+held back: the on-disk path, content hash, file size, filenames, and
+`file_id` (already a top-level field).
+
+For an authenticated target, pass `callback_token` — sent as `Authorization:
+Bearer <token>`, never in the URL or the payload. Picking a safe scheme for it
+is on the caller: `callback_url` is otherwise unchecked beyond the public
+`http(s)` requirement below.
+
+Best-effort: one attempt, no retries, a 5 s deadline, and any failure is
+logged without affecting the indexing result. Delivery is not guaranteed, so
+keep a client-side timeout and fall back to polling the task-status URL. The
+`callback_url` is checked against the SSRF guard and must be a public
+`http(s)` address — see
+[`INDEXING_CALLBACK_ALLOW_PRIVATE_URLS`](/openrag/documentation/env_vars) to
+target a local instance in development.
+
 ##### Temporal Filtering
 OpenRAG supports temporal filtering to retrieve documents from specific time periods.
 The client can include the temporal field to allow temporal-aware search in search endpoints.

docs/content/docs/documentation/env_vars.md

@@ -25,6 +25,7 @@ Openrag loads all files into a pivot markdown file format before proceeding to c
 | `CONTENT_DEDUPLICATION_ENABLED` | `bool` | `true` | Rejects a file when identical content already exists in the same partition. Set it to `false` when a test intentionally indexes duplicates. |
 | `PDFLOADER` | `str` | `PyMuPDFLoader` | PDF parsing engine. `PyMuPDFLoader` (default) is a lightweight, fast, CPU-friendly backend for searchable PDFs. Switch to `MarkerLoader` for OCR / scanned documents, complex layouts and embedded images (heavier; GPU-friendly). Other options: `DoclingLoader`, `DotsOCRLoader`.|
 | `PARSE_TIMEOUT` | `int` | `3600` | Outer wall-clock bound (in seconds) for a single file's parse stage, whichever loader runs it. Marker and Docling self-limit via their own timeouts, but `PyMuPDFLoader` has none — this bound stops a wedged parse from stalling indexing: the file fails and is reported instead. |
+| `INDEXING_CALLBACK_ALLOW_PRIVATE_URLS` | `bool` | `false` | Development-only escape hatch for the upload `callback_url`. By default a callback targeting `localhost`, a private or link-local address is refused (SSRF guard on a caller-supplied URL). Set it to `true` only in a dev stack whose callback target is a local instance (e.g. `http://cozy.localhost:8080`). Non-`http(s)` schemes stay rejected either way. |
 
 :::caution
 `PyMuPDFLoader` (the default) is a lightweight PDF loader that cannot process non-searchable (image-based) PDFs and does not extract or handle embedded images. Set `PDFLOADER=MarkerLoader` when you need those.

openrag/api/routers/admin/indexing.py

@@ -14,6 +14,7 @@
 import json
 from pathlib import Path
 from typing import Any
+from urllib.parse import urlparse
 
 from api.dependencies.auth import (
     check_user_file_quota,
@@ -36,6 +37,7 @@
 from core.utils.filename import sanitize_filename
 from core.utils.log_tail import app_log_file
 from core.utils.logging import get_logger
+from core.utils.url_safety import is_safe_url
 from di.providers import get_auth_service, get_config, get_indexing_service, get_partition_service
 from fastapi import (
     APIRouter,
@@ -52,6 +54,30 @@
 logger = get_logger()
 
 
+def _validate_callback_url(callback_url: str | None, config) -> None:
+    """Reject a callback_url the server must not POST to.
+
+    The sender re-checks; this is only so the caller hears about it now rather
+    than losing the callback silently.
+    """
+    if not callback_url:
+        return
+    allow_private = bool(getattr(getattr(config, "indexing_callback", None), "allow_private_urls", False))
+    try:
+        # is_safe_url never touches the port; a non-numeric one raises here.
+        urlparse(callback_url).port
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="callback_url is not a valid URL",
+        ) from exc
+    if not is_safe_url(callback_url, allow_private_hosts=allow_private):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="callback_url must be a public http(s) URL",
+        )
+
+
 def build_url(request: Request, route_name: str, *, preferred_url_scheme: str | None = None, **path_params) -> str:
     """Build a URL using the preferred scheme if configured."""
     url = request.url_for(route_name, **path_params)
@@ -126,11 +152,18 @@ async def add_file(
     file: UploadFile = Depends(validate_file_format),
     metadata: dict = Depends(validate_metadata),
     workspace_ids: str | None = Form(None, description="JSON array of workspace IDs to add the file to"),
+    callback_url: str | None = Form(None, description="Optional URL notified when async indexing finishes"),
+    callback_token: str | None = Form(
+        None,
+        description="Optional bearer token sent as `Authorization` on the callback_url request",
+    ),
     user=Depends(require_partition_editor),
     _quota_check=Depends(check_user_file_quota),
     config=Depends(get_config),
     service=Depends(get_indexing_service),
 ):
+    _validate_callback_url(callback_url, config)
+
     if await service.file_exists(file_id, partition):
         raise HTTPException(
             status_code=status.HTTP_409_CONFLICT,
@@ -195,6 +228,8 @@ async def add_file(
             user=user,
             workspace_ids=parsed_workspace_ids,
             content_sha256=content_sha256,
+            callback_url=callback_url,
+            callback_token=callback_token,
         )
     except BaseException as exc:
         # A submission whose outcome is unknown may have left a worker running
@@ -284,10 +319,17 @@ async def put_file(
     file_id: str = Depends(validate_file_id),
     file: UploadFile = Depends(validate_file_format),
     metadata: dict = Depends(validate_metadata),
+    callback_url: str | None = Form(None, description="Optional URL notified when async indexing finishes"),
+    callback_token: str | None = Form(
+        None,
+        description="Optional bearer token sent as `Authorization` on the callback_url request",
+    ),
     user=Depends(require_partition_editor),
     config=Depends(get_config),
     service=Depends(get_indexing_service),
 ):
+    _validate_callback_url(callback_url, config)
+
     if not await service.file_exists(file_id, partition):
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
@@ -334,6 +376,8 @@ async def put_file(
             user=user,
             replace=True,
             content_sha256=content_sha256,
+            callback_url=callback_url,
+            callback_token=callback_token,
         )
     except BaseException as exc:
         # A submission whose outcome is unknown may have left a worker running

openrag/core/config/indexation.py

@@ -201,3 +201,13 @@ class LoaderConfig(ConfigMixin):
     # Max depth of nested .eml-in-.eml attachments the EmlLoader will descend
     # into. Bounds recursion when .eml files are nested inside one another.
     eml_max_recursion_depth: int = 5
+
+
+class IndexingCallbackConfig(ConfigMixin):
+    """Server-side policy for the caller-supplied per-upload ``callback_url``.
+
+    With ``allow_private_urls`` on, anyone allowed to upload can make the
+    server POST to an internal address. Dev stacks only.
+    """
+
+    allow_private_urls: bool = False

openrag/core/config/loader.py

@@ -100,6 +100,8 @@
     ("SAVE_MARKDOWN", "loader.save_markdown", bool),
     ("SAVE_UPLOADED_FILES", "loader.save_uploaded_files", bool),
     ("CONTENT_DEDUPLICATION_ENABLED", "loader.content_deduplication_enabled", bool),
+    # Indexing callback
+    ("INDEXING_CALLBACK_ALLOW_PRIVATE_URLS", "indexing_callback.allow_private_urls", bool),
     ("PDFLOADER", "loader.file_loaders.pdf", str),
     ("AUDIOLOADER", "loader.file_loaders.wav", str),
     ("MARKER_MAX_TASKS_PER_CHILD", "loader.marker_max_tasks_per_child", int),

openrag/core/config/root.py

@@ -14,7 +14,7 @@
     SemaphoreConfig,
     VLMConfig,
 )
-from .indexation import LoaderConfig
+from .indexation import IndexingCallbackConfig, LoaderConfig
 from .infrastructure import (
     PathsConfig,
     PromptsConfig,
@@ -60,6 +60,7 @@ class Settings(ConfigMixin):
     paths: PathsConfig = Field(default_factory=PathsConfig)
     prompts: PromptsConfig = Field(default_factory=PromptsConfig)
     loader: LoaderConfig = Field(default_factory=LoaderConfig)
+    indexing_callback: IndexingCallbackConfig = Field(default_factory=IndexingCallbackConfig)
     ray: RayConfig = Field(default_factory=RayConfig)
     chunker: ChunkerConfig = Field(default_factory=ChunkerConfig)
     retriever: RetrieverConfig = Field(default_factory=SingleRetrieverConfig)

openrag/core/indexing/dispatcher.py

@@ -37,6 +37,8 @@ async def dispatch_indexing(
         replace: bool,
         indexation_config: dict | None = None,
         embedder_name: str | None = None,
+        callback_url: str | None = None,
+        callback_token: str | None = None,
         require_existing_partition: bool = False,
         allow_legacy_require_existing_partition_retry: bool = False,
     ) -> str:

openrag/core/utils/consts.py

@@ -51,3 +51,13 @@ def strip_protected_metadata(metadata: dict | None) -> tuple[dict, list[str]]:
     for key in removed:
         del md[key]
     return md, removed
+
+
+# The server-computed keys ``IndexingService._build_metadata`` merges into
+# upload metadata. The indexing-status callback excludes exactly these before
+# echoing metadata to a caller-supplied URL. Keep in sync with
+# ``_build_metadata`` — enforced by
+# ``test_build_metadata_only_adds_keys_in_upload_metadata_server_keys``.
+UPLOAD_METADATA_SERVER_KEYS: frozenset[str] = frozenset(
+    {"source", "filename", "original_filename", "file_size", "file_id", "content_sha256"}
+)