fix: make runtime prefetch task registration executable end-to-end

SchedulerClient.register_volatile_fetch (consolidation's prefetch
routing) registered tasks that were dead on arrival:
- JSON body stored under 'body', which rest_api_executor ignores
  (it only reads config['payload'])
- no auth block, so the scheduled POST to /volatile/fetch would 401
  against library-desk's verify_api_key
- user placed in the body while /volatile/fetch endpoints require it
  as a query parameter (RequiredUserQuery) - would 422 regardless

The task config now carries user in the URL query string (encoded),
an empty payload, and auth {type: bearer, token: ${LIBRARY_API_KEY}}
substituted Scheduler-side (never stored raw).

SchedulerClient also sent no Authorization to the Scheduler API itself,
so registration 401'd silently at consolidation time; it now sends
Bearer auth from the new scheduler_api_key setting.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QbFZyDvYksazX6nYQYZ67L

CHANGELOG.md

@@ -20,6 +20,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed (Phase D review batch)
 
+- **Runtime prefetch registration is executable** - `SchedulerClient.register_volatile_fetch` (used by consolidation's `_register_prefetch`) registered tasks that were dead on arrival three ways: the JSON body sat under the ignored `body` key (`rest_api_executor` only reads `config["payload"]`), there was no `auth` block (the scheduled POST would 401 against library-desk's `verify_api_key`), and `user` was in the body while every `/volatile/fetch` endpoint requires it as a QUERY parameter (would 422). The config now puts `user` in the URL query string (URL-encoded), an empty `payload`, and `auth: {type: bearer, token: "${LIBRARY_API_KEY}"}` (substituted from the Scheduler's environment; never stored raw). `SchedulerClient` itself also sent no Authorization to the Scheduler API, so registration failed silently at consolidation time — it now sends `Authorization: Bearer <SCHEDULER_API_KEY>` (new `scheduler_api_key` setting; a client without a key logs a warning).
 - **Registrar authenticates to the Scheduler API** - `scripts/register_scheduler_tasks.py --execute` sent no `Authorization` header while the Scheduler's task-management endpoints are Bearer-guarded (`verify_api_key`: 401 on missing key), so the existence probes 401'd (misread as "task absent") and every registration failed; only the public `/health` gate passed. `--execute` now requires `SCHEDULER_API_KEY` in the environment (refuses to run without it, key is never stored) and sends `Authorization: Bearer $SCHEDULER_API_KEY` on all of its own HTTP calls. Deploy note updated alongside the existing `LIBRARY_API_KEY` requirement.
 
 ### Fixed (hazards batch)

src/clients/scheduler_client.py

@@ -8,10 +8,17 @@ Registers and manages scheduled tasks for prefetch operations
 import httpx
 import logging
 from typing import Optional, Any
+from urllib.parse import quote
 from pydantic import BaseModel, Field
 
 logger = logging.getLogger(__name__)
 
+#: Literal placeholder stored in task auth.token; the Scheduler's
+#: rest_api_executor substitutes ${ENV_VAR} from ITS OWN environment at
+#: execution time, so the raw library-desk key is never stored in the
+#: scheduled_tasks.config column.
+LIBRARY_API_KEY_PLACEHOLDER = "${LIBRARY_API_KEY}"
+
 
 class SchedulerTask(BaseModel):
     """Task definition for scheduler registration."""
@@ -39,24 +46,38 @@ class SchedulerTask(BaseModel):
 class SchedulerClient:
     """Client for external scheduler service."""
 
-    def __init__(self, base_url: str, timeout: float = 30.0):
+    def __init__(self, base_url: str, api_key: str = "", timeout: float = 30.0):
         """
         Initialize scheduler client.
 
         Args:
             base_url: Scheduler API base URL (e.g., "http://scheduler:8090")
+            api_key: Bearer key for the Scheduler's task-management API.
+                The /tasks endpoints are guarded by verify_api_key (401 when
+                the Authorization header is missing), so without this key
+                every registration call fails.
             timeout: HTTP request timeout in seconds
         """
         self.base_url = base_url.rstrip("/")
+        self.api_key = api_key
         self.timeout = timeout
         self._client: Optional[httpx.AsyncClient] = None
+        if not api_key:
+            logger.warning(
+                "SchedulerClient created without an API key; task-management "
+                "calls will be rejected by the Scheduler (401)"
+            )
 
     async def _get_client(self) -> httpx.AsyncClient:
-        """Get or create HTTP client."""
+        """Get or create HTTP client (with Scheduler API Bearer auth)."""
         if self._client is None or self._client.is_closed:
+            headers = (
+                {"Authorization": f"Bearer {self.api_key}"} if self.api_key else {}
+            )
             self._client = httpx.AsyncClient(
                 base_url=self.base_url,
                 timeout=self.timeout,
+                headers=headers,
             )
         return self._client
 
@@ -301,13 +322,25 @@ class SchedulerClient:
             day_of_week=schedule.get("day_of_week", -1),
             config={
                 "method": "POST",
-                "url": f"http://library-desk:8089/volatile/fetch/{namespace}/{key}",
+                # /volatile/fetch endpoints take user as a REQUIRED QUERY
+                # parameter (RequiredUserQuery) - a body user would 422.
+                "url": (
+                    f"http://library-desk:8089/volatile/fetch/"
+                    f"{namespace}/{key}?user={quote(user, safe='')}"
+                ),
                 "headers": {
                     "Content-Type": "application/json"
                 },
-                "body": {
-                    "user": user
-                }
+                # rest_api_executor only reads config["payload"] as the JSON
+                # body (a "body" key is silently ignored).
+                "payload": {},
+                # Substituted from the Scheduler's environment at execution
+                # time; without it the scheduled POST 401s against
+                # library-desk's verify_api_key.
+                "auth": {
+                    "type": "bearer",
+                    "token": LIBRARY_API_KEY_PLACEHOLDER,
+                },
             }
         )
 

