feat(platform): Sprint 3 — HTTP/RAG connectors, control plane, sticky Redis/PG

Add HttpConnector and RagConnector with settings validation, apply PolicyRegistry
constraints at API boundaries, and implement get_pack_version_for_session for Redis
and Postgres run history. Document Sprint 3 and update CHANGELOG [Unreleased].

Author: Brescou

.env.example

@@ -179,6 +179,9 @@ SEARCH_PROVIDER=mock
 # (POST /run, POST /packs/research_analysis/run, and stream variants).
 CONNECTOR_ENABLED=false
 CONNECTOR_ID=example_memory
+# CONNECTOR_ID=http requires CONNECTOR_HTTP_URL (GET with ?q=...&limit=...)
+# CONNECTOR_HTTP_URL=https://your-search-api.example.com/retrieve
+# CONNECTOR_ID=rag requires RAG_ENABLED=true and: uv sync --extra rag
 
 # =============================================================================
 # RAG — Vector Store Integration

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- **Second domain pack** `research_only` (`ResearchOnlyPack`) with typed `/packs/research_only/run` routes.
+- **Retrieval connectors** — `example_memory`, `http` (`CONNECTOR_HTTP_URL`), and `rag` (`RAG_ENABLED`); API injection via `CONNECTOR_ENABLED` into `ResearchAnalysisPack`.
+- **Control plane enforcement** — `PolicyRegistry`, `control_plane/enforce.py` (per-pack query limits, budget ceiling, stream timeout).
+- **Sticky pack versions** on Redis and Postgres run-history backends (`get_pack_version_for_session`).
+
 ## [0.5.0] - 2026-05-04
 
 ### Added

README.md

@@ -79,6 +79,12 @@ User Query
 - **Second domain pack** — `research_only` (`ResearchOnlyPack`) registered alongside `research_analysis`; use `POST /packs/research_only/run` for research-only output.
 - **Optional retrieval connector** — `CONNECTOR_ENABLED` + `CONNECTOR_ID` inject a built-in connector into `ResearchAnalysisPack` on `/run` and `/packs/research_analysis/*` (default id: `example_memory`; query containing `demo` returns canned snippets).
 
+**Platform kernel (Sprint 3)**
+
+- **Connectors** — `http` (`CONNECTOR_HTTP_URL`) and `rag` (`RAG_ENABLED=true`, `CONNECTOR_ID=rag`) in addition to `example_memory`.
+- **Control plane** — `PolicyRegistry` + enforcement at API boundaries (query length, budget, stream timeout per pack).
+- **Sticky sessions** — `get_pack_version_for_session` implemented for **Redis** and **Postgres** run-history backends (SQLite unchanged).
+
 ## Quick Start
 
 **Prerequisites**
