diff --git a/README.md b/README.md
index 90024fc..a28cba5 100755
--- a/README.md
+++ b/README.md
@@ -659,6 +659,72 @@ class LMEvalAdapter(FrameworkAdapter):
         return JobResults(...)
 ```
 
+## Live Endpoint Response Collection (Experimental)
+
+The SDK includes a collector utility that adapters can call during `LOADING_DATA`
+to query a chatbot endpoint with test questions and collect responses for evaluation.
+
+### OpenAI-Compatible Endpoint
+
+```python
+from evalhub.adapter import (
+    CollectorConfig, collect_responses, is_collection_configured,
+    resolve_model_credentials,
+)
+
+# In your adapter's run_benchmark_job():
+if is_collection_configured(config.parameters):
+    creds = resolve_model_credentials()
+    collector_config = CollectorConfig.from_parameters(config.parameters)
+    manifest = collect_responses(collector_config, credentials=creds)
+    # manifest.output_path -> responses.jsonl for your evaluation framework
+```
+
+Job parameters:
+```json
+{
+  "live_collection": {
+    "questions_path": "/test_data/questions.csv",
+    "output_dir": "/tmp/collected",
+    "endpoint_url": "https://my-chatbot.example/v1",
+    "model": "my-chatbot-v2",
+    "protocol": "openai_chat_completions"
+  }
+}
+```
+
+### Generic HTTP Endpoint (MCP, Langflow, custom APIs)
+
+```json
+{
+  "live_collection": {
+    "questions_path": "/test_data/questions.csv",
+    "output_dir": "/tmp/collected",
+    "endpoint_url": "https://mcp-chatbot.example/mcp",
+    "protocol": "generic_http",
+    "request_template": {
+      "jsonrpc": "2.0",
+      "method": "tools/call",
+      "params": {"name": "chat", "arguments": {"message": "{question}"}},
+      "id": "{question_id}"
+    },
+    "response_path": "result.content.0.text",
+    "extra_response_paths": {
+      "retrieved_contexts": "result.sources"
+    }
+  }
+}
+```
+
+The collector uses SDK TLS auto-detection by default. In Kubernetes, the model
+secret contains a sidecar reference token and cannot authenticate a direct request
+to the live endpoint. Configure `api_key_env` or `request_headers` with the live
+endpoint's actual credential instead; the collector will not send the sidecar
+reference token directly to the endpoint.
+
+See [`src/evalhub/adapter/collector.py`](src/evalhub/adapter/collector.py) for the
+complete collector configuration reference.
+
 ## Deployment
 
 ### Container Structure
diff --git a/src/evalhub/adapter/__init__.py b/src/evalhub/adapter/__init__.py
index 0198816..f0872b0 100644
--- a/src/evalhub/adapter/__init__.py
+++ b/src/evalhub/adapter/__init__.py
@@ -79,6 +79,22 @@ def run_benchmark_job(
 )
 from .auth import ModelCredentials, read_model_auth_key, resolve_model_credentials
 from .callbacks import DefaultCallbacks
+from .collector import (
+    CollectedRecord,
+    CollectionManifest,
+    CollectorConfig,
+    CollectorError,
+    CollectorProtocol,
+    LiveQuestion,
+    collect_responses,
+    collect_responses_from_parameters,
+    extract_by_path,
+    load_questions,
+    substitute_template,
+)
+from .collector import (
+    is_configured as is_collection_configured,
+)
 from .config import MlflowBackend, get_job_spec_path
 from .models import (
     CapabilityEvalEntry,
@@ -142,4 +158,17 @@ def run_benchmark_job(
     "ModelConfig",
     "EvaluationResult",
     "OCICoordinates",
+    # Live endpoint collection (experimental)
+    "CollectorConfig",
+    "CollectorProtocol",
+    "CollectorError",
+    "CollectedRecord",
+    "CollectionManifest",
+    "LiveQuestion",
+    "collect_responses",
+    "collect_responses_from_parameters",
+    "is_collection_configured",
+    "load_questions",
+    "extract_by_path",
+    "substitute_template",
 ]
diff --git a/src/evalhub/adapter/collector.py b/src/evalhub/adapter/collector.py
new file mode 100644
index 0000000..f38d00c
--- /dev/null
+++ b/src/evalhub/adapter/collector.py
@@ -0,0 +1,646 @@
+"""Experimental live endpoint response collection for adapters.
+
+Lets adapters collect chatbot/model responses during ``JobPhase.LOADING_DATA``
+by querying a live endpoint with a set of test questions. Supports
+OpenAI-compatible chat completions and generic HTTP endpoints (MCP, Langflow,
+LangGraph, custom APIs).
+
+Config is passed through ``JobSpec.parameters["live_collection"]`` while the
+server-side API shape (``test_data_ref``) is still being validated.
+
+Trust boundary: the adapter operator controls the endpoint configuration.
+Hosted runtimes that accept config from untrusted users should add their own
+egress policy or endpoint allowlist before enabling this.
+"""
+
+from __future__ import annotations
+
+import csv
+import json
+import logging
+import os
+import time
+from collections.abc import Callable, Mapping
+from datetime import UTC, datetime
+from enum import StrEnum
+from pathlib import Path
+from typing import Any, Self, cast
+
+from pydantic import BaseModel, Field, field_validator, model_validator
+
+from .auth import ModelCredentials, resolve_model_credentials
+from .config import EvalHubMode
+from .settings import AdapterSettings
+
+logger = logging.getLogger(__name__)
+
+_PARAM_KEY = "live_collection"
+
+_CA_BUNDLE_CANDIDATES = [
+    Path("/etc/pki/ca-trust/source/anchors/service-ca.crt"),
+    Path("/var/run/secrets/kubernetes.io/serviceaccount/service-ca.crt"),
+    Path("/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"),
+]
+
+
+class CollectorProtocol(StrEnum):
+    OPENAI_CHAT = "openai_chat_completions"
+    GENERIC_HTTP = "generic_http"
+
+
+class CollectorError(Exception):
+    """Raised when collection fails in fail_fast mode."""
+
+    def __init__(
+        self,
+        message: str,
+        question_id: str | None = None,
+        cause: Exception | None = None,
+    ) -> None:
+        self.question_id = question_id
+        self.__cause__ = cause
+        super().__init__(message)
+
+
+class CollectorConfig(BaseModel):
+    """Configuration for live endpoint response collection."""
+
+    questions_path: Path = Field(
+        ..., description="CSV, JSON, or JSONL file with test questions"
+    )
+    output_dir: Path = Field(
+        ..., description="Directory for responses.jsonl and manifest.json"
+    )
+    endpoint_url: str = Field(
+        ..., description="Endpoint URL (OpenAI-compatible base or full HTTP URL)"
+    )
+    protocol: CollectorProtocol = Field(default=CollectorProtocol.OPENAI_CHAT)
+
+    question_column: str = Field(default="question")
+    id_column: str = Field(default="id")
+
+    model: str | None = Field(
+        default=None, description="Model name for OpenAI chat requests"
+    )
+    system_prompt: str | None = Field(default=None)
+    extra_body: dict[str, Any] = Field(default_factory=dict)
+
+    request_template: dict[str, Any] | None = Field(
+        default=None, description="Request body template for generic_http protocol"
+    )
+    response_path: str | None = Field(
+        default=None,
+        description="Dot-separated path to extract answer from response",
+    )
+    extra_response_paths: dict[str, str] | None = Field(
+        default=None,
+        description="Additional fields to extract from response (output_name -> dot-path)",
+    )
+
+    api_key_env: str | None = Field(
+        default=None, description="Env var containing a bearer token"
+    )
+    request_headers: dict[str, str] = Field(default_factory=dict)
+    use_model_credentials: bool = Field(
+        default=True,
+        description="Use resolve_model_credentials() for auth and TLS",
+    )
+
+    ca_bundle: str | None = Field(default=None, description="CA bundle path")
+    insecure: bool = Field(default=False, description="Skip TLS verification")
+
+    fail_fast: bool = Field(
+        default=False, description="Raise on first collection error"
+    )
+    timeout_seconds: float = Field(default=30.0, gt=0.0)
+    max_retries: int = Field(default=0, ge=0)
+    retry_backoff_seconds: float = Field(default=0.25, ge=0.0)
+    max_retry_backoff_seconds: float = Field(default=5.0, ge=0.0)
+
+    @classmethod
+    def from_parameters(cls, parameters: Mapping[str, Any]) -> CollectorConfig:
+        raw = parameters.get(_PARAM_KEY)
+        if raw is None:
+            raise ValueError(f'JobSpec.parameters["{_PARAM_KEY}"] is required')
+        if not isinstance(raw, dict):
+            raise ValueError(f'parameters["{_PARAM_KEY}"] must be an object')
+        return cls.model_validate(raw)
+
+    @field_validator("endpoint_url")
+    @classmethod
+    def _validate_endpoint_url(cls, value: str) -> str:
+        if not value.lower().startswith("https://"):
+            raise ValueError("endpoint_url must use https://")
+        return value.rstrip("/")
+
+    @model_validator(mode="after")
+    def _validate_protocol_requirements(self) -> Self:
+        if self.protocol == CollectorProtocol.OPENAI_CHAT and not self.model:
+            raise ValueError("model is required for openai_chat_completions protocol")
+        if self.protocol == CollectorProtocol.GENERIC_HTTP:
+            if not self.request_template:
+                raise ValueError(
+                    "request_template is required for generic_http protocol"
+                )
+            if not self.response_path:
+                raise ValueError("response_path is required for generic_http protocol")
+        return self
+
+
+class LiveQuestion(BaseModel):
+    question: str
+    question_id: str | None = None
+    source_row: dict[str, Any] = Field(default_factory=dict)
+
+
+class CollectedRecord(BaseModel):
+    """One collected response row.
+
+    ``source_fields`` preserves all original input columns as-is.
+    Collector-added fields (``response``, ``error``, etc.) are merged
+    on top when serialized to JSONL.
+    """
+
+    source_fields: dict[str, Any] = Field(default_factory=dict)
+    response: str | None = None
+    extra_fields: dict[str, Any] = Field(default_factory=dict)
+    raw_response: dict[str, Any] | None = None
+    error: str | None = None
+    latency_ms: float | None = None
+    error_cause: Any = Field(default=None, exclude=True)
+
+
+class CollectionManifest(BaseModel):
+    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
+    experimental: bool = True
+    protocol: str
+    endpoint_url: str
+    model: str | None = None
+    questions_path: str
+    output_path: str
+    total: int
+    completed: int
+    failed: int
+
+
+def is_configured(parameters: dict[str, Any] | None) -> bool:
+    """Check if parameters contain live collection config."""
+    if not parameters:
+        return False
+    raw = parameters.get(_PARAM_KEY)
+    return isinstance(raw, dict) and bool(raw)
+
+
+def collect_responses(
+    config: CollectorConfig,
+    *,
+    client: Any | None = None,
+    credentials: ModelCredentials | None = None,
+    progress_callback: Callable[[float], None] | None = None,
+) -> CollectionManifest:
+    """Collect responses from a live endpoint.
+
+    Reads test questions, queries the endpoint, and writes an evaluation-ready
+    JSONL dataset plus a manifest summary.
+    """
+    if credentials is None and config.use_model_credentials:
+        try:
+            credentials = resolve_model_credentials()
+        except Exception:
+            logger.debug("Could not resolve model credentials", exc_info=True)
+            credentials = None
+
+    headers = _resolve_auth_headers(config, credentials)
+    verify = _resolve_verify(config, credentials)
+    questions = load_questions(
+        config.questions_path,
+        question_column=config.question_column,
+        id_column=config.id_column,
+    )
+
+    config.output_dir.mkdir(parents=True, exist_ok=True)
+    output_path = config.output_dir / "responses.jsonl"
+
+    active_client = client
+    owns_client = active_client is None
+    if active_client is None:
+        try:
+            import httpx
+        except ImportError as exc:
+            raise RuntimeError(
+                "Live collection requires httpx. "
+                "Install eval-hub-sdk[adapter] or eval-hub-sdk[core]."
+            ) from exc
+        active_client = httpx.Client(
+            follow_redirects=False,
+            timeout=config.timeout_seconds,
+            verify=verify,
+        )
+
+    completed = 0
+    failed = 0
+    total = len(questions)
+
+    try:
+        with output_path.open("w", encoding="utf-8") as out:
+            for i, question in enumerate(questions):
+                record = _collect_one(config, active_client, headers, question)
+                if record.error:
+                    failed += 1
+                    if config.fail_fast:
+                        raise CollectorError(
+                            record.error,
+                            question_id=question.question_id,
+                            cause=record.error_cause,
+                        )
+                else:
+                    completed += 1
+                out.write(json.dumps(_flatten_record(record)) + "\n")
+                if progress_callback and total > 0:
+                    progress_callback((i + 1) / total)
+    finally:
+        if owns_client and hasattr(active_client, "close"):
+            active_client.close()
+
+        # Preserve a partial manifest when fail_fast aborts collection.
+        manifest = CollectionManifest(
+            protocol=config.protocol.value,
+            endpoint_url=config.endpoint_url,
+            model=config.model,
+            questions_path=str(config.questions_path),
+            output_path=str(output_path),
+            total=total,
+            completed=completed,
+            failed=failed,
+        )
+        manifest_path = config.output_dir / "manifest.json"
+        manifest_path.write_text(
+            json.dumps(manifest.model_dump(mode="json"), indent=2, default=str) + "\n",
+            encoding="utf-8",
+        )
+    return manifest
+
+
+def collect_responses_from_parameters(
+    parameters: Mapping[str, Any],
+    **kwargs: Any,
+) -> CollectionManifest:
+    """Convenience wrapper: load config from parameters and collect."""
+    config = CollectorConfig.from_parameters(parameters)
+    return collect_responses(config, **kwargs)
+
+
+def load_questions(
+    path: Path | str,
+    *,
+    question_column: str = "question",
+    id_column: str = "id",
+) -> list[LiveQuestion]:
+    """Load questions from CSV, JSON, or JSONL."""
+    qpath = Path(path)
+    if not qpath.exists():
+        raise FileNotFoundError(f"Question file not found: {qpath}")
+
+    suffix = qpath.suffix.lower()
+    if suffix == ".csv":
+        questions = _load_csv(qpath, question_column, id_column)
+    elif suffix == ".jsonl":
+        questions = _load_jsonl(qpath, question_column, id_column)
+    elif suffix == ".json":
+        questions = _load_json(qpath, question_column, id_column)
+    else:
+        raise ValueError(
+            f"Unsupported question file type '{suffix}'. Use CSV, JSON, or JSONL."
+        )
+
+    if not questions:
+        raise ValueError(f"No questions found in {qpath}")
+    return questions
+
+
+def extract_by_path(data: Any, path: str) -> Any:
+    """Extract a value from nested data using a dot-separated path."""
+    current = data
+    for segment in path.split("."):
+        if isinstance(current, dict):
+            current = current.get(segment)
+        elif isinstance(current, list):
+            try:
+                current = current[int(segment)]
+            except (ValueError, IndexError):
+                return None
+        else:
+            return None
+        if current is None:
+            return None
+    return current
+
+
+def substitute_template(template: Any, variables: dict[str, str]) -> Any:
+    """Recursively substitute ``{key}`` placeholders in string values."""
+    if isinstance(template, str):
+        try:
+            return template.format_map(variables)
+        except KeyError as exc:
+            raise ValueError(
+                f"Unknown placeholder {exc} in template. "
+                f"Available: {', '.join(variables.keys())}"
+            ) from exc
+    if isinstance(template, dict):
+        return {k: substitute_template(v, variables) for k, v in template.items()}
+    if isinstance(template, list):
+        return [substitute_template(item, variables) for item in template]
+    return template
+
+
+def _resolve_auth_headers(
+    config: CollectorConfig,
+    credentials: ModelCredentials | None,
+) -> dict[str, str]:
+    headers: dict[str, str] = {}
+    if credentials and credentials.api_key:
+        if AdapterSettings.from_env().mode == EvalHubMode.K8S:
+            logger.warning(
+                "Ignoring Kubernetes model credential ref token for direct live "
+                "endpoint access; configure api_key_env or request_headers with "
+                "the endpoint credential"
+            )
+        else:
+            headers["Authorization"] = f"Bearer {credentials.api_key}"
+    if config.api_key_env:
+        api_key = os.getenv(config.api_key_env)
+        if not api_key:
+            raise ValueError(f"Environment variable {config.api_key_env!r} is required")
+        headers["Authorization"] = f"Bearer {api_key}"
+    headers.update(config.request_headers)
+    return headers
+
+
+def _resolve_verify(
+    config: CollectorConfig,
+    credentials: ModelCredentials | None,
+) -> bool | str:
+    if config.insecure:
+        return False
+    if config.ca_bundle:
+        return config.ca_bundle
+    if credentials and credentials.ca_cert_path:
+        return str(credentials.ca_cert_path)
+    for candidate in _CA_BUNDLE_CANDIDATES:
+        if candidate.exists():
+            return str(candidate)
+    return True
+
+
+def _collect_one(
+    config: CollectorConfig,
+    client: Any,
+    headers: dict[str, str],
+    question: LiveQuestion,
+) -> CollectedRecord:
+    if config.protocol == CollectorProtocol.OPENAI_CHAT:
+        return _collect_openai(config, client, headers, question)
+    return _collect_generic_http(config, client, headers, question)
+
+
+def _collect_openai(
+    config: CollectorConfig,
+    client: Any,
+    headers: dict[str, str],
+    question: LiveQuestion,
+) -> CollectedRecord:
+    messages: list[dict[str, str]] = []
+    if config.system_prompt:
+        messages.append({"role": "system", "content": config.system_prompt})
+    messages.append({"role": "user", "content": question.question})
+
+    body: dict[str, Any] = dict(config.extra_body)
+    body["model"] = config.model
+    body["messages"] = messages
+
+    url = f"{config.endpoint_url}/chat/completions"
+    return _send_request(config, client, headers, question, url, body, _extract_openai)
+
+
+def _collect_generic_http(
+    config: CollectorConfig,
+    client: Any,
+    headers: dict[str, str],
+    question: LiveQuestion,
+) -> CollectedRecord:
+    request_template = config.request_template
+    response_path = config.response_path
+    if not request_template or not response_path:
+        raise ValueError(
+            "request_template and response_path are required for generic_http"
+        )
+
+    variables = {
+        "question": question.question,
+        "question_id": question.question_id or "",
+    }
+    body = substitute_template(request_template, variables)
+    return _send_request(
+        config,
+        client,
+        headers,
+        question,
+        config.endpoint_url,
+        body,
+        lambda resp: _extract_generic(resp, response_path),
+    )
+
+
+def _send_request(
+    config: CollectorConfig,
+    client: Any,
+    headers: dict[str, str],
+    question: LiveQuestion,
+    url: str,
+    body: dict[str, Any],
+    extractor: Callable[[dict[str, Any]], str | None],
+) -> CollectedRecord:
+    last_error: str | None = None
+    last_cause: Exception | None = None
+    for attempt in range(config.max_retries + 1):
+        start = time.monotonic()
+        try:
+            response = client.post(
+                url,
+                json=body,
+                headers=dict(headers),
+                timeout=config.timeout_seconds,
+            )
+            latency_ms = (time.monotonic() - start) * 1000
+
+            status_code = getattr(response, "status_code", None)
+            if isinstance(status_code, int) and 300 <= status_code < 400:
+                raise ValueError(f"Redirect not allowed: HTTP {status_code}")
+
+            response.raise_for_status()
+            raw_response = response.json()
+            if not isinstance(raw_response, dict):
+                raise ValueError("Response must be a JSON object")
+
+            text = extractor(cast(dict[str, Any], raw_response))
+            extra_fields = _extract_extra_fields(
+                cast(dict[str, Any], raw_response), config.extra_response_paths
+            )
+            return CollectedRecord(
+                source_fields=question.source_row,
+                response=text,
+                extra_fields=extra_fields,
+                raw_response=cast(dict[str, Any], raw_response),
+                latency_ms=latency_ms,
+            )
+        except Exception as exc:
+            last_cause = exc
+            last_error = f"{type(exc).__name__}: {exc}"
+            status_code = getattr(getattr(exc, "response", None), "status_code", None)
+            if isinstance(exc, ValueError) or (
+                isinstance(status_code, int)
+                and 400 <= status_code < 500
+                and status_code != 429
+            ):
+                break
+            if attempt < config.max_retries and config.retry_backoff_seconds > 0:
+                delay = min(
+                    config.retry_backoff_seconds * (2**attempt),
+                    config.max_retry_backoff_seconds,
+                )
+                time.sleep(delay)
+
+    return CollectedRecord(
+        source_fields=question.source_row,
+        error=last_error,
+        error_cause=last_cause,
+    )
+
+
+_RESERVED_FIELDS = {"response", "raw_response", "error", "latency_ms"}
+
+
+def _flatten_record(record: CollectedRecord) -> dict[str, Any]:
+    """Flatten a record for JSONL output.
+
+    Starts with the original input columns (``source_fields``), then adds
+    collector-produced fields (``response``, ``error``, etc.) and any
+    ``extra_fields`` on top.
+    """
+    data = dict(record.source_fields)
+    collisions = _RESERVED_FIELDS & record.source_fields.keys()
+    if collisions:
+        logger.warning(
+            "Source fields %s will be overwritten by collector output", collisions
+        )
+    if record.extra_fields:
+        data.update(record.extra_fields)
+    # Extra response paths are user-configurable, so reapply collector-owned
+    # fields after merging to prevent them from overwriting collected values.
+    data["response"] = record.response
+    data["raw_response"] = record.raw_response
+    data["error"] = record.error
+    data["latency_ms"] = record.latency_ms
+    return data
+
+
+def _extract_extra_fields(
+    raw_response: dict[str, Any],
+    extra_paths: dict[str, str] | None,
+) -> dict[str, Any]:
+    if not extra_paths:
+        return {}
+    result: dict[str, Any] = {}
+    for output_name, path in extra_paths.items():
+        result[output_name] = extract_by_path(raw_response, path)
+    return result
+
+
+def _extract_openai(raw_response: dict[str, Any]) -> str | None:
+    value = extract_by_path(raw_response, "choices.0.message.content")
+    if value is not None:
+        return str(value)
+    if extract_by_path(raw_response, "choices.0.message.tool_calls"):
+        raise ValueError(
+            "OpenAI response contains tool_calls but no message.content; "
+            "configure generic_http or a custom extractor for tool responses"
+        )
+    raise ValueError("OpenAI response is missing choices.0.message.content")
+
+
+def _extract_generic(raw_response: dict[str, Any], response_path: str) -> str | None:
+    value = extract_by_path(raw_response, response_path)
+    if value is None:
+        return None
+    return str(value) if not isinstance(value, str) else value
+
+
+def _load_csv(path: Path, question_column: str, id_column: str) -> list[LiveQuestion]:
+    with path.open(newline="", encoding="utf-8") as f:
+        reader = csv.DictReader(f)
+        if reader.fieldnames is None or question_column not in reader.fieldnames:
+            raise ValueError(f"CSV must include a '{question_column}' column")
+
+        questions: list[LiveQuestion] = []
+        for i, row in enumerate(reader, start=1):
+            text = (row.get(question_column) or "").strip()
+            if not text:
+                continue
+            qid = (row.get(id_column) or "").strip() or str(i)
+            questions.append(
+                LiveQuestion(question=text, question_id=qid, source_row=dict(row))
+            )
+        return questions
+
+
+def _load_jsonl(path: Path, question_column: str, id_column: str) -> list[LiveQuestion]:
+    questions: list[LiveQuestion] = []
+    with path.open(encoding="utf-8") as f:
+        for i, line in enumerate(f, start=1):
+            if not line.strip():
+                continue
+            raw = json.loads(line)
+            if not isinstance(raw, dict):
+                raise ValueError(f"JSONL row {i} must be an object")
+            q = _question_from_row(raw, question_column, id_column, i)
+            if q:
+                questions.append(q)
+    return questions
+
+
+def _load_json(path: Path, question_column: str, id_column: str) -> list[LiveQuestion]:
+    raw_data = json.loads(path.read_text(encoding="utf-8"))
+    if isinstance(raw_data, dict) and "questions" in raw_data:
+        raw_data = raw_data["questions"]
+    if not isinstance(raw_data, list):
+        raise ValueError("JSON input must be a list or contain a 'questions' list")
+
+    questions: list[LiveQuestion] = []
+    for i, raw in enumerate(raw_data, start=1):
+        if not isinstance(raw, dict):
+            raise ValueError(f"JSON row {i} must be an object")
+        q = _question_from_row(raw, question_column, id_column, i)
+        if q:
+            questions.append(q)
+    return questions
+
+
+def _question_from_row(
+    row: dict[str, Any],
+    question_column: str,
+    id_column: str,
+    row_index: int,
+) -> LiveQuestion | None:
+    raw_q = row.get(question_column)
+    if raw_q is None:
+        raise ValueError(f"Row {row_index} must include '{question_column}'")
+    if not isinstance(raw_q, str):
+        raise ValueError(f"Row {row_index} '{question_column}' must be a string")
+    text = raw_q.strip()
+    if not text:
+        return None
+
+    raw_id = row.get(id_column)
+    qid = str(raw_id).strip() if raw_id is not None else str(row_index)
+
+    return LiveQuestion(question=text, question_id=qid, source_row=dict(row))
diff --git a/tests/unit/test_collector.py b/tests/unit/test_collector.py
new file mode 100644
index 0000000..0d0c47f
--- /dev/null
+++ b/tests/unit/test_collector.py
@@ -0,0 +1,878 @@
+"""Unit tests for the live endpoint response collector."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+from unittest.mock import MagicMock
+
+import httpx
+import pytest
+from evalhub.adapter.auth import ModelCredentials
+from evalhub.adapter.collector import (
+    CollectedRecord,
+    CollectorConfig,
+    CollectorError,
+    CollectorProtocol,
+    _extract_openai,
+    _flatten_record,
+    _resolve_auth_headers,
+    collect_responses,
+    collect_responses_from_parameters,
+    extract_by_path,
+    is_configured,
+    load_questions,
+    substitute_template,
+)
+
+pytestmark = pytest.mark.unit
+
+
+def _response(status_code: int, payload: dict[str, Any]) -> httpx.Response:
+    return httpx.Response(
+        status_code,
+        json=payload,
+        request=httpx.Request("POST", "https://endpoint.example/v1/chat/completions"),
+    )
+
+
+def _openai_ok(content: str = "test answer") -> httpx.Response:
+    return _response(
+        200,
+        {
+            "choices": [{"message": {"content": content}, "finish_reason": "stop"}],
+            "usage": {"total_tokens": 10},
+        },
+    )
+
+
+# -- is_configured -----------------------------------------------------------
+
+
+class TestIsConfigured:
+    def test_configured_with_valid_dict(self) -> None:
+        assert is_configured({"live_collection": {"endpoint_url": "http://x"}})
+
+    def test_not_configured_empty_params(self) -> None:
+        assert not is_configured({})
+
+    def test_not_configured_none(self) -> None:
+        assert not is_configured(None)
+
+    def test_not_configured_empty_dict(self) -> None:
+        assert not is_configured({"live_collection": {}})
+
+    def test_not_configured_non_dict(self) -> None:
+        assert not is_configured({"live_collection": "string"})
+
+
+# -- CollectorConfig validation -----------------------------------------------
+
+
+class TestCollectorConfigValidation:
+    def test_rejects_non_http_endpoint(self, tmp_path: Path) -> None:
+        with pytest.raises(ValueError, match="https://"):
+            CollectorConfig(
+                questions_path=tmp_path / "q.csv",
+                output_dir=tmp_path / "out",
+                endpoint_url="file:///etc/passwd",
+                model="chatbot",
+            )
+
+    def test_rejects_http_endpoint(self, tmp_path: Path) -> None:
+        with pytest.raises(ValueError, match="https://"):
+            CollectorConfig(
+                questions_path=tmp_path / "q.csv",
+                output_dir=tmp_path / "out",
+                endpoint_url="http://api.example/v1",
+                model="chatbot",
+            )
+
+    def test_openai_requires_model(self, tmp_path: Path) -> None:
+        with pytest.raises(ValueError, match="model is required"):
+            CollectorConfig(
+                questions_path=tmp_path / "q.csv",
+                output_dir=tmp_path / "out",
+                endpoint_url="https://api.example/v1",
+                protocol=CollectorProtocol.OPENAI_CHAT,
+            )
+
+    def test_generic_http_requires_template(self, tmp_path: Path) -> None:
+        with pytest.raises(ValueError, match="request_template is required"):
+            CollectorConfig(
+                questions_path=tmp_path / "q.csv",
+                output_dir=tmp_path / "out",
+                endpoint_url="https://api.example",
+                protocol=CollectorProtocol.GENERIC_HTTP,
+                response_path="result.text",
+            )
+
+    def test_generic_http_requires_response_path(self, tmp_path: Path) -> None:
+        with pytest.raises(ValueError, match="response_path is required"):
+            CollectorConfig(
+                questions_path=tmp_path / "q.csv",
+                output_dir=tmp_path / "out",
+                endpoint_url="https://api.example",
+                protocol=CollectorProtocol.GENERIC_HTTP,
+                request_template={"input": "{question}"},
+            )
+
+    def test_from_parameters_loads_config(self, tmp_path: Path) -> None:
+        config = CollectorConfig.from_parameters(
+            {
+                "live_collection": {
+                    "questions_path": str(tmp_path / "q.csv"),
+                    "output_dir": str(tmp_path / "out"),
+                    "endpoint_url": "https://api.example/v1",
+                    "model": "chatbot",
+                }
+            }
+        )
+        assert config.model == "chatbot"
+        assert config.protocol == CollectorProtocol.OPENAI_CHAT
+
+    def test_kubernetes_ref_token_is_not_sent_to_endpoint(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setenv("EVALHUB_MODE", "k8s")
+        config = CollectorConfig(
+            questions_path=tmp_path / "q.csv",
+            output_dir=tmp_path / "out",
+            endpoint_url="https://api.example/v1",
+            model="chatbot",
+        )
+
+        headers = _resolve_auth_headers(config, ModelCredentials(api_key="api-key:ref"))
+
+        assert "Authorization" not in headers
+
+    def test_kubernetes_explicit_endpoint_key_is_allowed(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setenv("EVALHUB_MODE", "k8s")
+        monkeypatch.setenv("LIVE_ENDPOINT_KEY", "real-key")
+        config = CollectorConfig(
+            questions_path=tmp_path / "q.csv",
+            output_dir=tmp_path / "out",
+            endpoint_url="https://api.example/v1",
+            model="chatbot",
+            api_key_env="LIVE_ENDPOINT_KEY",
+        )
+
+        headers = _resolve_auth_headers(config, ModelCredentials(api_key="api-key:ref"))
+
+        assert headers["Authorization"] == "Bearer real-key"
+
+    def test_openai_tool_call_response_is_not_silently_empty(self) -> None:
+        with pytest.raises(ValueError, match="tool_calls"):
+            _extract_openai(
+                {
+                    "choices": [
+                        {
+                            "message": {
+                                "content": None,
+                                "tool_calls": [{"function": {"name": "search"}}],
+                            }
+                        }
+                    ]
+                }
+            )
+
+    def test_from_parameters_missing_key(self) -> None:
+        with pytest.raises(ValueError, match="live_collection"):
+            CollectorConfig.from_parameters({})
+
+
+# -- Auth resolution ----------------------------------------------------------
+
+
+class TestAuthResolution:
+    def test_api_key_env_sets_bearer(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setenv("TEST_KEY", "secret")
+        qpath = tmp_path / "q.csv"
+        qpath.write_text("question\nhello\n", encoding="utf-8")
+
+        calls: list[dict[str, Any]] = []
+
+        def fake_post(*_: Any, **kwargs: Any) -> httpx.Response:
+            calls.append(kwargs)
+            return _openai_ok()
+
+        config = CollectorConfig(
+            questions_path=qpath,
+            output_dir=tmp_path / "out",
+            endpoint_url="https://api.example/v1",
+            model="chatbot",
+            api_key_env="TEST_KEY",
+            use_model_credentials=False,
+        )
+        mock_client = MagicMock()
+        mock_client.post = fake_post
+        collect_responses(config, client=mock_client, credentials=None)
+
+        assert calls[0]["headers"]["Authorization"] == "Bearer secret"
+
+    def test_missing_api_key_env_raises(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.delenv("MISSING_KEY", raising=False)
+        qpath = tmp_path / "q.csv"
+        qpath.write_text("question\nhello\n", encoding="utf-8")
+
+        config = CollectorConfig(
+            questions_path=qpath,
+            output_dir=tmp_path / "out",
+            endpoint_url="https://api.example/v1",
+            model="chatbot",
+            api_key_env="MISSING_KEY",
+            use_model_credentials=False,
+        )
+        with pytest.raises(ValueError, match="MISSING_KEY"):
+            collect_responses(config, client=MagicMock(), credentials=None)
+
+    def test_request_headers_override(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setenv("TEST_KEY", "from-env")
+        qpath = tmp_path / "q.csv"
+        qpath.write_text("question\nhello\n", encoding="utf-8")
+
+        calls: list[dict[str, Any]] = []
+
+        def fake_post(*_: Any, **kwargs: Any) -> httpx.Response:
+            calls.append(kwargs)
+            return _openai_ok()
+
+        config = CollectorConfig(
+            questions_path=qpath,
+            output_dir=tmp_path / "out",
+            endpoint_url="https://api.example/v1",
+            model="chatbot",
+            api_key_env="TEST_KEY",
+            request_headers={"Authorization": "Custom override"},
+            use_model_credentials=False,
+        )
+        mock_client = MagicMock()
+        mock_client.post = fake_post
+        collect_responses(config, client=mock_client, credentials=None)
+
+        assert calls[0]["headers"]["Authorization"] == "Custom override"
+
+
+# -- Question loading ---------------------------------------------------------
+
+
+class TestLoadQuestions:
+    def test_csv(self, tmp_path: Path) -> None:
+        p = tmp_path / "q.csv"
+        p.write_text("id,question\n1,hello\n2,world\n", encoding="utf-8")
+        qs = load_questions(p)
+        assert len(qs) == 2
+        assert qs[0].question == "hello"
+        assert qs[0].question_id == "1"
+
+    def test_jsonl(self, tmp_path: Path) -> None:
+        p = tmp_path / "q.jsonl"
+        p.write_text(
+            json.dumps({"question": "hi", "id": "a"})
+            + "\n"
+            + json.dumps({"question": "bye", "id": "b"})
+            + "\n",
+            encoding="utf-8",
+        )
+        qs = load_questions(p)
+        assert len(qs) == 2
+        assert qs[1].question == "bye"
+
+    def test_json_array(self, tmp_path: Path) -> None:
+        p = tmp_path / "q.json"
+        p.write_text(json.dumps([{"question": "test"}]), encoding="utf-8")
+        qs = load_questions(p)
+        assert len(qs) == 1
+
+    def test_json_wrapper(self, tmp_path: Path) -> None:
+        p = tmp_path / "q.json"
+        p.write_text(
+            json.dumps({"questions": [{"question": "test"}]}), encoding="utf-8"
+        )
+        qs = load_questions(p)
+        assert len(qs) == 1
+
+    def test_empty_file_raises(self, tmp_path: Path) -> None:
+        p = tmp_path / "q.csv"
+        p.write_text("question\n", encoding="utf-8")
+        with pytest.raises(ValueError, match="No questions"):
+            load_questions(p)
+
+    def test_missing_file_raises(self, tmp_path: Path) -> None:
+        with pytest.raises(FileNotFoundError):
+            load_questions(tmp_path / "missing.csv")
+
+    def test_skips_empty_questions(self, tmp_path: Path) -> None:
+        p = tmp_path / "q.csv"
+        p.write_text("question\nhello\n\nworld\n", encoding="utf-8")
+        qs = load_questions(p)
+        assert len(qs) == 2
+
+    def test_preserves_source_row(self, tmp_path: Path) -> None:
+        p = tmp_path / "q.csv"
+        p.write_text(
+            "question,category,difficulty\nhello,greet,easy\n", encoding="utf-8"
+        )
+        qs = load_questions(p)
+        assert qs[0].source_row == {
+            "question": "hello",
+            "category": "greet",
+            "difficulty": "easy",
+        }
+
+
+# -- Template substitution and path extraction --------------------------------
+
+
+class TestSubstituteTemplate:
+    def test_string(self) -> None:
+        assert substitute_template("Hi {question}", {"question": "hello"}) == "Hi hello"
+
+    def test_nested_dict(self) -> None:
+        result = substitute_template({"a": {"b": "{question}"}}, {"question": "test"})
+        assert result == {"a": {"b": "test"}}
+
+    def test_list(self) -> None:
+        result = substitute_template(["{question}", 42], {"question": "hi"})
+        assert result == ["hi", 42]
+
+    def test_non_string_passthrough(self) -> None:
+        assert substitute_template(42, {"question": "hi"}) == 42
+        assert substitute_template(True, {"question": "hi"}) is True
+
+
+class TestExtractByPath:
+    def test_simple_key(self) -> None:
+        assert extract_by_path({"a": "val"}, "a") == "val"
+
+    def test_nested(self) -> None:
+        assert extract_by_path({"a": {"b": {"c": "val"}}}, "a.b.c") == "val"
+
+    def test_array_index(self) -> None:
+        assert extract_by_path({"a": [10, 20, 30]}, "a.1") == 20
+
+    def test_missing_returns_none(self) -> None:
+        assert extract_by_path({"a": 1}, "b") is None
+
+    def test_openai_path(self) -> None:
+        data = {"choices": [{"message": {"content": "answer"}}]}
+        assert extract_by_path(data, "choices.0.message.content") == "answer"
+
+    def test_mcp_path(self) -> None:
+        data = {"result": {"content": [{"text": "answer"}]}}
+        assert extract_by_path(data, "result.content.0.text") == "answer"
+
+
+# -- OpenAI chat collection ---------------------------------------------------
+
+
+class TestCollectOpenAI:
+    def test_collects_responses(self, tmp_path: Path) -> None:
+        qpath = tmp_path / "q.csv"
+        qpath.write_text("question\nhello\nworld\n", encoding="utf-8")
+
+        def fake_post(*_: Any, **kwargs: Any) -> httpx.Response:
+            return _openai_ok("answer")
+
+        config = CollectorConfig(
+            questions_path=qpath,
+            output_dir=tmp_path / "out",
+            endpoint_url="https://api.example/v1",
+            model="chatbot",
+            use_model_credentials=False,
+        )
+        mock_client = MagicMock()
+        mock_client.post = fake_post
+        manifest = collect_responses(config, client=mock_client, credentials=None)
+
+        assert manifest.total == 2
+        assert manifest.completed == 2
+        assert manifest.failed == 0
+
+        output = tmp_path / "out" / "responses.jsonl"
+        rows = [json.loads(line) for line in output.read_text().splitlines()]
+        assert rows[0]["response"] == "answer"
+        assert rows[0]["error"] is None
+        assert rows[0]["latency_ms"] is not None
+
+    def test_writes_manifest(self, tmp_path: Path) -> None:
+        qpath = tmp_path / "q.csv"
+        qpath.write_text("question\nhello\n", encoding="utf-8")
+
+        mock_client = MagicMock()
+        mock_client.post = MagicMock(return_value=_openai_ok())
+
+        config = CollectorConfig(
+            questions_path=qpath,
+            output_dir=tmp_path / "out",
+            endpoint_url="https://api.example/v1",
+            model="chatbot",
+            use_model_credentials=False,
+        )
+        collect_responses(config, client=mock_client, credentials=None)
+
+        manifest_path = tmp_path / "out" / "manifest.json"
+        assert manifest_path.exists()
+        manifest = json.loads(manifest_path.read_text())
+        assert manifest["experimental"] is True
+        assert manifest["protocol"] == "openai_chat_completions"
+        assert manifest["total"] == 1
+        assert manifest["completed"] == 1
+
+    def test_system_prompt_included(self, tmp_path: Path) -> None:
+        qpath = tmp_path / "q.csv"
+        qpath.write_text("question\nhello\n", encoding="utf-8")
+
+        calls: list[dict[str, Any]] = []
+
+        def fake_post(*_: Any, **kwargs: Any) -> httpx.Response:
+            calls.append(kwargs)
+            return _openai_ok()
+
+        config = CollectorConfig(
+            questions_path=qpath,
+            output_dir=tmp_path / "out",
+            endpoint_url="https://api.example/v1",
+            model="chatbot",
+            system_prompt="You are a helpful assistant.",
+            use_model_credentials=False,
+        )
+        mock_client = MagicMock()
+        mock_client.post = fake_post
+        collect_responses(config, client=mock_client, credentials=None)
+
+        body = calls[0]["json"]
+        assert body["messages"][0] == {
+            "role": "system",
+            "content": "You are a helpful assistant.",
+        }
+        assert body["messages"][1] == {"role": "user", "content": "hello"}
+
+    def test_redirect_recorded_as_error(self, tmp_path: Path) -> None:
+        qpath = tmp_path / "q.csv"
+        qpath.write_text("question\nhello\n", encoding="utf-8")
+
+        def fake_post(*_: Any, **kwargs: Any) -> httpx.Response:
+            return httpx.Response(
+                307,
+                headers={"location": "https://other.example"},
+                request=httpx.Request(
+                    "POST", "https://api.example/v1/chat/completions"
+                ),
+            )
+
+        config = CollectorConfig(
+            questions_path=qpath,
+            output_dir=tmp_path / "out",
+            endpoint_url="https://api.example/v1",
+            model="chatbot",
+            use_model_credentials=False,
+        )
+        mock_client = MagicMock()
+        mock_client.post = fake_post
+        manifest = collect_responses(config, client=mock_client, credentials=None)
+
+        assert manifest.failed == 1
+        rows = [
+            json.loads(line)
+            for line in (tmp_path / "out" / "responses.jsonl").read_text().splitlines()
+        ]
+        assert "Redirect" in rows[0]["error"]
+
+    def test_fail_fast_raises(self, tmp_path: Path) -> None:
+        qpath = tmp_path / "q.csv"
+        qpath.write_text("question\nhello\nworld\n", encoding="utf-8")
+
+        def fake_post(*_: Any, **kwargs: Any) -> httpx.Response:
+            return _response(500, {"error": "server error"})
+
+        config = CollectorConfig(
+            questions_path=qpath,
+            output_dir=tmp_path / "out",
+            endpoint_url="https://api.example/v1",
+            model="chatbot",
+            fail_fast=True,
+            use_model_credentials=False,
+        )
+        mock_client = MagicMock()
+        mock_client.post = fake_post
+        with pytest.raises(CollectorError):
+            collect_responses(config, client=mock_client, credentials=None)
+
+        manifest = json.loads(
+            (tmp_path / "out" / "manifest.json").read_text(encoding="utf-8")
+        )
+        assert manifest["total"] == 2
+        assert manifest["completed"] == 0
+        assert manifest["failed"] == 1
+
+    def test_best_effort_records_errors(self, tmp_path: Path) -> None:
+        qpath = tmp_path / "q.csv"
+        qpath.write_text("question\nhello\nworld\n", encoding="utf-8")
+
+        call_count = 0
+
+        def fake_post(*_: Any, **kwargs: Any) -> httpx.Response:
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                return _response(500, {"error": "fail"})
+            return _openai_ok()
+
+        config = CollectorConfig(
+            questions_path=qpath,
+            output_dir=tmp_path / "out",
+            endpoint_url="https://api.example/v1",
+            model="chatbot",
+            fail_fast=False,
+            use_model_credentials=False,
+        )
+        mock_client = MagicMock()
+        mock_client.post = fake_post
+        manifest = collect_responses(config, client=mock_client, credentials=None)
+
+        assert manifest.failed == 1
+        assert manifest.completed == 1
+
+    def test_retries_transient_errors(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        qpath = tmp_path / "q.csv"
+        qpath.write_text("question\nhello\n", encoding="utf-8")
+        monkeypatch.setattr("evalhub.adapter.collector.time.sleep", lambda _: None)
+
+        responses = [
+            _response(500, {"error": "temporary"}),
+            _openai_ok("after retry"),
+        ]
+
+        def fake_post(*_: Any, **kwargs: Any) -> httpx.Response:
+            return responses.pop(0)
+
+        config = CollectorConfig(
+            questions_path=qpath,
+            output_dir=tmp_path / "out",
+            endpoint_url="https://api.example/v1",
+            model="chatbot",
+            max_retries=1,
+            use_model_credentials=False,
+        )
+        mock_client = MagicMock()
+        mock_client.post = fake_post
+        manifest = collect_responses(config, client=mock_client, credentials=None)
+
+        assert manifest.completed == 1
+        rows = [
+            json.loads(line)
+            for line in (tmp_path / "out" / "responses.jsonl").read_text().splitlines()
+        ]
+        assert rows[0]["response"] == "after retry"
+
+    def test_does_not_retry_client_errors(self, tmp_path: Path) -> None:
+        qpath = tmp_path / "q.csv"
+        qpath.write_text("question\nhello\n", encoding="utf-8")
+        mock_client = MagicMock()
+        mock_client.post.return_value = _response(401, {"error": "unauthorized"})
+
+        config = CollectorConfig(
+            questions_path=qpath,
+            output_dir=tmp_path / "out",
+            endpoint_url="https://api.example/v1",
+            model="chatbot",
+            max_retries=3,
+            use_model_credentials=False,
+        )
+
+        manifest = collect_responses(config, client=mock_client, credentials=None)
+
+        assert manifest.failed == 1
+        assert mock_client.post.call_count == 1
+
+    def test_retries_rate_limited_response(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        qpath = tmp_path / "q.csv"
+        qpath.write_text("question\nhello\n", encoding="utf-8")
+        monkeypatch.setattr("evalhub.adapter.collector.time.sleep", lambda _: None)
+        mock_client = MagicMock()
+        mock_client.post.side_effect = [
+            _response(429, {"error": "rate limited"}),
+            _openai_ok("after backoff"),
+        ]
+
+        config = CollectorConfig(
+            questions_path=qpath,
+            output_dir=tmp_path / "out",
+            endpoint_url="https://api.example/v1",
+            model="chatbot",
+            max_retries=1,
+            retry_backoff_seconds=0.1,
+            use_model_credentials=False,
+        )
+
+        manifest = collect_responses(config, client=mock_client, credentials=None)
+
+        assert manifest.completed == 1
+        assert mock_client.post.call_count == 2
+
+    def test_progress_callback(self, tmp_path: Path) -> None:
+        qpath = tmp_path / "q.csv"
+        qpath.write_text("question\na\nb\nc\n", encoding="utf-8")
+
+        progress: list[float] = []
+
+        config = CollectorConfig(
+            questions_path=qpath,
+            output_dir=tmp_path / "out",
+            endpoint_url="https://api.example/v1",
+            model="chatbot",
+            use_model_credentials=False,
+        )
+        mock_client = MagicMock()
+        mock_client.post = MagicMock(return_value=_openai_ok())
+        collect_responses(
+            config,
+            client=mock_client,
+            credentials=None,
+            progress_callback=progress.append,
+        )
+
+        assert len(progress) == 3
+        assert progress[-1] == pytest.approx(1.0)
+
+
+# -- Generic HTTP collection --------------------------------------------------
+
+
+class TestCollectGenericHTTP:
+    def test_template_substitution(self, tmp_path: Path) -> None:
+        qpath = tmp_path / "q.csv"
+        qpath.write_text("question\nhello\n", encoding="utf-8")
+
+        calls: list[dict[str, Any]] = []
+
+        def fake_post(*_: Any, **kwargs: Any) -> httpx.Response:
+            calls.append(kwargs)
+            return _response(200, {"result": {"text": "answer"}})
+
+        config = CollectorConfig(
+            questions_path=qpath,
+            output_dir=tmp_path / "out",
+            endpoint_url="https://api.example/invoke",
+            protocol=CollectorProtocol.GENERIC_HTTP,
+            request_template={"input": "{question}", "id": "{question_id}"},
+            response_path="result.text",
+            use_model_credentials=False,
+        )
+        mock_client = MagicMock()
+        mock_client.post = fake_post
+        manifest = collect_responses(config, client=mock_client, credentials=None)
+
+        assert calls[0]["json"]["input"] == "hello"
+        assert manifest.completed == 1
+
+        rows = [
+            json.loads(line)
+            for line in (tmp_path / "out" / "responses.jsonl").read_text().splitlines()
+        ]
+        assert rows[0]["response"] == "answer"
+
+    def test_mcp_style_request(self, tmp_path: Path) -> None:
+        qpath = tmp_path / "q.jsonl"
+        qpath.write_text(
+            json.dumps({"question": "What is K8s?"}) + "\n", encoding="utf-8"
+        )
+
+        def fake_post(*_: Any, **kwargs: Any) -> httpx.Response:
+            return _response(
+                200,
+                {
+                    "jsonrpc": "2.0",
+                    "result": {
+                        "content": [{"type": "text", "text": "Kubernetes is..."}]
+                    },
+                    "id": "1",
+                },
+            )
+
+        config = CollectorConfig(
+            questions_path=qpath,
+            output_dir=tmp_path / "out",
+            endpoint_url="https://mcp-chatbot.example/mcp",
+            protocol=CollectorProtocol.GENERIC_HTTP,
+            request_template={
+                "jsonrpc": "2.0",
+                "method": "tools/call",
+                "params": {"name": "chat", "arguments": {"message": "{question}"}},
+                "id": "{question_id}",
+            },
+            response_path="result.content.0.text",
+            use_model_credentials=False,
+        )
+        mock_client = MagicMock()
+        mock_client.post = fake_post
+        manifest = collect_responses(config, client=mock_client, credentials=None)
+
+        assert manifest.completed == 1
+        rows = [
+            json.loads(line)
+            for line in (tmp_path / "out" / "responses.jsonl").read_text().splitlines()
+        ]
+        assert rows[0]["response"] == "Kubernetes is..."
+
+    def test_extra_fields_cannot_overwrite_collector_fields(self) -> None:
+        record = CollectedRecord(
+            response="collected answer",
+            raw_response={"answer": "collected answer"},
+            error=None,
+            latency_ms=12.5,
+            extra_fields={
+                "response": "wrong answer",
+                "raw_response": {"answer": "wrong answer"},
+                "error": "wrong error",
+                "latency_ms": 999.0,
+            },
+        )
+
+        flattened = _flatten_record(record)
+
+        assert flattened["response"] == "collected answer"
+        assert flattened["raw_response"] == {"answer": "collected answer"}
+        assert flattened["error"] is None
+        assert flattened["latency_ms"] == 12.5
+
+    def test_extra_response_paths_extracted(self, tmp_path: Path) -> None:
+        qpath = tmp_path / "q.csv"
+        qpath.write_text("question\nhello\n", encoding="utf-8")
+
+        def fake_post(*_: Any, **kwargs: Any) -> httpx.Response:
+            return _response(
+                200,
+                {
+                    "output": {
+                        "answer": "the answer",
+                        "context": [{"source": "doc1", "text": "relevant text"}],
+                        "score": 0.95,
+                    }
+                },
+            )
+
+        config = CollectorConfig(
+            questions_path=qpath,
+            output_dir=tmp_path / "out",
+            endpoint_url="https://api.example",
+            protocol=CollectorProtocol.GENERIC_HTTP,
+            request_template={"input": {"question": "{question}"}},
+            response_path="output.answer",
+            extra_response_paths={
+                "contexts": "output.context",
+                "confidence": "output.score",
+            },
+            use_model_credentials=False,
+        )
+        mock_client = MagicMock()
+        mock_client.post = fake_post
+        collect_responses(config, client=mock_client, credentials=None)
+
+        rows = [
+            json.loads(line)
+            for line in (tmp_path / "out" / "responses.jsonl").read_text().splitlines()
+        ]
+        assert rows[0]["response"] == "the answer"
+        assert rows[0]["contexts"] == [{"source": "doc1", "text": "relevant text"}]
+        assert rows[0]["confidence"] == 0.95
+
+    def test_input_metadata_flattened_to_top_level(self, tmp_path: Path) -> None:
+        qpath = tmp_path / "q.csv"
+        qpath.write_text(
+            "question,reference,category\nWhat is X?,X is Y,general\n",
+            encoding="utf-8",
+        )
+
+        def fake_post(*_: Any, **kwargs: Any) -> httpx.Response:
+            return _response(200, {"output": {"answer": "X is Y indeed"}})
+
+        config = CollectorConfig(
+            questions_path=qpath,
+            output_dir=tmp_path / "out",
+            endpoint_url="https://api.example",
+            protocol=CollectorProtocol.GENERIC_HTTP,
+            request_template={"input": {"question": "{question}"}},
+            response_path="output.answer",
+            use_model_credentials=False,
+        )
+        mock_client = MagicMock()
+        mock_client.post = fake_post
+        collect_responses(config, client=mock_client, credentials=None)
+
+        rows = [
+            json.loads(line)
+            for line in (tmp_path / "out" / "responses.jsonl").read_text().splitlines()
+        ]
+        assert rows[0]["response"] == "X is Y indeed"
+        assert rows[0]["reference"] == "X is Y"
+        assert rows[0]["category"] == "general"
+        assert "metadata" not in rows[0]
+
+    def test_response_path_missing_returns_none(self, tmp_path: Path) -> None:
+        qpath = tmp_path / "q.csv"
+        qpath.write_text("question\nhello\n", encoding="utf-8")
+
+        def fake_post(*_: Any, **kwargs: Any) -> httpx.Response:
+            return _response(200, {"unexpected": "shape"})
+
+        config = CollectorConfig(
+            questions_path=qpath,
+            output_dir=tmp_path / "out",
+            endpoint_url="https://api.example",
+            protocol=CollectorProtocol.GENERIC_HTTP,
+            request_template={"q": "{question}"},
+            response_path="result.text",
+            use_model_credentials=False,
+        )
+        mock_client = MagicMock()
+        mock_client.post = fake_post
+        collect_responses(config, client=mock_client, credentials=None)
+
+        rows = [
+            json.loads(line)
+            for line in (tmp_path / "out" / "responses.jsonl").read_text().splitlines()
+        ]
+        assert rows[0]["response"] is None
+
+
+# -- collect_responses_from_parameters ----------------------------------------
+
+
+class TestCollectFromParameters:
+    def test_convenience_wrapper(self, tmp_path: Path) -> None:
+        qpath = tmp_path / "q.csv"
+        qpath.write_text("question\nhello\n", encoding="utf-8")
+
+        mock_client = MagicMock()
+        mock_client.post = MagicMock(return_value=_openai_ok())
+
+        manifest = collect_responses_from_parameters(
+            {
+                "live_collection": {
+                    "questions_path": str(qpath),
+                    "output_dir": str(tmp_path / "out"),
+                    "endpoint_url": "https://api.example/v1",
+                    "model": "chatbot",
+                    "use_model_credentials": False,
+                }
+            },
+            client=mock_client,
+            credentials=None,
+        )
+
+        assert manifest.completed == 1