src/config.py

@@ -145,6 +145,13 @@ class Settings(BaseSettings):
 
     # Scheduler Service
     scheduler_url: str = Field(default="http://scheduler:8090", description="Scheduler service URL")
+    scheduler_api_key: str = Field(
+        default="",
+        description=(
+            "Bearer key for the Scheduler's auth-guarded task-management API; "
+            "required for runtime prefetch task registration"
+        ),
+    )
 
     @property
     def qdrant_url(self) -> str:

src/core/dependencies.py

@@ -259,7 +259,10 @@ def get_scheduler_client() -> SchedulerClient:
     Note: Used for registering prefetch tasks discovered during HybridRAG searches
     """
     settings = get_settings()
-    client = SchedulerClient(base_url=settings.scheduler_url)
+    client = SchedulerClient(
+        base_url=settings.scheduler_url,
+        api_key=settings.scheduler_api_key,
+    )
     logger.debug(f"Created Scheduler client: {settings.scheduler_url}")
     return client
 

tests/test_scheduler_client.py

@@ -0,0 +1,122 @@
+"""
+Offline unit tests for SchedulerClient (Phase D review batch).
+
+Pins the runtime prefetch-registration fixes:
+- SchedulerClient sends Authorization: Bearer <SCHEDULER_API_KEY> on its
+  own calls (the Scheduler's /tasks endpoints are auth-guarded; a bare
+  client 401s and registration fails silently at consolidation time)
+- register_volatile_fetch task config is actually executable by the
+  Scheduler's rest_api_executor:
+  * JSON body under config["payload"] (a "body" key is silently ignored)
+  * user as a QUERY parameter (the /volatile/fetch endpoints use
+    RequiredUserQuery; a body user would 422)
+  * an auth block with the ${LIBRARY_API_KEY} placeholder so the
+    scheduled POST passes library-desk's verify_api_key
+"""
+
+import httpx
+import pytest
+
+from src.clients.scheduler_client import (
+    LIBRARY_API_KEY_PLACEHOLDER,
+    SchedulerClient,
+)
+
+
+def _mock_transport(seen):
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.append(request)
+        path = request.url.path
+        if request.method == "GET" and path.startswith("/tasks/"):
+            return httpx.Response(404)  # task does not exist yet
+        if request.method == "POST" and path == "/tasks":
+            return httpx.Response(200, json={"ok": True})
+        return httpx.Response(200, json={})
+
+    return httpx.MockTransport(handler)
+
+
+@pytest.fixture
+def seen_requests():
+    return []
+
+
+@pytest.fixture
+def client(seen_requests, monkeypatch):
+    """SchedulerClient whose real _get_client builds against a MockTransport."""
+    import src.clients.scheduler_client as mod
+
+    real_async_client = httpx.AsyncClient
+
+    def client_factory(**kwargs):
+        kwargs["transport"] = _mock_transport(seen_requests)
+        return real_async_client(**kwargs)
+
+    monkeypatch.setattr(mod.httpx, "AsyncClient", client_factory)
+    return SchedulerClient(base_url="http://scheduler.test:8090", api_key="sched-key")
+
+
+@pytest.mark.unit
+class TestSchedulerApiAuth:
+    @pytest.mark.asyncio
+    async def test_bearer_auth_sent_on_all_calls(self, client, seen_requests):
+        ok = await client.register_volatile_fetch(
+            namespace="weather",
+            key="rotterdam",
+            user="llm_tester",
+            schedule={"minute": 0},
+        )
+
+        assert ok is True
+        assert seen_requests, "no HTTP calls were made"
+        for request in seen_requests:
+            assert request.headers.get("Authorization") == "Bearer sched-key"
+
+    def test_missing_api_key_is_flagged(self, caplog):
+        with caplog.at_level("WARNING"):
+            SchedulerClient(base_url="http://scheduler.test:8090")
+        assert any("without an API key" in r.message for r in caplog.records)
+
+
+@pytest.mark.unit
+class TestRegisterVolatileFetchConfig:
+    @pytest.mark.asyncio
+    async def _registered_config(self, client, seen_requests, user="llm_tester"):
+        ok = await client.register_volatile_fetch(
+            namespace="weather",
+            key="rotterdam",
+            user=user,
+            schedule={"minute": 0, "hour": -1},
+        )
+        assert ok is True
+        create = next(
+            r for r in seen_requests
+            if r.method == "POST" and r.url.path == "/tasks"
+        )
+        import json
+        return json.loads(create.content)["config"]
+
+    @pytest.mark.asyncio
+    async def test_body_key_not_used_payload_is(self, client, seen_requests):
+        config = await self._registered_config(client, seen_requests)
+        # rest_api_executor only reads config["payload"]; "body" is ignored
+        assert "body" not in config
+        assert config["payload"] == {}
+
+    @pytest.mark.asyncio
+    async def test_user_is_query_parameter(self, client, seen_requests):
+        config = await self._registered_config(client, seen_requests)
+        assert config["url"] == (
+            "http://library-desk:8089/volatile/fetch/weather/rotterdam"
+            "?user=llm_tester"
+        )
+
+    @pytest.mark.asyncio
+    async def test_auth_block_uses_placeholder(self, client, seen_requests):
+        config = await self._registered_config(client, seen_requests)
+        assert config["auth"] == {
+            "type": "bearer",
+            "token": LIBRARY_API_KEY_PLACEHOLDER,
+        }
+        # never the raw key, never in plain headers (not substituted there)
+        assert "Authorization" not in config.get("headers", {})