@@ -521,7 +527,8 @@ All configuration is loaded from environment variables. Copy `.env.example` to `
 | `API_PORT` | `8000` | TCP port the FastAPI server listens on |
 | `API_KEY` | — | Bearer token for API auth. Leave unset to disable auth. |
 | `CONNECTOR_ENABLED` | `false` | Inject retrieval connector into `research_analysis` runs |
-| `CONNECTOR_ID` | `example_memory` | Built-in connector when enabled (see `core/connectors.py`) |
+| `CONNECTOR_ID` | `example_memory` | `example_memory`, `http`, or `rag` (see `core/connectors.py`) |
+| `CONNECTOR_HTTP_URL` | — | Required when `CONNECTOR_ID=http` |
 
 ### LLM providers
 

api/main.py

@@ -74,6 +74,11 @@
     RunRequest,
     RunResponse,
 )
+from control_plane.enforce import (
+    effective_budget_usd,
+    effective_stream_timeout_seconds,
+    validate_query_for_pack,
+)
 from core.config import Settings, get_settings
 from core.connectors import resolve_connector
 from core.graph import MultiAgentGraph
@@ -347,13 +352,28 @@ def get_shared_memory() -> Any:
     return _shared_memory
 
 
+def _validate_pack_query(pack_id: str, raw_query: str) -> str:
+    """Validate query text using pack policy constraints and global sanitizer."""
+    try:
+        return validate_query_for_pack(raw_query, pack_id, _input_validator)
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=str(exc),
+        ) from exc
+
+
 def _pack_runtime_kwargs(pack_cls: type) -> dict[str, Any]:
-    """Extra constructor kwargs for packs that support an optional connector."""
-    if _shared_connector is None:
-        return {}
-    if getattr(pack_cls, "pack_id", None) != "research_analysis":
-        return {}
-    return {"connector": _shared_connector}
+    """Extra constructor kwargs: policy budget and optional connector."""
+    kwargs: dict[str, Any] = {}
+    pack_id = getattr(pack_cls, "pack_id", None)
+    if pack_id:
+        budget = effective_budget_usd(pack_id, get_settings())
+        if budget is not None:
+            kwargs["budget_usd"] = budget
+    if _shared_connector is not None and pack_id == "research_analysis":
+        kwargs["connector"] = _shared_connector
+    return kwargs
 
 
 def _legacy_pipeline_pack_cls() -> Any:
@@ -515,7 +535,7 @@ async def run_pack(  # type: ignore[misc]
 
         raw_query = body.query if hasattr(body, "query") else str(body)
         try:
-            query = _input_validator.validate(raw_query)
+            query = _validate_pack_query(pack_id, raw_query)
         except AgentValidationError as exc:
             raise HTTPException(
                 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
@@ -534,6 +554,7 @@ def _execute() -> Any:
 
                 # Record run in history with pack_version metadata
                 if _shared_memory is not None:
+                    session_id_for_history = getattr(body, "session_id", None) or None
                     _shared_memory.save_run(
                         run_id=run_id,
                         query=query,
@@ -543,6 +564,11 @@ def _execute() -> Any:
                         metadata={
                             "pack_id": pack_id,
                             "pack_version": used_version,
+                            **(
+                                {"session_id": session_id_for_history}
+                                if session_id_for_history
+                                else {}
+                            ),
                         },
                     )
 
@@ -607,7 +633,7 @@ async def stream_pack(  # type: ignore[misc]
 
         raw_query = body.query if hasattr(body, "query") else str(body)
         try:
-            query = _input_validator.validate(raw_query)
+            query = _validate_pack_query(pack_id, raw_query)
         except AgentValidationError as exc:
             raise HTTPException(
                 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
@@ -628,13 +654,15 @@ async def _event_generator() -> AsyncGenerator[str, None]:
             finally:
                 pack.close()
 
+        stream_timeout = effective_stream_timeout_seconds(pack_id, get_settings())
+
         async def _timed_event_generator() -> AsyncGenerator[str, None]:
             try:
-                async with asyncio.timeout(get_settings().stream_timeout_seconds):
+                async with asyncio.timeout(stream_timeout):
                     async for chunk in _event_generator():
                         yield chunk
             except TimeoutError:
-                yield f"data: {json.dumps({'type': 'error', 'message': f'Stream timed out after {get_settings().stream_timeout_seconds}s'})}\n\n"
+                yield f"data: {json.dumps({'type': 'error', 'message': f'Stream timed out after {stream_timeout}s'})}\n\n"
 
         return StreamingResponse(
             _timed_event_generator(),
@@ -1115,8 +1143,9 @@ async def run_pipeline(
             detail="Server is shutting down.",
         )
 
+    settings = get_settings()
     try:
-        query = _input_validator.validate(body.query)
+        query = _validate_pack_query(settings.default_pack_id, body.query)
     except ValueError as exc:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
@@ -1415,7 +1444,7 @@ async def run_stream(
         )
 
     try:
-        query = _input_validator.validate(body.query)
+        query = _validate_pack_query(settings.default_pack_id, body.query)
     except ValueError as exc:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
@@ -1436,6 +1465,9 @@ async def run_stream(
 
     session_id = body.session_id or str(uuid.uuid4())
     run_id = str(uuid.uuid4())
+    stream_timeout = effective_stream_timeout_seconds(
+        settings.default_pack_id, settings
+    )
 
     logger.info(
         "POST /run/stream — pipeline started",
@@ -1448,11 +1480,11 @@ async def run_stream(
 
     async def _guarded_stream() -> AsyncGenerator[str, None]:
         try:
-            async with asyncio.timeout(settings.stream_timeout_seconds):
+            async with asyncio.timeout(stream_timeout):
                 async for event in _stream_pipeline(query, session_id, run_id):
                     yield event
         except TimeoutError:
-            yield f"data: {json.dumps({'type': 'error', 'message': f'Stream timed out after {settings.stream_timeout_seconds}s'})}\n\n"
+            yield f"data: {json.dumps({'type': 'error', 'message': f'Stream timed out after {stream_timeout}s'})}\n\n"
 
     return StreamingResponse(
         _guarded_stream(),

connectors/http_connector.py

@@ -0,0 +1,91 @@
+"""
+connectors/http_connector.py — HTTP GET connector for JSON or text retrieval APIs.
+
+Expects ``CONNECTOR_HTTP_URL`` (base URL). Appends ``q`` and ``limit`` query params.
+"""
+
+from __future__ import annotations
+
+from typing import Any, ClassVar
+from urllib.parse import urlencode, urlparse, urlunparse
+
+import httpx
+
+from connectors.base import BaseConnector, ConnectorRequest, ConnectorResult
+
+
+class HttpConnector(BaseConnector):
+    """Fetches retrieval snippets from a configurable HTTP endpoint."""
+
+    connector_id: ClassVar[str] = "http"
+    name: ClassVar[str] = "HTTP retrieval connector"
+    description: ClassVar[str] = (
+        "GET request to CONNECTOR_HTTP_URL with query/limit parameters."
+    )
+
+    def __init__(self, base_url: str, timeout_seconds: float = 10.0) -> None:
+        if not base_url or not base_url.strip():
+            raise ValueError("HttpConnector requires a non-empty base_url")
+        self._base_url = base_url.strip().rstrip("/")
+        self._timeout = timeout_seconds
+
+    async def fetch(self, request: ConnectorRequest) -> ConnectorResult:
+        params = {
+            "q": request.query,
+            "limit": str(request.limit),
+            **{k: str(v) for k, v in request.filters.items()},
+        }
+        url = self._build_url(params)
+
+        async with httpx.AsyncClient(timeout=self._timeout) as client:
+            response = await client.get(url)
+            response.raise_for_status()
+
+        content_type = response.headers.get("content-type", "")
+        if "application/json" in content_type:
+            records = _parse_json_payload(response.json())
+        else:
+            records = _parse_text_payload(response.text)
+
+        return ConnectorResult(
+            records=tuple(records[: request.limit]),
+            metadata={"url": url, "status_code": response.status_code},
+        )
+
+    def _build_url(self, params: dict[str, str]) -> str:
+        parsed = urlparse(self._base_url)
+        query = urlencode(params)
+        if parsed.query:
+            query = f"{parsed.query}&{query}"
+        return urlunparse(parsed._replace(query=query))
+
+
+def _parse_json_payload(payload: Any) -> list[dict[str, Any]]:
+    if isinstance(payload, list):
+        return [_normalize_record(item) for item in payload if isinstance(item, dict)]
+    if isinstance(payload, dict):
+        for key in ("results", "records", "items", "data"):
+            items = payload.get(key)
+            if isinstance(items, list):
+                return [
+                    _normalize_record(item) for item in items if isinstance(item, dict)
+                ]
+        if "snippet" in payload or "text" in payload:
+            return [_normalize_record(payload)]
+    return []
+
+
+def _parse_text_payload(text: str) -> list[dict[str, Any]]:
+    lines = [line.strip() for line in text.splitlines() if line.strip()]
+    return [{"source": "http", "snippet": line} for line in lines]
+
+
+def _normalize_record(item: dict[str, Any]) -> dict[str, Any]:
+    snippet = item.get("snippet") or item.get("text") or item.get("content")
+    if snippet is None and len(item) == 1:
+        snippet = next(iter(item.values()))
+    record = dict(item)
+    if snippet is not None:
+        record.setdefault("snippet", str(snippet))
+    record.setdefault("source", str(record.get("source", "http")))
+    return record

connectors/rag_connector.py

@@ -0,0 +1,47 @@
+"""
+connectors/rag_connector.py — Vector-store retrieval via core.vectorstore.
+
+Requires ``RAG_ENABLED=true`` and the ``rag`` optional dependencies.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, ClassVar
+
+from connectors.base import BaseConnector, ConnectorRequest, ConnectorResult
+
+if TYPE_CHECKING:
+    from core.config import Settings
+
+
+class RagConnector(BaseConnector):
+    """Runs similarity search against the configured RAG vector store."""
+
+    connector_id: ClassVar[str] = "rag"
+    name: ClassVar[str] = "RAG vector store connector"
+    description: ClassVar[str] = (
+        "Similarity search via get_vectorstore() when RAG_ENABLED=true."
+    )
+
+    def __init__(self, settings: Settings) -> None:
+        self._settings = settings
+
+    async def fetch(self, request: ConnectorRequest) -> ConnectorResult:
+        from core.vectorstore import get_vectorstore
+
+        store = get_vectorstore(self._settings)
+        documents = store.similarity_search(request.query, k=request.limit)
+        records: list[dict[str, Any]] = []
+        for index, doc in enumerate(documents):
+            meta = dict(doc.metadata) if doc.metadata else {}
+            records.append(
+                {
+                    "source": str(meta.get("source", f"rag:{index}")),
+                    "snippet": doc.page_content,
+                    "score": meta.get("score"),
+                }
+            )
+        return ConnectorResult(
+            records=tuple(records),
+            metadata={"backend": "rag", "count": len(records)},
+        )

control_plane/README.md

@@ -1,33 +1,30 @@
-# Control plane (foundation — Sprint 2)
+# Control plane
 
-This folder holds **types and documentation only**. There is no evaluator, no tenant model, no feature-flag service, and no integration with FastAPI or `PackRegistry` beyond naming compatibility (`pack_id` aligns with registered packs).
+Pack-level policies are registered explicitly in `control_plane/__init__.py` (Approach B, same spirit as `PackRegistry`).
 
-## What the control plane will own (directional)
+## What is enforced today
 
-| Area | Intent (future) |
-|------|-----------------|
-| **Pack-level policies** | Named bundles keyed by `pack_id` — caps, labels, extension metadata. |
-| **Execution constraints** | Advisory limits (`ExecutionConstraints`) that orchestration or packs may honour. |
-| **Governance hooks** | Labels and `extensions` dict reserved for audit, cost centres, or routing hints — **not interpreted here**. |
-| **Feature flags / policy engine** | Explicitly **out of scope** for this skeleton; strings like `labels` are placeholders only. |
+| Mechanism | Where |
+|-----------|--------|
+| `max_query_chars` | `validate_query_for_pack()` — used by API before pack and legacy runs |
+| `budget_usd_ceiling` | `effective_budget_usd()` — passed as `budget_usd` when constructing packs (overridden by `PACK_DEFAULT_BUDGET_USD`) |
+| `stream_timeout_seconds` | `effective_stream_timeout_seconds()` — caps SSE timeouts per pack route and legacy `/run/stream` |
 
-## What this is **not**
+## Registry
 
-- Not a dynamic policy DSL or OPA integration.
-- Not multi-tenant isolation or quota enforcement.
-- Not a replacement for `DEFAULT_PACK_ID` or `PackRegistry.register()`.
+```python
+from control_plane import PolicyRegistry, PackPolicy, ExecutionConstraints
 
-## Compatibility with `PackRegistry`
+PolicyRegistry.register(
+    PackPolicy(
+        pack_id="my_pack",
+        constraints=ExecutionConstraints(max_query_chars=1500, budget_usd_ceiling=0.50),
+    )
+)
+```
 
-Policies use the same **`pack_id`** strings as `platform.registry.PackRegistry`. Registration remains **explicit and static** in `platform/__init__.py`. This package does **not** register packs or duplicate the registry.
+## What is still not here
 
-## Files
-
-| File | Role |
-|------|------|
-| `policies.py` | `ExecutionConstraints`, `PackPolicy` dataclasses. |
-| `__init__.py` | Public exports for imports: `from control_plane import PackPolicy`. |
-
-## Minimalism
-
-Two frozen dataclasses avoid inventing interfaces before call sites exist. Enforcement belongs in API middleware or pack code in a later sprint.
+- Dynamic policy DSL / OPA
+- Multi-tenant quotas
+- Automatic pack registration from policies (policies reference existing `pack_id` values only)