diff --git a/CHANGELOG.md b/CHANGELOG.md
index 03e2de49..afce8dd7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -24,6 +24,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Native observability connectors: connection tests verify bounded query access instead of public health endpoints. Explicit instance reads require ID-bound credentials; migrated defaults must be backfilled through the administrator save flow before rollout. Prometheus and Mimir default to a 10-second execution ceiling, configurable per datasource and tightened by Explore; provider response bytes and body reads are bounded. Valid short Jaeger trace IDs are retained, and local loss, search saturation and unknown results carry distinct completeness evidence.
 - Security scanning hardening: deployment workflows select only explicit preview secrets; chat history deletion hides internal errors and session IDs always use cryptographic randomness. The configurator escapes literal HCL strings, screenshot login requires exact configured origins and valid TLS, presenter titles retain text semantics, datasource hostname classification checks the AWS DNS suffix including root-dot and China-partition forms, and rotation diagnostics omit unrelated secret identifiers.
 - AgentCore Gateway inbound authentication: the provisioner creates every section gateway with IAM (SigV4) inbound authentication instead of none. Because a gateway's authorizer type cannot be changed after creation, `make agentcore` fails on any existing gateway that still accepts unauthenticated calls, and the AgentCore reference documents the operator replacement procedure.
 - AI chat grounding: the chat agent is told the current time, the account and region its tools read, and how to treat capped, unfinished or failed results. Capped AWS list results report how many items were returned and found and whether more exist, a list is called complete only when its tool says so, and a successful empty result is reported as empty. Unfinished or failed Logs Insights queries, AWS-capped route searches, ECS lists with failed descriptions, unreadable serverless-cache listings and Cost Explorer comparisons that stopped early are reported as incomplete or unknown rather than complete or empty. Synced-inventory answers state that they cover the host account and warn when another account was asked for; a requested account that is unregistered, disabled or unverifiable is named instead of silently falling back to the host account; an answer produced without live tools starts with a notice that it uses no real data; and long conversations send only the most recent turns that fit the request size. A follow-up question without a domain keyword ("why is that?", "more detail") stays with the section that answered the previous turn instead of falling back to the general Ops agent, while an explicit keyword, a pin or (with hybrid routing) a classifier choice still re-routes. On AgentCore agent answers, the agent is told which regions AWSops is configured to inventory for the target account (when that scope can be read; it is an AWSops setting, not which regions AWS has enabled), so it queries them and names the regions it checked rather than answering account-wide questions from one region. When a thread switches accounts, earlier turns are labelled with the account they were about (where that was recorded), so answers about one account are not presented as facts about another. Long answers get an explicit output budget instead of the model default, so root-cause answers are no longer cut mid-sentence, and an answer that still stops early says whether it hit the length limit or lost the connection. Synced-inventory rows returned by the inventory query tool carry their region and, for EC2, instance id, type, power state and last state change (the reader view allowlists those payload keys; the last-change time is empty for running instances), so region-scoped and running/stopped questions use recorded values instead of resource-name prefixes or an "unknown" answer.
@@ -691,6 +692,7 @@ First release of the **v2 line** (versioned independently from the v1 1.x line,
 
 ### 수정
 
+- 외부 관측 데이터소스: 연결 테스트가 공개 상태 엔드포인트 대신 제한된 실제 조회 권한을 검증합니다. 인스턴스 ID 조회는 해당 ID의 자격 증명만 허용하며, 이전 기본 연결은 배포 전에 관리자 저장 흐름으로 자격 증명을 보완해야 합니다. Prometheus·Mimir의 기본 실행 제한은 10초이며 데이터소스별 설정과 Explore의 더 짧은 제한을 적용합니다. 응답 크기와 본문 읽기 시간을 제한하고, 짧은 Jaeger trace ID를 보존하며 실제 누락·조회 한도 도달·판별 불가 결과를 구분합니다.
 - 보안 검사 개선: 배포 워크플로가 명시된 미리보기 시크릿만 선택하며, 채팅 기록 삭제 오류는 내부 정보를 숨기고 세션 ID는 암호학적 난수만 사용합니다. 설정 도구는 HCL 문자열을 안전하게 인코딩하고, 스크린샷 로그인은 지정한 정확한 출처와 유효한 TLS를 요구합니다. 발표자 제목은 텍스트로 보존하며, 데이터소스 호스트 분류는 DNS 루트 점과 중국 파티션을 포함한 AWS 도메인 접미사를 확인하고, 회전 진단에서는 무관한 시크릿 식별자를 생략합니다.
 - AgentCore Gateway 유입 인증: 프로비저너가 모든 섹션 Gateway를 인증 없음 대신 IAM(SigV4) 유입 인증으로 생성합니다. Gateway의 인증 방식은 생성 후 바꿀 수 없으므로, 아직 인증 없는 호출을 받는 기존 Gateway가 있으면 `make agentcore`가 실패하며, 운영자 교체 절차는 AgentCore 참조 문서에 안내합니다.
 - AI 채팅 근거 강화: 채팅 에이전트가 현재 시각, 도구가 조회하는 계정·리전, 잘리거나 끝나지 않았거나 실패한 결과를 다루는 방법을 전달받습니다. 상한이 걸린 AWS 목록 결과는 반환·조회 개수와 추가 항목 여부를 알려 주고, 도구가 완전하다고 표시한 목록만 완전하다고 말하며, 성공했지만 비어 있는 결과는 비어 있다고 답합니다. 끝나지 않았거나 실패한 Logs Insights 쿼리, AWS가 상한을 둔 경로 검색, 상세 조회가 실패한 ECS 목록, 읽을 수 없는 서버리스 캐시 목록, 중간에 멈춘 Cost Explorer 비교는 완전하거나 비어 있다고 하지 않고 불완전하거나 알 수 없다고 보고합니다. 동기화 인벤토리 답변은 호스트 계정 기준임을 밝히며 다른 계정을 물었을 때 경고하고, 요청한 계정이 등록되지 않았거나 비활성화되었거나 확인할 수 없으면 조용히 호스트 계정으로 바꾸지 않고 그 사실을 밝히며, 실시간 도구 없이 만든 답변은 실제 데이터가 아니라는 안내로 시작하고, 긴 대화는 요청 크기에 맞는 최근 대화만 보냅니다. 도메인 키워드가 없는 후속 질문("그건 왜 그래?", "더 자세히")은 일반 Ops 에이전트로 넘어가지 않고 직전 답변을 한 섹션이 이어서 답합니다. 명시적 키워드, 섹션 고정, (하이브리드 라우팅에서는) 분류기의 선택은 이전처럼 다시 라우팅합니다. AgentCore 에이전트 답변에서는 에이전트가 AWSops가 대상 계정에 대해 수집하도록 설정된 리전 목록을(조회할 수 있을 때. AWS에서 활성화된 리전이 아니라 AWSops 설정입니다) 받아, 계정 전체 질문을 한 리전만 보고 답하지 않고 해당 리전들을 조회한 뒤 확인한 리전을 밝힙니다. 한 스레드에서 계정을 바꾸면 이전 턴에 (기록된 경우) 해당 계정이 표시되어, 다른 계정에 대한 답변을 현재 계정의 사실로 제시하지 않습니다. 긴 답변에는 모델 기본값 대신 명시적인 출력 한도를 적용해 원인 분석 답변이 문장 중간에 끊기지 않으며, 그래도 일찍 끝난 답변은 출력 길이 한도에 도달했는지 연결이 끊겼는지를 알려 줍니다. 인벤토리 조회 도구가 반환하는 동기화 행에 리전이 포함되고 EC2는 인스턴스 ID·유형·전원 상태·마지막 상태 변경 시각도 포함되어(리더 뷰가 이 페이로드 키를 허용하며 실행 중 인스턴스의 상태 변경 시각은 비어 있음), 리전 한정 질문과 실행/중지 질문이 이름 접두사 추정이나 "알 수 없음" 대신 기록된 값으로 답변됩니다.
diff --git a/agent/lambda/clickhouse_mcp.py b/agent/lambda/clickhouse_mcp.py
index a8295ef9..44eb9528 100644
--- a/agent/lambda/clickhouse_mcp.py
+++ b/agent/lambda/clickhouse_mcp.py
@@ -17,13 +17,14 @@
 
 from cross_account import resolve_tool_name
 from datasource_http import (
+    MAX_TRACE_RESPONSE_BYTES,
     NotConnected,
     SsrfBlocked,
     assert_host_allowed,
     auth_headers,
-    health,
     http_json,
     load_datasource,
+    query_probe,
     set_request_conn,
 )
 from sql_readonly_guard import assert_read_only as _shared_assert_read_only
@@ -161,7 +162,9 @@ def _run_sql(sql, max_rows, trusted=False, max_execution_time=None):
     # HTTP timeout ALIGNED ABOVE the execution bound (+3s margin): the shared default of 12s
     # would client-abort a legitimately long query while the server kept scanning under
     # timeout_overflow_mode=throw — the server-side bound must fire first.
-    status, data = http_json("POST", url, headers=headers, body=body, timeout=max_execution_time + 3)
+    # Wide max_result_rows-bounded results can exceed the default cap.
+    status, data = http_json("POST", url, headers=headers, body=body, timeout=max_execution_time + 3,
+                             max_bytes=MAX_TRACE_RESPONSE_BYTES)
     if status >= 400:
         snippet = (data.get("raw") or data.get("exception") or data) if isinstance(data, dict) else "non-object error response"
         return err(f"ClickHouse query failed ({status}): {str(snippet)[:300]}")
@@ -273,8 +276,12 @@ def clickhouse_schema(args):
 
 
 def clickhouse_health(args):
-    """Connectivity probe for the pre-save Test / status badge: GET /ping."""
-    return ok(health(load_datasource(SLUG), "/ping"))
+    """Verify read-only SQL access using one constant row and a one-second ceiling."""
+    def run():
+        result = _run_sql("SELECT 1 AS probe", 1, max_execution_time=1)
+        return json.loads(result["body"]).get("collectionStatus") if result["statusCode"] == 200 else None
+    # "empty" is impossible for SELECT 1; "partial" means the result cap applied, which still proves access.
+    return ok(query_probe(run, "ClickHouse query access could not be verified", accepted=("ok", "partial")))
 
 
 _TOOLS = {
diff --git a/agent/lambda/datadog_mcp.py b/agent/lambda/datadog_mcp.py
index 9ef1041f..6dea0288 100644
--- a/agent/lambda/datadog_mcp.py
+++ b/agent/lambda/datadog_mcp.py
@@ -16,7 +16,7 @@
 from cross_account import resolve_tool_name
 from datasource_http import (
     NotConnected, SsrfBlocked, assert_host_allowed, auth_headers, http_json, load_datasource,
-    set_request_conn,
+    query_probe, set_request_conn, request_timeout, finite_number, observed_payload, HTTP_TIMEOUT,
 )
 
 SLUG = "datadog"
@@ -43,11 +43,13 @@ def _parse_time_s(v, default_delta_s=None):
 
 def _headers(creds):
     """DD dual-key headers: dedicated apiKey/appKey fields win; else the generic custom-header pair
-    (headerName/headerValue [+2]) from auth_headers covers users who configured them directly."""
+    (headerName/headerValue [+2]) from auth_headers covers users who configured them directly.
+    Dedicated keys apply only to a blob with no authType; any explicit type (including custom_header
+    and none) uses its own headers."""
     h = dict(auth_headers(creds))
-    if creds.get("apiKey"):
+    if creds.get("authType") is None and creds.get("apiKey"):
         h["DD-API-KEY"] = str(creds["apiKey"])
-    if creds.get("appKey"):
+    if creds.get("authType") is None and creds.get("appKey"):
         h["DD-APPLICATION-KEY"] = str(creds["appKey"])
     return h
 
@@ -58,13 +60,14 @@ def _ds():
     return creds
 
 
-def _get(creds, path, params=None):
+def _get(creds, path, params=None, *, timeout=None):
     url = creds["endpoint"].rstrip("/") + path + ("?" + urlencode(params, doseq=True) if params else "")
-    status, data = http_json("GET", url, headers=_headers(creds))
+    status, data = http_json("GET", url, headers=_headers(creds),
+                             timeout=timeout if timeout is not None else request_timeout(creds))
     if status >= 400:
         errs = data.get("errors") if isinstance(data, dict) else None
         raise _ApiError(f"Datadog HTTP {status}: {str(errs or (data.get('raw') if isinstance(data, dict) else data))[:300]}")
-    return data
+    return observed_payload(data, status)
 
 
 def datadog_query(args):
@@ -77,38 +80,76 @@ def datadog_query(args):
         "to": _parse_time_s(args.get("end")),
     }
     data = _get(_ds(), "/api/v1/query", params)
-    series = data.get("series", []) if isinstance(data, dict) else []
+    if isinstance(data, dict) and (data.get("status") == "error" or data.get("errors") or data.get("error")):
+        return ok({"series": [], "truncated": False, "collectionStatus": "error"})
+    series = data.get("series") if isinstance(data, dict) else None
+    if not isinstance(data, dict) or data.get("status") != "ok" or not isinstance(series, list):
+        return ok({"series": [], "truncated": False, "collectionStatus": "unknown"})
     out = []
+    truncated = len(series) > MAX_SERIES
+    invalid = False
     for s in series[:MAX_SERIES]:
-        if not isinstance(s, dict):
+        if not isinstance(s, dict) or not isinstance(s.get("metric"), str) or not isinstance(s.get("pointlist"), list):
+            invalid = True
             continue
-        pts = s.get("pointlist") or []
+        pts = s["pointlist"]
+        truncated = truncated or len(pts) > MAX_POINTS
+        points = []
+        for point in pts[:MAX_POINTS]:
+            if (not isinstance(point, list) or len(point) != 2
+                    or not finite_number(point[0], 8640000000000000)
+                    or point[1] is not None and not finite_number(point[1])):
+                invalid = True
+                continue
+            points.append(point)
+        units = s.get("unit")
+        unit = units[0].get("name") if isinstance(units, list) and units and isinstance(units[0], dict) else None
         out.append({
             "metric": s.get("metric", ""), "scope": s.get("scope", ""),
-            "unit": (s.get("unit") or [{}])[0].get("name") if isinstance(s.get("unit"), list) and s.get("unit") else None,
-            "pointlist": pts[:MAX_POINTS],
+            "unit": unit, "pointlist": points,
         })
-    return ok({"series": out, "truncated": len(series) > MAX_SERIES,
+    http_status = getattr(data, "http_status", 200)
+    state = "unknown" if invalid or http_status not in (200, 206) else "partial" if truncated or http_status == 206 else "ok" if out else "empty"
+    return ok({"series": out, "truncated": truncated,
+               "collectionStatus": state,
                "status": data.get("status") if isinstance(data, dict) else None})
 
 
 def datadog_schema(args):
     """Actively-reporting metric names from the last hour (query-gen vocabulary)."""
-    data = _get(_ds(), "/api/v1/metrics", {"from": _parse_time_s(None, 3600)})
+    data = _get(_ds(), "/api/v1/metrics", {"from": _parse_time_s(None, 3600)}, timeout=HTTP_TIMEOUT)
     metrics = data.get("metrics", []) if isinstance(data, dict) else []
     metrics = metrics if isinstance(metrics, list) else []
     return ok({"metrics": metrics[:200], "truncated": len(metrics) > 200})
 
 
 def datadog_health(args):
-    """Key validation probe: GET /api/v1/validate (API key), then a 1-metric list (app key scope)."""
-    creds = load_datasource(SLUG)
-    assert_host_allowed(creds["endpoint"])
-    try:
-        v = _get(creds, "/api/v1/validate")
-        return ok({"ok": bool(isinstance(v, dict) and v.get("valid"))})
-    except _ApiError as e:
-        return ok({"ok": False, "error": str(e)})
+    """Validate configured API keys, then metric-query access (key pair or bearer) under one deadline."""
+    def run():
+        creds = load_datasource(SLUG)
+        assert_host_allowed(creds["endpoint"])
+        deadline = time.monotonic() + request_timeout(creds)
+
+        def remaining():
+            left = deadline - time.monotonic()
+            if left <= 0:
+                raise TimeoutError("Datadog connection test deadline exceeded")
+            return left
+        # The API-key validator does not validate bearer access tokens. Those use
+        # the supported metric-query operation itself as the authentication proof.
+        if any(name.lower() == "dd-api-key" for name in _headers(creds)):
+            v = _get(creds, "/api/v1/validate", timeout=remaining())
+            if not isinstance(v, dict) or v.get("valid") is not True:
+                return None
+        end = int(time.time())
+        # A one-minute, single-metric query with a probe-only host filter verifies
+        # metric-query permission without fetching the fleet's real metric series.
+        probe = _get(creds, "/api/v1/query", {
+            "query": "avg:system.cpu.user{host:awsops-connection-probe}",
+            "from": str(end - 60), "to": str(end),
+        }, timeout=remaining())
+        return "ok" if isinstance(probe, dict) and probe.get("status") == "ok" and isinstance(probe.get("series"), list) else None
+    return ok(query_probe(run, "Datadog credentials or metric-query access could not be verified"))
 
 
 _TOOLS = {
diff --git a/agent/lambda/datasource_http.py b/agent/lambda/datasource_http.py
index 9de17c9a..66d4c4fb 100644
--- a/agent/lambda/datasource_http.py
+++ b/agent/lambda/datasource_http.py
@@ -4,7 +4,10 @@
 endpoint + credential queried in a query language. This module
 centralizes: reading the per-slug credential from the single integrations secret, SSRF host guarding
 (always-block metadata/loopback/...; private allowed — in-cluster datasources are the target),
-auth headers (Basic/Bearer), and a no-redirect HTTP fetch. Stdlib + boto3 only.
+auth headers (Basic/Bearer), and a no-redirect HTTP fetch. Explicit instance IDs require an
+ID-bound credential; kind-only reads retain the default mirror. Responses are byte-bounded before
+parsing and body reads share the request's monotonic deadline. Connection tests execute bounded
+queries instead of trusting public health endpoints. Stdlib + boto3 only.
 
 SECURITY: never log credential material. The endpoint is user-supplied → assert_host_allowed before
 every request; redirects are NOT auto-followed (a malicious endpoint can't 30x to metadata/internal).
@@ -30,6 +33,7 @@
 stay resolve-based, connection time is pinned on both transports.
 """
 import base64
+import concurrent.futures
 import functools
 import http.client
 import ipaddress
@@ -37,11 +41,17 @@
 import os
 import re
 import socket
+import time
 import urllib.error
 import urllib.request
 from urllib.parse import urlparse
 
 HTTP_TIMEOUT = 12
+MAX_RESPONSE_BYTES = 4 * 1024 * 1024
+MAX_ERROR_BYTES = 64 * 1024
+# single-trace, log-query and ClickHouse SQL responses may exceed the default cap; trace/log connectors re-bound them to 1 MB after parsing, ClickHouse SQL by max_result_rows.
+# 8 MiB keeps the buffered body plus parsed JSON inside the 256 MB connector Lambda.
+MAX_TRACE_RESPONSE_BYTES = 8 * 1024 * 1024
 
 _SM = None
 _SECRET_CACHE = None
@@ -54,6 +64,10 @@
 # Reset every invocation by set_request_conn — see module docstring.
 _PINNED_IP_CACHE = {}
 
+# Shared (not per-call) pool for _open_within; 4 workers: a connection that ignores shutdown()
+# must not stall every later request.
+_OPEN_POOL = concurrent.futures.ThreadPoolExecutor(max_workers=4, thread_name_prefix="ds-open")
+
 
 class NotConnected(Exception):
     """Raised when a datasource slug has no stored credential/endpoint."""
@@ -63,6 +77,46 @@ class SsrfBlocked(Exception):
     """Raised when an endpoint host/scheme is disallowed."""
 
 
+class ResponseTooLarge(ValueError):
+    """The upstream body exceeded the read budget; this is never an empty result."""
+
+
+def _seconds(value, default, ceiling=55):
+    if isinstance(value, bool):
+        return default
+    text = str(value).strip()
+    if not re.fullmatch(r"\d{1,6}s?", text):
+        return default
+    return max(1, min(ceiling, int(text.removesuffix("s"))))
+
+
+def request_timeout(creds):
+    """Saved transport deadline for APIs without a provider execution-time parameter."""
+    return _seconds(creds.get("timeoutS"), HTTP_TIMEOUT)
+
+
+def execution_timeout(creds, requested=None):
+    """Provider execution ceiling, with a caller allowed only to tighten it."""
+    configured = _seconds(creds.get("timeoutS"), 10)
+    return min(configured, _seconds(requested, configured))
+
+
+def finite_number(value, bound=1.7976931348623157e308):
+    """JSON chart numbers exclude booleans, non-finite values and oversized integers."""
+    return type(value) in (int, float) and abs(value) <= bound
+
+
+class ObservedPayload(dict):
+    """Transport evidence is an attribute, so upstream JSON cannot impersonate it."""
+    def __init__(self, data, status):
+        super().__init__(data)
+        self.http_status = status
+
+
+def observed_payload(data, status):
+    return ObservedPayload(data, status) if isinstance(data, dict) else data
+
+
 # Cloud instance-metadata endpoints blocked even though private is otherwise allowed (mirror agent.py).
 _METADATA_IPS = frozenset({
     ipaddress.ip_address("169.254.169.254"),
@@ -179,6 +233,27 @@ def set_request_conn(conn):
     _PINNED_IP_CACHE = {}
 
 
+def query_probe(run, error, accepted=("ok", "empty", "partial")):
+    """Time one real bounded query for the connection test. `run` executes the connector's own
+    query tool and returns its `collectionStatus` (None when the response was unrecognizable).
+    Returns {ok, verified, latency_ms, error?} — fixed sentences only, never upstream text; the
+    connector's own classification (SSRF policy vs. unverifiable) is kept because an admin needs it."""
+    t0 = time.monotonic()
+    try:
+        state = run()
+    except SsrfBlocked:
+        state, error = None, "endpoint rejected by the private-network policy"
+    except NotConnected:
+        state, error = None, "datasource not connected"
+    except Exception as exc:
+        print(f"[datasource_http] connection probe failed: {type(exc).__name__}")
+        state = None
+    latency = int((time.monotonic() - t0) * 1000)
+    verified = state in accepted
+    return {"ok": verified, "verified": ["query"] if verified else [], "latency_ms": latency,
+            **({} if verified else {"error": error})}
+
+
 def health(creds, path):
     """Lightweight connectivity probe: GET endpoint+path with auth, SSRF-guarded. {ok, latency_ms, error?}."""
     import time as _t
@@ -229,17 +304,24 @@ def _load_secret_map():
 def load_datasource(slug, instance_id=None):
     """Resolve a datasource's connection blob. Precedence:
        1. the request-scoped inline conn-config (`_REQUEST_CONN`) — the trusted BFF path only;
-       2. the per-instance secret key `map[str(instance_id)]` — the credential-blind worker path,
-          where the caller sends ONLY an instance_id and the connector reads the secret server-side;
-       3. the `map[slug]` kind-mirror (the default instance) as the fallback.
+       2. for an explicit instance_id, the per-instance secret key `map[str(instance_id)]` only —
+          the credential-blind worker path, where the caller sends ONLY an instance_id and the
+          connector reads the secret server-side. It never falls back to the kind mirror: a sibling
+          must not inherit the default's credential, and a migrated default whose id key was never
+          written must be re-saved first (the administrator save writes it);
+       3. the `map[slug]` kind mirror, only when no explicit instance was requested.
     `instance_id` is a pure argument (no module-global), so warm-container reuse cannot bleed one
     invocation's instance into the next."""
     if _REQUEST_CONN is not None:
         return _REQUEST_CONN
     m = _load_secret_map()
-    creds = m.get(str(instance_id)) if instance_id is not None else None
-    if not (isinstance(creds, dict) and creds.get("endpoint")):
-        creds = m.get(slug)  # kind-mirror (default instance) fallback
+    if instance_id is not None:
+        creds = m.get(str(instance_id))
+        if not isinstance(creds, dict) or not creds.get("endpoint"):
+            print(f"[datasource_http] strict instance lookup failed: slug={slug} instance_id={instance_id}")
+            raise NotConnected("requested datasource instance is not configured")
+    else:
+        creds = m.get(slug)
     if not isinstance(creds, dict) or not creds.get("endpoint"):
         raise NotConnected(f"{slug} not connected (no endpoint configured in the Connectors UI)")
     return creds
@@ -253,7 +335,34 @@ def redirect_request(self, req, fp, code, msg, headers, newurl):
 # ProxyHandler({}) disables HTTP_PROXY/HTTPS_PROXY env-var routing — pinning implies a direct
 # connection to the validated IP; honoring a proxy would route the request elsewhere and defeat
 # the point of pinning (and break it, since the socket wouldn't reach the pinned IP at all).
-_opener = urllib.request.build_opener(_NoRedirect, urllib.request.ProxyHandler({}))
+class _ConnTracker:
+    """Per-request handle: the connection urllib opens for this request, so the deadline watchdog
+    in _open_within can close it (which unblocks the worker's recv) instead of abandoning it."""
+    conn = None
+
+
+def _tracked(conn_class, req):
+    def make(*args, **kwargs):
+        conn = conn_class(*args, **kwargs)
+        tracker = getattr(req, "_awsops_tracker", None)
+        if tracker is not None:
+            tracker.conn = conn
+        return conn
+    return make
+
+
+class _TrackingHTTPHandler(urllib.request.HTTPHandler):
+    def http_open(self, req):
+        return self.do_open(_tracked(http.client.HTTPConnection, req), req)
+
+
+class _TrackingHTTPSHandler(urllib.request.HTTPSHandler):
+    def https_open(self, req):
+        return self.do_open(_tracked(http.client.HTTPSConnection, req), req, context=self._context)
+
+
+_opener = urllib.request.build_opener(
+    _NoRedirect, urllib.request.ProxyHandler({}), _TrackingHTTPHandler, _TrackingHTTPSHandler)
 
 
 def _pin_create_connection(base_create, pinned_ips):
@@ -298,7 +407,8 @@ def __init__(self, pinned_ips):
         self._pinned_ips = pinned_ips
 
     def http_open(self, req):
-        return self.do_open(functools.partial(_PinnedHTTPConnection, pinned_ips=self._pinned_ips), req)
+        return self.do_open(
+            _tracked(functools.partial(_PinnedHTTPConnection, pinned_ips=self._pinned_ips), req), req)
 
 
 class _PinnedHTTPSHandler(urllib.request.HTTPSHandler):
@@ -314,7 +424,7 @@ def https_open(self, req):
         # inside HTTPSConnection, and self.host (unchanged — only the socket target is pinned) is
         # passed as server_hostname for SNI + the default context's built-in hostname/cert check.
         return self.do_open(
-            functools.partial(_PinnedHTTPSConnection, pinned_ips=self._pinned_ips),
+            _tracked(functools.partial(_PinnedHTTPSConnection, pinned_ips=self._pinned_ips), req),
             req,
             context=self._context,
         )
@@ -342,18 +452,117 @@ def _parse(raw):
         return {"raw": raw.decode("utf-8", "replace")[:4000]}
 
 
-def http_json(method, url, headers=None, body=None, timeout=HTTP_TIMEOUT):
+def _response_socket(response):
+    """Find urllib's HTTPResponse socket, including HTTPError's extra fp wrapper."""
+    stream = response
+    for _ in range(4):
+        raw = getattr(stream, "raw", None)
+        sock = getattr(raw, "_sock", None)
+        if sock is not None:
+            return sock
+        stream = getattr(stream, "fp", None)
+        if stream is None:
+            break
+    return None
+
+
+def _read_bounded(response, max_bytes, deadline):
+    """Read at most budget+1 bytes before the absolute monotonic `deadline`; never trust
+    Content-Length or parse before the bound. The deadline is the request's — connect, headers
+    and body share one budget, so a slow-but-alive upstream cannot double it."""
+    chunks = []
+    total = 0
+    read = getattr(response, "read1", response.read)
+    try:
+        while True:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise TimeoutError("datasource response read deadline exceeded")
+            sock = _response_socket(response)
+            if sock is not None:
+                sock.settimeout(remaining)
+            try:
+                chunk = read(min(65536, max_bytes - total + 1))
+            except socket.timeout as exc:
+                raise TimeoutError("datasource response read deadline exceeded") from exc
+            if time.monotonic() >= deadline:
+                raise TimeoutError("datasource response read deadline exceeded")
+            if not chunk:
+                return b"".join(chunks)
+            total += len(chunk)
+            if total > max_bytes:
+                raise ResponseTooLarge(f"datasource response exceeds {max_bytes} bytes")
+            chunks.append(chunk)
+    finally:
+        response.close()
+
+
+def _open_within(opener, req, timeout, deadline):
+    """Open the connection and read the response headers under the request's absolute deadline.
+    urllib's `timeout` is per socket operation, so a server dripping header bytes could hold
+    open() past the deadline. The open runs on the shared worker pool; at the deadline the
+    watchdog shuts the request's socket down (close() alone is a no-op while the response is
+    being read, because the file wrapper holds an io ref) and cancels an open that never
+    started; any response that still arrives later is closed by the completion callback."""
+    remaining = deadline - time.monotonic()
+    if remaining <= 0:
+        raise TimeoutError("datasource request deadline exceeded before connect")
+    req._awsops_tracker = tracker = _ConnTracker()
+    future = _OPEN_POOL.submit(opener.open, req, timeout=min(timeout, remaining))
+    try:
+        return future.result(timeout=remaining)
+    except concurrent.futures.TimeoutError as exc:
+        if future.cancel():
+            raise TimeoutError("datasource request deadline exceeded before connect") from exc
+        sock = getattr(tracker.conn, "sock", None)
+        if sock is not None:
+            try:
+                sock.shutdown(socket.SHUT_RDWR)
+            except OSError:
+                pass
+        if tracker.conn is not None:
+            try:
+                tracker.conn.close()
+            except OSError:
+                pass
+        future.add_done_callback(_close_late)
+        raise TimeoutError("datasource response header deadline exceeded") from exc
+
+
+def _close_late(future):
+    """Close whatever a timed-out open eventually produced: a response, or an HTTPError's body."""
+    error = future.exception()
+    late = future.result() if error is None else getattr(error, "fp", None)
+    if late is not None:
+        try:
+            late.close()
+        except OSError:
+            pass
+
+
+def http_json(method, url, headers=None, body=None, timeout=HTTP_TIMEOUT, max_bytes=MAX_RESPONSE_BYTES):
     """Send a request (no auto-redirect). Returns (status, parsed_dict). Non-2xx → (status, body).
     Pinned to the IPs assert_host_allowed(url) most recently validated for this host, if any (tried
     in order with fallback) — see module docstring (IP-pinning)."""
     data = body if isinstance(body, (bytes, bytearray)) else (body.encode() if isinstance(body, str) else None)
     req = urllib.request.Request(url, data=data, method=method, headers=headers or {})
+    # One absolute deadline covers connect, headers and body. The socket timeout is per operation, so
+    # the header phase is bounded by the worker wait in _open_within and the body by _read_bounded;
+    # the connector Lambda's 60s wall assumes the aligned HTTP timeout (at most 55s + 3s) is the whole request.
+    deadline = time.monotonic() + timeout
     try:
-        resp = _opener_for(urlparse(url).hostname).open(req, timeout=timeout)
-        return resp.getcode(), _parse(resp.read())
+        resp = _open_within(_opener_for(urlparse(url).hostname), req, timeout, deadline)
+        return resp.getcode(), _parse(_read_bounded(resp, max_bytes, deadline))
     except urllib.error.HTTPError as e:
+        # Keep the (status, body) contract for error responses: an oversized or slow error body is
+        # reported as such under its real HTTP status instead of replacing the status with a
+        # transport exception.
         try:
-            raw = e.read()
-        except Exception:  # noqa: BLE001
-            raw = b""
+            raw = _read_bounded(e, min(max_bytes, MAX_ERROR_BYTES), deadline)
+        except ResponseTooLarge:
+            return e.code, {"error": f"error body exceeded {MAX_ERROR_BYTES} bytes"}
+        except TimeoutError:
+            return e.code, {"error": "error body read deadline exceeded"}
+        except (OSError, http.client.HTTPException):
+            return e.code, {"error": "error body could not be read"}
         return e.code, _parse(raw)
diff --git a/agent/lambda/dynatrace_mcp.py b/agent/lambda/dynatrace_mcp.py
index cb68b58f..ec28ac64 100644
--- a/agent/lambda/dynatrace_mcp.py
+++ b/agent/lambda/dynatrace_mcp.py
@@ -16,11 +16,12 @@
 from cross_account import resolve_tool_name
 from datasource_http import (
     NotConnected, SsrfBlocked, assert_host_allowed, auth_headers, http_json, load_datasource,
-    set_request_conn,
+    query_probe, set_request_conn, request_timeout, finite_number, observed_payload, HTTP_TIMEOUT,
 )
 
 SLUG = "dynatrace"
 MAX_SERIES = 50
+MAX_POINTS = 1500
 _REL = re.compile(r"^(\d+)([smhdw])$")
 _NATIVE = re.compile(r"^now(-\d+[smhdw])?$")
 
@@ -46,7 +47,7 @@ def _headers(creds):
     """Api-Token scheme: an explicit token (authType bearer) becomes `Authorization: Api-Token …`;
     custom-header auth passes through unchanged (auth_headers refuses to set Authorization itself)."""
     h = dict(auth_headers(creds))
-    if creds.get("token"):
+    if creds.get("token") and creds.get("authType") in (None, "bearer"):
         h["Authorization"] = f"Api-Token {creds['token']}"
     return h
 
@@ -57,14 +58,15 @@ def _ds():
     return creds
 
 
-def _get(creds, path, params=None):
+def _get(creds, path, params=None, *, timeout=None):
     url = creds["endpoint"].rstrip("/") + path + ("?" + urlencode(params, doseq=True) if params else "")
-    status, data = http_json("GET", url, headers=_headers(creds))
+    status, data = http_json("GET", url, headers=_headers(creds),
+                             timeout=timeout if timeout is not None else request_timeout(creds))
     if status >= 400:
         detail = data.get("error") if isinstance(data, dict) else None
         msg = detail.get("message") if isinstance(detail, dict) else (detail or data.get("raw") if isinstance(data, dict) else data)
         raise _ApiError(f"Dynatrace HTTP {status}: {str(msg)[:300]}")
-    return data
+    return observed_payload(data, status)
 
 
 def dynatrace_query(args):
@@ -79,24 +81,47 @@ def dynatrace_query(args):
     if args.get("resolution"):
         params["resolution"] = str(args["resolution"])
     data = _get(_ds(), "/api/v2/metrics/query", params)
-    result = data.get("result", []) if isinstance(data, dict) else []
+    if isinstance(data, dict) and data.get("error") not in (None, ""):
+        return ok({"result": [], "truncated": False, "collectionStatus": "error"})
+    result = data.get("result") if isinstance(data, dict) else None
+    if not isinstance(result, list):
+        return ok({"result": [], "truncated": False, "collectionStatus": "unknown"})
     out = []
+    # Current Metrics v2 query responses have a null nextPageKey. Retain partial
+    # evidence if an older/compatible response advertises an unconsumed page.
+    truncated = len(result) > MAX_SERIES or bool(isinstance(data, dict) and data.get("nextPageKey"))
+    invalid = False
     for metric in result[:MAX_SERIES]:
-        if not isinstance(metric, dict):
+        if not isinstance(metric, dict) or not isinstance(metric.get("metricId"), str) or not isinstance(metric.get("data"), list):
+            invalid = True
             continue
+        truncated = truncated or len(metric["data"]) > MAX_SERIES
+        series = []
+        for entry in metric["data"][:MAX_SERIES]:
+            if (not isinstance(entry, dict) or not isinstance(entry.get("dimensions"), list)
+                    or not all(isinstance(value, str) for value in entry["dimensions"])
+                    or not isinstance(entry.get("timestamps"), list) or not isinstance(entry.get("values"), list)
+                    or len(entry["timestamps"]) != len(entry["values"])):
+                invalid = True
+                continue
+            timestamps, values = [], []
+            truncated = truncated or len(entry["values"]) > MAX_POINTS
+            for timestamp, value in zip(entry["timestamps"][:MAX_POINTS], entry["values"][:MAX_POINTS]):
+                if (not finite_number(timestamp, 8640000000000000)
+                        or value is not None and not finite_number(value)):
+                    invalid = True
+                    continue
+                timestamps.append(timestamp)
+                values.append(value)
+            series.append({"dimensions": entry["dimensions"], "timestamps": timestamps, "values": values})
         out.append({
-            "metricId": metric.get("metricId", ""),
-            "data": [
-                {
-                    "dimensions": d.get("dimensions", []),
-                    "timestamps": d.get("timestamps", []),
-                    "values": d.get("values", []),
-                }
-                for d in (metric.get("data") or [])[:MAX_SERIES]
-                if isinstance(d, dict)
-            ],
+            "metricId": metric["metricId"], "data": series,
         })
-    return ok({"result": out, "truncated": len(result) > MAX_SERIES,
+    http_status = getattr(data, "http_status", 200)
+    warned = bool(isinstance(data, dict) and data.get("warnings"))  # provider warnings (e.g. limited results) are not completeness proof
+    state = "unknown" if invalid or http_status not in (200, 206) else "partial" if truncated or warned or http_status == 206 else "ok" if out else "empty"
+    return ok({"result": out, "truncated": truncated,
+               "collectionStatus": state,
                "resolution": data.get("resolution") if isinstance(data, dict) else None})
 
 
@@ -118,21 +143,22 @@ def dynatrace_problems(args):
 
 def dynatrace_schema(args):
     """Metric ids (query-gen vocabulary for metricSelector)."""
-    data = _get(_ds(), "/api/v2/metrics", {"pageSize": "200", "fields": "metricId"})
+    data = _get(_ds(), "/api/v2/metrics", {"pageSize": "200", "fields": "metricId"}, timeout=HTTP_TIMEOUT)
     metrics = [m.get("metricId", "") for m in (data.get("metrics") or []) if isinstance(m, dict)] if isinstance(data, dict) else []
     metrics = [m for m in metrics if m]
     return ok({"metrics": metrics[:200], "truncated": bool(isinstance(data, dict) and data.get("nextPageKey"))})
 
 
 def dynatrace_health(args):
-    """Connectivity + token probe: metrics list pageSize=1 (requires metrics.read scope)."""
-    creds = load_datasource(SLUG)
-    assert_host_allowed(creds["endpoint"])
-    try:
-        _get(creds, "/api/v2/metrics", {"pageSize": "1"})
-        return ok({"ok": True})
-    except _ApiError as e:
-        return ok({"ok": False, "error": str(e)})
+    """Verify Metrics v2 query access with a one-minute, one-resolution query."""
+    def run():
+        creds = load_datasource(SLUG)
+        assert_host_allowed(creds["endpoint"])
+        return json.loads(dynatrace_query({
+            "query": "builtin:host.cpu.usage:avg", "start": "now-1m", "end": "now",
+            "resolution": "1m",
+        })["body"]).get("collectionStatus")
+    return ok(query_probe(run, "Dynatrace metric-query access could not be verified"))
 
 
 _TOOLS = {
diff --git a/agent/lambda/jaeger_mcp.py b/agent/lambda/jaeger_mcp.py
index 7edf3ded..b71a2656 100644
--- a/agent/lambda/jaeger_mcp.py
+++ b/agent/lambda/jaeger_mcp.py
@@ -17,8 +17,8 @@
 
 from cross_account import resolve_tool_name
 from datasource_http import (
-    NotConnected, SsrfBlocked, assert_host_allowed, auth_headers, health, http_json, load_datasource,
-    set_request_conn,
+    NotConnected, SsrfBlocked, assert_host_allowed, auth_headers, http_json, load_datasource,
+    query_probe, set_request_conn, request_timeout, observed_payload, HTTP_TIMEOUT, MAX_TRACE_RESPONSE_BYTES,
 )
 
 SLUG = "jaeger"
@@ -52,12 +52,14 @@ def _ds():
     return creds
 
 
-def _get(creds, path, params=None):
+def _get(creds, path, params=None, *, timeout=None, max_bytes=None):
     url = creds["endpoint"].rstrip("/") + path + ("?" + urlencode(params, doseq=True) if params else "")
-    status, data = http_json("GET", url, headers=auth_headers(creds))
+    extra = {"max_bytes": max_bytes} if max_bytes is not None else {}
+    status, data = http_json("GET", url, headers=auth_headers(creds),
+                             timeout=timeout if timeout is not None else request_timeout(creds), **extra)
     if status >= 400:  # Jaeger errors carry {"errors":[...]} — success = HTTP 2xx
         raise _ApiError(f"Jaeger HTTP {status}: {str(data.get('errors') or data.get('raw') or data)[:300]}")
-    return data
+    return observed_payload(data, status)
 
 
 def _byte_bound(obj):
@@ -88,16 +90,40 @@ def jaeger_search(args):
     except _ApiError as e:
         return err(str(e))
     params.setdefault("limit", "20")
-    params["limit"] = str(min(int(params["limit"]), MAX_TRACES))
+    if not re.fullmatch(r"\d{1,9}", str(params["limit"])):
+        return err("limit must be a positive integer")
+    if int(params["limit"]) == 0:
+        return err("limit must be a positive integer")
+    requested = max(1, min(int(params["limit"]), MAX_TRACES))
+    params["limit"] = str(requested)
     params["start"] = _parse_time_us(args.get("start"), 3600)
     params["end"] = _parse_time_us(args.get("end"))
     data = _get(_ds(), "/api/traces", params)
-    traces = data.get("data", []) if isinstance(data, dict) else []
+    if isinstance(data, dict) and data.get("errors"):
+        return ok({"traces": [], "truncated": False, "collectionStatus": "error"})
+    traces = data.get("data") if isinstance(data, dict) else None
+    if not isinstance(traces, list):
+        return ok({"traces": [], "truncated": False, "collectionStatus": "unknown"})
     # Compact rows for the console/agent: full span trees are huge — summarize per trace.
     rows = []
+    invalid = False
+    skipped = 0
     for t in traces[:MAX_TRACES]:
-        spans = t.get("spans", []) if isinstance(t, dict) else []
-        procs = t.get("processes", {}) if isinstance(t, dict) else {}
+        if (not isinstance(t, dict) or not isinstance(t.get("traceID"), str)
+                or not 1 <= len(t["traceID"]) <= 32 or not _HEX.fullmatch(t["traceID"])
+                or not isinstance(t.get("spans"), list)
+                or not isinstance(t.get("processes"), dict)):
+            invalid = True
+            continue
+        spans, procs = t["spans"], t["processes"]
+        if not spans:
+            skipped += 1
+            continue  # a trace with no spans is disclosed as a skipped row, never a missing one
+        if not all(isinstance(s, dict) and type(s.get("startTime")) is int and 0 <= s["startTime"] <= 8640000000000000000
+                   and type(s.get("duration")) is int and 0 <= s["duration"] <= 8640000000000000000
+                   and isinstance(s.get("operationName"), str) for s in spans):
+            invalid = True
+            continue
         root = min(spans, key=lambda s: s.get("startTime", 0)) if spans else {}
         dur_us = max((s.get("startTime", 0) + s.get("duration", 0)) for s in spans) - root.get("startTime", 0) if spans else 0
         svc = ""
@@ -112,21 +138,33 @@ def jaeger_search(args):
             "durationMs": round(dur_us / 1000, 1),
             "startTime": root.get("startTime", 0),
         })
+    # Reaching the requested limit is saturation, not loss: Jaeger returns no marker, so more traces
+    # may exist and the result is `partial`; a local row/byte drop sets `truncated`.
+    saturated = len(traces) >= requested
+    http_status = getattr(data, "http_status", 200)
+    dropped = len(traces) > MAX_TRACES
+    state = "unknown" if invalid or http_status not in (200, 206) else "partial" if saturated or dropped or skipped or http_status == 206 else "ok" if rows else "empty"
     payload, btr = _byte_bound({"traces": rows})
-    return ok(payload if btr else {"truncated": len(traces) > MAX_TRACES, **payload})
+    if btr:
+        payload.pop("preview", None)  # never echo provider content, even as a bounded preview
+    emitted = payload.get("traces", [])
+    return ok({**payload, "traces": emitted, "truncated": btr or dropped,
+               "limitSaturated": saturated,
+               "collectionStatus": "unknown" if invalid or http_status not in (200, 206) else "partial" if btr else state,
+               "requestedLimit": requested, "returnedCount": len(emitted), "skippedCount": skipped})
 
 
 def jaeger_get_trace(args):
     tid = (args.get("trace_id") or "").strip()
     if not tid or not _HEX.match(tid):
         return err("trace_id must be a hex string")
-    data = _get(_ds(), f"/api/traces/{quote(tid, safe='')}")
+    data = _get(_ds(), f"/api/traces/{quote(tid, safe='')}", max_bytes=MAX_TRACE_RESPONSE_BYTES)
     payload, btr = _byte_bound(data if isinstance(data, dict) else {"trace": data})
     return ok(payload)
 
 
 def jaeger_services(args):
-    data = _get(_ds(), "/api/services")
+    data = _get(_ds(), "/api/services", timeout=HTTP_TIMEOUT)
     services = data.get("data", []) if isinstance(data, dict) else []
     return ok({"services": services if isinstance(services, list) else []})
 
@@ -135,22 +173,28 @@ def jaeger_operations(args):
     svc = (args.get("service") or "").strip()
     if not svc:
         return err("service required")
-    data = _get(_ds(), f"/api/services/{quote(svc, safe='')}/operations")
+    data = _get(_ds(), f"/api/services/{quote(svc, safe='')}/operations", timeout=HTTP_TIMEOUT)
     ops = data.get("data", []) if isinstance(data, dict) else []
     return ok({"operations": ops if isinstance(ops, list) else []})
 
 
 def jaeger_schema(args):
     """Service names (the query-gen 'vocabulary' — a Jaeger search is service-anchored)."""
-    data = _get(_ds(), "/api/services")
+    data = _get(_ds(), "/api/services", timeout=HTTP_TIMEOUT)
     services = data.get("data", []) if isinstance(data, dict) else []
     services = services if isinstance(services, list) else []
     return ok({"services": services[:100], "truncated": len(services) > 100})
 
 
 def jaeger_health(args):
-    """Connectivity probe for the pre-save Test / status badge: GET /api/services."""
-    return ok(health(load_datasource(SLUG), "/api/services"))
+    """Verify a recognizable query response, not an arbitrary HTTP 200."""
+    def run():
+        end = int(time.time())
+        return json.loads(jaeger_search({
+            "query": "service=awsops-connection-probe&limit=1",
+            "start": str(end - 60), "end": str(end),
+        })["body"]).get("collectionStatus")
+    return ok(query_probe(run, "Jaeger query access could not be verified"))
 
 
 _TOOLS = {
diff --git a/agent/lambda/loki_mcp.py b/agent/lambda/loki_mcp.py
index a514fdf7..64eb49ae 100644
--- a/agent/lambda/loki_mcp.py
+++ b/agent/lambda/loki_mcp.py
@@ -15,8 +15,9 @@
 
 from cross_account import resolve_tool_name
 from datasource_http import (
-    NotConnected, SsrfBlocked, assert_host_allowed, auth_headers, health, http_json, load_datasource,
-    set_request_conn,
+    NotConnected, SsrfBlocked, assert_host_allowed, auth_headers, http_json, load_datasource,
+    query_probe, set_request_conn, request_timeout, observed_payload, finite_number, HTTP_TIMEOUT,
+    MAX_TRACE_RESPONSE_BYTES,
 )
 
 SLUG = "loki"
@@ -62,14 +63,86 @@ def _ds():
     return creds
 
 
-def _get(creds, path, params):
+def _get(creds, path, params, *, verify="query", timeout=None, max_bytes=None):
     url = creds["endpoint"].rstrip("/") + path + ("?" + urlencode(params, doseq=True) if params else "")
-    status, data = http_json("GET", url, headers=_headers(creds))
+    extra = {"max_bytes": max_bytes} if max_bytes is not None else {}
+    status, data = http_json("GET", url, headers=_headers(creds),
+                             timeout=timeout if timeout is not None else request_timeout(creds), **extra)
     if status >= 400:
         raise _ApiError(f"Loki HTTP {status}: {str(data.get('raw') or data.get('error') or data)[:300]}")
+    if verify == "raw":  # buildinfo has no status/data envelope
+        if isinstance(data, dict):
+            return observed_payload(data, status)
+        raise _ApiError("Loki response could not be verified")
     if isinstance(data, dict) and data.get("status") and data.get("status") != "success":
         raise _ApiError(f"Loki query failed: {data.get('error', 'unknown')}")
-    return data.get("data") if isinstance(data, dict) else data
+    if verify == "list" and isinstance(data, dict) and data.get("status") == "success" and "data" not in data:
+        data = {**data, "data": []}  # older Loki omits `data` on empty label results
+    if not isinstance(data, dict) or data.get("status") != "success" or "data" not in data:
+        raise _ApiError("Loki response could not be verified")
+    result = observed_payload(data["data"], status)
+    if isinstance(result, dict):
+        result.source_partial = bool(data.get("warnings") or data.get("partial") or data.get("truncated"))
+        result.source_unverified = (
+            "warnings" in data and (not isinstance(data["warnings"], list)
+                                   or not all(isinstance(value, str) for value in data["warnings"]))
+            or any(key in data and type(data[key]) is not bool for key in ("partial", "truncated"))
+        )
+    return result
+
+
+def _valid_row(row, kind):
+    if not isinstance(row, dict):
+        return False
+    labels = row.get("stream" if kind == "streams" else "metric")
+    if not isinstance(labels, dict) or not all(isinstance(key, str) and isinstance(value, str) for key, value in labels.items()):
+        return False
+    samples = row.get("values") if kind != "vector" else [row.get("value")]
+    if not isinstance(samples, list):
+        return False
+    for pair in samples[:MAX_LINES_PER_STREAM]:
+        if not isinstance(pair, list) or len(pair) != 2 or type(pair[0]) not in (str, int, float):
+            return False
+        try:
+            if not finite_number(float(pair[0]), 8640000000000000000000 if kind == "streams" else 8640000000000):
+                return False
+        except (ValueError, OverflowError):
+            return False
+        if kind == "streams":
+            if not isinstance(pair[1], str):
+                return False
+        elif (type(pair[1]) not in (str, int, float)
+              or isinstance(pair[1], str) and not re.fullmatch(r"(?:NaN|[+-]?Inf|[+-]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?)", pair[1])
+              or not isinstance(pair[1], str) and not finite_number(pair[1])):
+            return False
+    return True
+
+
+def _query_response(data, requested=None):
+    """Normalize a Loki query envelope. `requested` is the caller's `limit`: Loki returns at most
+    that many log lines with no marker, so reaching it is disclosed as `limitSaturated` and the
+    result is `partial` — more lines may exist (mirrors Jaeger's requested/returned contract)."""
+    valid = (isinstance(data, dict) and data.get("resultType") in ("streams", "matrix", "vector")
+             and isinstance(data.get("result"), list) and not getattr(data, "source_unverified", False))
+    kind = data.get("resultType") if isinstance(data, dict) else None
+    source_rows = data.get("result") if isinstance(data, dict) else None
+    source_rows = source_rows if isinstance(source_rows, list) else []
+    retained = source_rows[:MAX_STREAMS]
+    rows = [row for row in retained if _valid_row(row, kind)] if kind in ("streams", "matrix", "vector") else []
+    valid = valid and len(rows) == len(retained)
+    bounded, truncated = _bound({"resultType": kind, "result": rows})
+    truncated = truncated or len(source_rows) > MAX_STREAMS
+    upstream_lines = sum(len(row.get("values") or []) for row in rows) if kind == "streams" else 0
+    emitted = bounded.get("result", []) if isinstance(bounded, dict) else []
+    returned = sum(len(row.get("values") or []) for row in emitted) if kind == "streams" else len(emitted)
+    saturated = kind == "streams" and bool(requested) and upstream_lines >= requested
+    status = getattr(data, "http_status", 200)
+    state = ("unknown" if not valid or status not in (200, 206) else
+             "partial" if truncated or saturated or status == 206 or getattr(data, "source_partial", False) else
+             "ok" if data["result"] else "empty")
+    return ok({"truncated": truncated, **(bounded if isinstance(bounded, dict) else {"result": None}),
+               "collectionStatus": state,
+               **({"requestedLimit": requested, "returnedCount": returned, "limitSaturated": saturated} if requested else {})})
 
 
 def _bound(data):
@@ -115,9 +188,8 @@ def loki_query_range(args):
               "limit": str(args.get("limit") or DEFAULT_LIMIT), "direction": args.get("direction") or "backward"}
     if args.get("step"):
         params["step"] = str(args["step"])
-    data = _get(_ds(), "/loki/api/v1/query_range", params)
-    bounded, tr = _bound(data)
-    return ok({"truncated": tr, **(bounded if isinstance(bounded, dict) else {"result": bounded})})
+    data = _get(_ds(), "/loki/api/v1/query_range", params, max_bytes=MAX_TRACE_RESPONSE_BYTES)
+    return _query_response(data, _requested(params["limit"]))
 
 
 def loki_query(args):
@@ -127,13 +199,17 @@ def loki_query(args):
     params = {"query": query, "limit": str(args.get("limit") or DEFAULT_LIMIT)}
     if args.get("time"):
         params["time"] = _parse_time_ns(args.get("time"))
-    data = _get(_ds(), "/loki/api/v1/query", params)
-    bounded, tr = _bound(data)
-    return ok({"truncated": tr, **(bounded if isinstance(bounded, dict) else {"result": bounded})})
+    data = _get(_ds(), "/loki/api/v1/query", params, max_bytes=MAX_TRACE_RESPONSE_BYTES)
+    return _query_response(data, _requested(params["limit"]))
+
+
+def _requested(limit):
+    """The positive integer limit actually sent to Loki, or None when it was not a usable number."""
+    return int(limit) if re.fullmatch(r"[1-9]\d{0,8}", str(limit)) else None
 
 
 def loki_labels(args):
-    data = _get(_ds(), "/loki/api/v1/labels", {})
+    data = _get(_ds(), "/loki/api/v1/labels", {}, verify="list", timeout=HTTP_TIMEOUT)
     names = data if isinstance(data, list) else []
     return ok({"labels": names[:1000], "truncated": len(names) > 1000})
 
@@ -142,7 +218,7 @@ def loki_label_values(args):
     label = (args.get("label") or "").strip()
     if not label:
         return err("label required")
-    data = _get(_ds(), f"/loki/api/v1/label/{quote(label, safe='')}/values", {})
+    data = _get(_ds(), f"/loki/api/v1/label/{quote(label, safe='')}/values", {}, verify="list", timeout=HTTP_TIMEOUT)
     values = data if isinstance(data, list) else []
     return ok({"values": values[:1000], "truncated": len(values) > 1000})
 
@@ -150,12 +226,14 @@ def loki_label_values(args):
 def loki_schema(args):
     creds = _ds()
     try:  # best-effort server version for version-aware LogQL
-        bi = _get(creds, "/loki/api/v1/status/buildinfo", {})
-        version = bi.get("version") if isinstance(bi, dict) else None
+        bi = _get(creds, "/loki/api/v1/status/buildinfo", {}, verify="raw", timeout=HTTP_TIMEOUT)
+        # Loki itself serves buildinfo bare; a gateway may wrap it in a status/data envelope.
+        info = bi.get("data") if isinstance(bi, dict) and isinstance(bi.get("data"), dict) and "version" in bi["data"] else bi
+        version = info.get("version") if isinstance(info, dict) else None
     except _ApiError:
         version = None
     try:
-        labels = _get(creds, "/loki/api/v1/labels", {})
+        labels = _get(creds, "/loki/api/v1/labels", {}, verify="list", timeout=HTTP_TIMEOUT)
     except _ApiError:
         labels = None
     labels_ok = isinstance(labels, list)  # a FAILED label fetch is not an empty schema
@@ -173,8 +251,17 @@ def loki_schema(args):
 
 
 def loki_health(args):
-    """Connectivity probe for the pre-save Test / status badge: GET /ready."""
-    return ok(health(load_datasource(SLUG), "/ready"))
+    """Verify the tenant-scoped query endpoint with a one-minute, one-line search, judged by the
+    same `_query_response` verdict the query tools use."""
+    def run():
+        end = int(time.time())
+        data = _get(_ds(), "/loki/api/v1/query_range", {
+            "query": '{service_name="awsops-connection-probe"}',
+            "start": str((end - 60) * 1_000_000_000), "end": str(end * 1_000_000_000),
+            "limit": "1",
+        })
+        return json.loads(_query_response(data, 1)["body"]).get("collectionStatus")
+    return ok(query_probe(run, "Loki tenant query access could not be verified"))
 
 
 _TOOLS["loki_health"] = loki_health
@@ -214,4 +301,4 @@ def ok(body):
 
 
 def err(msg):
-    return {"statusCode": 400, "body": json.dumps({"error": msg})}
+    return {"statusCode": 400, "body": json.dumps({"error": msg, "collectionStatus": "error"})}
diff --git a/agent/lambda/mimir_mcp.py b/agent/lambda/mimir_mcp.py
index 9c8746c1..122ec361 100644
--- a/agent/lambda/mimir_mcp.py
+++ b/agent/lambda/mimir_mcp.py
@@ -9,12 +9,13 @@
 import math
 import re
 import time
-from urllib.parse import urlencode
+from urllib.parse import urlencode, urlparse
 
 from cross_account import resolve_tool_name
 from datasource_http import (
-    NotConnected, SsrfBlocked, assert_host_allowed, auth_headers, health, http_json, load_datasource,
-    set_request_conn,
+    NotConnected, SsrfBlocked, assert_host_allowed, auth_headers, http_json, load_datasource,
+    query_probe, set_request_conn,
+    execution_timeout, HTTP_TIMEOUT,
 )
 
 SLUG = "mimir"
@@ -61,8 +62,19 @@ def _ds():
 
 
 def _get(creds, path, params, http_timeout=None, *, with_status=False):
+    if path.startswith(BASE):
+        prefix = creds.get("apiPrefix")
+        if prefix is None:  # absent or stored null = default mount; an explicit "" = API at the root
+            prefix = "/prometheus"
+        if not isinstance(prefix, str) or len(prefix) > 128 or not re.fullmatch(r"(?:/[A-Za-z0-9_-]+)*", prefix):
+            raise _ApiError("invalid Mimir API prefix")
+        # An API base already ending in the configured prefix must not duplicate it.
+        endpoint_path = urlparse(creds["endpoint"]).path.rstrip("/")
+        if prefix and endpoint_path.endswith(prefix):
+            prefix = ""
+        path = prefix + "/api/v1" + path[len(BASE):]
     url = creds["endpoint"].rstrip("/") + path + ("?" + urlencode(params, doseq=True) if params else "")
-    kwargs = {"headers": _headers(creds)}
+    kwargs = {"headers": _headers(creds), "timeout": HTTP_TIMEOUT}
     if http_timeout is not None:
         kwargs["timeout"] = http_timeout
     status, data = http_json("GET", url, **kwargs)
@@ -185,29 +197,15 @@ def _query_result(observed, *, allow_scalar=False):
 
 
 
-def _timeout_param(v):
-    """Optional Prometheus-API `timeout` (e.g. "5s"), clamped to 1..60s. Same knob prometheus_mcp.py grew:
-    the connector bounds response SIZE but had no execution bound, which matters now that a caller can run
-    model-written PromQL through it (review — the diag-signal dry run passes 5s for mimir too)."""
-    # No digit cap: `\d{1,3}` made "1000" or "3600s" fall through to None, i.e. NO bound — a clamp that
-    # fails open for exactly the values that need clamping (review MINOR). Only non-numeric input is None.
-    # `v or ""` turns numeric 0 into "" (falsy), which then fails the match and returns None —
-    # i.e. NO bound for exactly the value that most needs clamping. Only skip on a true absence.
-    if v is None or v == "":
-        return None
-    m = re.fullmatch(r"\s*(\d+)s?\s*", str(v))
-    return f"{max(1, min(60, int(m.group(1))))}s" if m else None
-
-
 def mimir_query(args):
     query = (args.get("query") or "").strip()
     if not query:
         return err("query (PromQL) required")
     params = {"query": query, "time": _parse_time(args.get("time"))}
-    timeout = _timeout_param(args.get("timeout"))
-    if timeout:
-        params["timeout"] = timeout
-    return _query_result(_get(_ds(), f"{BASE}/query", params, with_status=True), allow_scalar=True)
+    creds = _ds()
+    timeout = execution_timeout(creds, args.get("timeout"))
+    params["timeout"] = f"{timeout}s"
+    return _query_result(_get(creds, f"{BASE}/query", params, http_timeout=timeout + 3, with_status=True), allow_scalar=True)
 
 
 def mimir_query_range(args):
@@ -216,10 +214,10 @@ def mimir_query_range(args):
         return err("query (PromQL) required")
     params = {"query": query, "start": _parse_time(args.get("start"), 3600),
               "end": _parse_time(args.get("end")), "step": str(args.get("step") or "60")}
-    timeout = _timeout_param(args.get("timeout"))
-    if timeout:
-        params["timeout"] = timeout
-    return _query_result(_get(_ds(), f"{BASE}/query_range", params, with_status=True))
+    creds = _ds()
+    timeout = execution_timeout(creds, args.get("timeout"))
+    params["timeout"] = f"{timeout}s"
+    return _query_result(_get(creds, f"{BASE}/query_range", params, http_timeout=timeout + 3, with_status=True))
 
 
 def _list_result(observed, field, limit, kind):
@@ -346,8 +344,11 @@ def mimir_metric_meta(args):
 
 
 def mimir_health(args):
-    """Connectivity probe for the pre-save Test / status badge: GET /ready."""
-    return ok(health(load_datasource(SLUG), "/ready"))
+    """Verify tenant-scoped query access, independently of process readiness. An empty or
+    upstream-warned (`partial`) result still proves access."""
+    return ok(query_probe(
+        lambda: json.loads(mimir_query({"query": "vector(1)", "timeout": "1s"})["body"]).get("collectionStatus"),
+        "Mimir tenant query access could not be verified"))
 
 
 _TOOLS["mimir_health"] = mimir_health
diff --git a/agent/lambda/prometheus_mcp.py b/agent/lambda/prometheus_mcp.py
index 1cc79f1b..def201b6 100644
--- a/agent/lambda/prometheus_mcp.py
+++ b/agent/lambda/prometheus_mcp.py
@@ -21,10 +21,12 @@
     SsrfBlocked,
     assert_host_allowed,
     auth_headers,
-    health,
     http_json,
     load_datasource,
+    query_probe,
     set_request_conn,
+    execution_timeout,
+    HTTP_TIMEOUT,
 )
 
 SLUG = "prometheus"
@@ -67,7 +69,7 @@ def _ds():
 
 def _get(creds, path, params, http_timeout=None, *, with_status=False):
     url = creds["endpoint"].rstrip("/") + path + ("?" + urlencode(params, doseq=True) if params else "")
-    kwargs = {"headers": auth_headers(creds)}
+    kwargs = {"headers": auth_headers(creds), "timeout": HTTP_TIMEOUT}
     if http_timeout is not None:
         kwargs["timeout"] = http_timeout
     status, data = http_json("GET", url, **kwargs)
@@ -194,28 +196,15 @@ def _query_result(observed, *, allow_scalar=False):
 
 
 
-def _timeout_param(v):
-    """Optional Prometheus API `timeout` (e.g. "5s"), clamped to 1..60s. The connector bounds response SIZE
-    but had no execution bound, which matters now that a caller can run model-written PromQL (review)."""
-    # No digit cap: `\d{1,3}` made "1000" or "3600s" fall through to None, i.e. NO bound — a clamp that
-    # fails open for exactly the values that need clamping (review MINOR). Only non-numeric input is None.
-    # `v or ""` turns numeric 0 into "" (falsy), which then fails the match and returns None —
-    # i.e. NO bound for exactly the value that most needs clamping. Only skip on a true absence.
-    if v is None or v == "":
-        return None
-    m = re.fullmatch(r"\s*(\d+)s?\s*", str(v))
-    return f"{max(1, min(60, int(m.group(1))))}s" if m else None
-
-
 def prometheus_query(args):
     query = (args.get("query") or "").strip()
     if not query:
         return err("query (PromQL) required")
     params = {"query": query, "time": _parse_time(args.get("time"))}
-    timeout = _timeout_param(args.get("timeout"))
-    if timeout:
-        params["timeout"] = timeout
-    return _query_result(_get(_ds(), "/api/v1/query", params, with_status=True), allow_scalar=True)
+    creds = _ds()
+    timeout = execution_timeout(creds, args.get("timeout"))
+    params["timeout"] = f"{timeout}s"
+    return _query_result(_get(creds, "/api/v1/query", params, http_timeout=timeout + 3, with_status=True), allow_scalar=True)
 
 
 def prometheus_query_range(args):
@@ -226,10 +215,10 @@ def prometheus_query_range(args):
     end = _parse_time(args.get("end"))
     step = str(args.get("step") or "60").strip()
     params = {"query": query, "start": start, "end": end, "step": step}
-    timeout = _timeout_param(args.get("timeout"))
-    if timeout:
-        params["timeout"] = timeout
-    return _query_result(_get(_ds(), "/api/v1/query_range", params, with_status=True))
+    creds = _ds()
+    timeout = execution_timeout(creds, args.get("timeout"))
+    params["timeout"] = f"{timeout}s"
+    return _query_result(_get(creds, "/api/v1/query_range", params, http_timeout=timeout + 3, with_status=True))
 
 
 def _list_result(observed, field, limit, kind):
@@ -297,8 +286,11 @@ def prometheus_schema(args):
 
 
 def prometheus_health(args):
-    """Connectivity probe for the pre-save Test / status badge: GET /-/healthy."""
-    return ok(health(_ds(), "/-/healthy"))
+    """Verify the actual authenticated query API using a bounded scalar query. A successful query
+    whose result is empty or carries an upstream warning (`partial`) still proves access."""
+    return ok(query_probe(
+        lambda: json.loads(prometheus_query({"query": "vector(1)", "timeout": "1s"})["body"]).get("collectionStatus"),
+        "Prometheus query access could not be verified"))
 
 
 def prometheus_metric_meta(args):
diff --git a/agent/lambda/tempo_mcp.py b/agent/lambda/tempo_mcp.py
index a3672ec8..ecb6927c 100644
--- a/agent/lambda/tempo_mcp.py
+++ b/agent/lambda/tempo_mcp.py
@@ -18,8 +18,8 @@
 
 from cross_account import resolve_tool_name
 from datasource_http import (
-    NotConnected, SsrfBlocked, assert_host_allowed, auth_headers, health, http_json, load_datasource,
-    set_request_conn,
+    MAX_TRACE_RESPONSE_BYTES, NotConnected, SsrfBlocked, assert_host_allowed, auth_headers, http_json,
+    load_datasource, query_probe, set_request_conn, request_timeout,
 )
 
 SLUG = "tempo"
@@ -76,9 +76,12 @@ def _ds():
     return creds
 
 
-def _get(creds, path, params=None, *, timeout=None, with_status=False):
+def _get(creds, path, params=None, *, timeout=None, with_status=False, max_bytes=None):
     url = creds["endpoint"].rstrip("/") + path + ("?" + urlencode(params, doseq=True) if params else "")
-    request_options = {"timeout": timeout} if timeout is not None else {}
+    # An explicit discovery budget wins over the saved query bound (timeoutS).
+    request_options = {"timeout": timeout if timeout is not None else request_timeout(creds)}
+    if max_bytes is not None:
+        request_options["max_bytes"] = max_bytes
     status, data = http_json("GET", url, headers=_headers(creds), **request_options)
     if status >= 400:  # Tempo has no envelope status → HTTP 2xx is success
         detail = (data.get("raw") or data.get("error") or data) if isinstance(data, dict) else "non-object error response"
@@ -252,7 +255,7 @@ def _trace_projection(data, requested_trace_id):
     return out if out[key] else None
 
 
-def tempo_search(args):
+def tempo_search(args, bare_empty=False):
     query = (args.get("query") or "").strip()
     if not query:
         return err("query (TraceQL) required")
@@ -274,7 +277,8 @@ def tempo_search(args):
     # Require a recognizable message. Bare {} is unverified, not universal API malformation.
     metrics = data.get("metrics") if isinstance(data, dict) else None
     known_metrics = isinstance(metrics, dict) and (not metrics or any(key in metrics for key in _SEARCH_FIELDS))
-    recognizable = isinstance(data, dict) and ("traces" in data or known_metrics) and "raw" not in data
+    # Only the connection test accepts an idle tenant's bare {} 200 as an empty search.
+    recognizable = isinstance(data, dict) and ("traces" in data or known_metrics or bare_empty and data == {}) and "raw" not in data
     raw = data.get("traces", []) if recognizable else None
     traces = raw[:MAX_TRACES] if isinstance(raw, list) else []
     truncated = isinstance(raw, list) and len(raw) > MAX_TRACES
@@ -314,7 +318,7 @@ def tempo_get_trace(args):
     tid = (args.get("trace_id") or "").strip()
     if not tid or not _HEX.match(tid):
         return err("trace_id must be a hex string")
-    data = _get(_ds(), f"/api/traces/{quote(tid, safe='')}")
+    data = _get(_ds(), f"/api/traces/{quote(tid, safe='')}", max_bytes=MAX_TRACE_RESPONSE_BYTES)
     if isinstance(data, dict) and (data.get("status") == "error"
             or any(data.get(key) not in (None, "") for key in ("error", "errorType", "exception"))):
         return err("Tempo trace fetch returned an error")
@@ -542,8 +546,14 @@ def tempo_schema(args):
 
 
 def tempo_health(args):
-    """Connectivity probe for the pre-save Test / status badge: GET /ready."""
-    return ok(health(load_datasource(SLUG), "/ready"))
+    """Verify query access using a one-minute, one-trace search, not process readiness."""
+    def run():
+        end = int(time.time())
+        return json.loads(tempo_search({
+            "query": '{ resource.service.name = "awsops-connection-probe" }',
+            "start": str(end - 60), "end": str(end), "limit": "1",
+        }, bare_empty=True)["body"]).get("collectionStatus")
+    return ok(query_probe(run, "Tempo tenant query access could not be verified"))
 
 
 _TOOLS["tempo_health"] = tempo_health
diff --git a/agent/lambda/test_clickhouse_mcp.py b/agent/lambda/test_clickhouse_mcp.py
index 298decd4..7b466f7c 100644
--- a/agent/lambda/test_clickhouse_mcp.py
+++ b/agent/lambda/test_clickhouse_mcp.py
@@ -7,6 +7,7 @@
 
 sys.path.insert(0, os.path.dirname(__file__))
 import clickhouse_mcp as ch  # noqa: E402
+import datasource_http  # noqa: E402
 
 DS = {"endpoint": "http://ch:8123", "username": "default", "password": "pw"}
 
@@ -113,8 +114,8 @@ def setUp(self):
     def test_query_builds_request(self):
         captured = {}
 
-        def fake_http(method, url, headers=None, body=None, timeout=None):
-            captured.update(method=method, url=url, headers=headers, body=body)
+        def fake_http(method, url, headers=None, body=None, timeout=None, max_bytes=None):
+            captured.update(method=method, url=url, headers=headers, body=body, max_bytes=max_bytes)
             return 200, {"data": [{"x": 1}, {"x": 2}, {"x": 3}], "rows": 3}
 
         with mock.patch.object(ch, "http_json", side_effect=fake_http):
@@ -131,12 +132,14 @@ def fake_http(method, url, headers=None, body=None, timeout=None):
         self.assertNotIn("output_format_json_quote_64bit_integers", captured["url"])
         self.assertIn("FORMAT JSON", captured["body"])
         self.assertEqual(captured["headers"]["Authorization"][:6], "Basic ")
+        # wide max_result_rows-bounded results get the larger transport budget
+        self.assertEqual(captured["max_bytes"], datasource_http.MAX_TRACE_RESPONSE_BYTES)
 
     def test_database_setting_appends_validated_param(self):
         # gap L203: a per-datasource default database rides the conn config → &database=
         captured = {}
 
-        def fake_http(method, url, headers=None, body=None, timeout=None):
+        def fake_http(method, url, headers=None, body=None, timeout=None, max_bytes=None):
             captured.update(url=url)
             return 200, {"data": []}
 
@@ -151,7 +154,7 @@ def fake_http(method, url, headers=None, body=None, timeout=None):
     def test_execution_bound_defaults_to_10_and_conn_timeoutS_overrides(self):
         captured = {}
 
-        def fake_http(method, url, headers=None, body=None, timeout=None):
+        def fake_http(method, url, headers=None, body=None, timeout=None, max_bytes=None):
             captured.update(url=url, timeout=timeout)
             return 200, {"data": []}
 
@@ -211,7 +214,7 @@ def test_bound_relaxing_settings_rejected_but_benign_settings_pass(self):
     def test_configured_timeout_is_a_ceiling_not_a_default(self):
         captured = {}
 
-        def fake_http(method, url, headers=None, body=None, timeout=None):
+        def fake_http(method, url, headers=None, body=None, timeout=None, max_bytes=None):
             captured.update(url=url)
             return 200, {"data": []}
 
diff --git a/agent/lambda/test_collection_markers.py b/agent/lambda/test_collection_markers.py
index d27ba0b6..146d5941 100644
--- a/agent/lambda/test_collection_markers.py
+++ b/agent/lambda/test_collection_markers.py
@@ -70,12 +70,16 @@ def test_catalog_guides_all_collection_marker_tools(self):
                         for target in catalog.TARGETS.values() for tool in target["tools"]}
         names = {f"{kind}_{action}" for kind in ("prometheus", "mimir")
                  for action in ("query", "query_range", "labels", "series")}
-        names.update(("tempo_search", "clickhouse_query", "clickhouse_tables", "clickhouse_describe"))
+        names.update(("tempo_search", "clickhouse_query", "clickhouse_tables", "clickhouse_describe", "loki_query", "loki_query_range"))
         for name in names:
             with self.subTest(tool=name):
                 for marker in ("collectionStatus", "ok", "empty", "partial", "unknown", "error"):
                     self.assertIn(marker, descriptions[name])
 
+    def test_loki_errors_carry_the_error_marker(self):
+        body = json.loads(importlib.import_module("loki_mcp").err("synthetic failure")["body"])
+        self.assertEqual(body["collectionStatus"], "error")
+
     def test_native_histogram_output_is_explicitly_unsupported_and_never_echoed(self):
         for kind in ("prometheus", "mimir"):
             for result_type, field, tool in (("vector", "histogram", "query"), ("matrix", "histograms", "query_range")):
diff --git a/agent/lambda/test_datadog_mcp.py b/agent/lambda/test_datadog_mcp.py
new file mode 100644
index 00000000..ddf5e889
--- /dev/null
+++ b/agent/lambda/test_datadog_mcp.py
@@ -0,0 +1,106 @@
+"""Datadog query-access and result-loss regressions; provider transport is mocked."""
+import json
+import sys
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).parent))
+import datadog_mcp as dd
+
+
+def body(value):
+    return json.loads(value["body"])
+
+
+CREDS = {"endpoint": "https://provider.example.test", "apiKey": "synthetic-api", "appKey": "synthetic-app"}
+
+
+def test_api_key_validation_alone_cannot_pass_when_metric_query_is_denied():
+    with patch.object(dd, "load_datasource", return_value=CREDS), patch.object(dd, "assert_host_allowed"), \
+            patch.object(dd, "_get", side_effect=[{"valid": True}, dd._ApiError("synthetic-app denied")]) as request:
+        result = body(dd.datadog_health({}))
+    assert result["ok"] is False
+    assert request.call_args_list[1].args[1] == "/api/v1/query"
+    assert "synthetic-app" not in json.dumps(result)
+
+
+def test_health_accepts_a_successfully_queried_empty_probe_and_bounds_its_window():
+    with patch.object(dd, "load_datasource", return_value=CREDS), patch.object(dd, "assert_host_allowed"), \
+            patch.object(dd.time, "time", return_value=1000), \
+            patch.object(dd, "_get", side_effect=[{"valid": True}, {"status": "ok", "series": []}]) as request:
+        probe = body(dd.datadog_health({}))
+        assert probe["ok"] is True and probe["verified"] == ["query"] and isinstance(probe["latency_ms"], int)
+    params = request.call_args_list[1].args[2]
+    assert params["from"] == "940" and params["to"] == "1000"
+    assert "{*}" not in params["query"]
+
+
+def test_invalid_api_key_never_starts_a_metric_probe():
+    with patch.object(dd, "load_datasource", return_value=CREDS), patch.object(dd, "assert_host_allowed"), \
+            patch.object(dd, "_get", return_value={"valid": False}) as request:
+        assert body(dd.datadog_health({}))["ok"] is False
+    request.assert_called_once()
+
+
+def test_health_does_not_accept_an_unverified_query_envelope():
+    with patch.object(dd, "load_datasource", return_value=CREDS), patch.object(dd, "assert_host_allowed"), \
+            patch.object(dd, "_get", side_effect=[{"valid": True}, {"status": "error", "series": []}]):
+        assert body(dd.datadog_health({}))["ok"] is False
+
+
+def test_bearer_connection_tests_the_metric_operation_without_api_key_validation():
+    creds = {"endpoint": CREDS["endpoint"], "authType": "bearer", "token": "synthetic-bearer"}
+    with patch.object(dd, "load_datasource", return_value=creds), patch.object(dd, "assert_host_allowed"), \
+            patch.object(dd, "_get", return_value={"status": "ok", "series": []}) as request:
+        probe = body(dd.datadog_health({}))
+        assert probe["ok"] is True and probe["verified"] == ["query"] and isinstance(probe["latency_ms"], int)
+    request.assert_called_once()
+    assert request.call_args.args[1] == "/api/v1/query"
+
+
+def test_health_requests_share_one_deadline():
+    for cost, passed in ((15, True), (25, False)):  # validate consumes `cost` of a 20s budget
+        clock, calls = [100.0], []
+
+        def fake_get(creds, path, params=None, *, timeout=None):
+            calls.append((path, timeout))
+            clock[0] += cost if path == "/api/v1/validate" else 0
+            return {"valid": True} if path == "/api/v1/validate" else {"status": "ok", "series": []}
+        with patch.object(dd, "load_datasource", return_value={**CREDS, "timeoutS": 20}), \
+                patch.object(dd, "assert_host_allowed"), patch.object(dd, "_get", side_effect=fake_get), \
+                patch.object(dd.time, "monotonic", side_effect=lambda: clock[0]):
+            assert body(dd.datadog_health({}))["ok"] is passed
+        assert calls[0] == ("/api/v1/validate", 20)
+        assert calls[1:] == ([("/api/v1/query", 5)] if passed else [])
+
+
+def test_point_limit_is_disclosed_and_preserves_other_series():
+    points = [[1000 + i, i] for i in range(dd.MAX_POINTS + 1)]
+    response = {"status": "ok", "series": [
+        {"metric": "large", "pointlist": points}, {"metric": "small", "pointlist": [[1000, 0]]},
+    ]}
+    with patch.object(dd, "_ds", return_value=CREDS), patch.object(dd, "_get", return_value=response):
+        result = body(dd.datadog_query({"query": "avg:fixture.metric{*}"}))
+    assert len(result["series"][0]["pointlist"]) == dd.MAX_POINTS
+    assert result["series"][1]["pointlist"] == [[1000, 0]]
+    assert result["truncated"] is True and result["collectionStatus"] == "partial"
+    assert len(response["series"][0]["pointlist"]) == dd.MAX_POINTS + 1
+
+
+def test_outer_series_limit_is_disclosed_and_an_exact_point_limit_is_not():
+    with patch.object(dd, "_ds", return_value=CREDS), patch.object(dd, "_get", return_value={
+        "status": "ok", "series": [{"metric": str(i), "pointlist": [[1000, 1]]} for i in range(dd.MAX_SERIES + 1)],
+    }):
+        result = body(dd.datadog_query({"query": "avg:fixture.metric{*}"}))
+    assert len(result["series"]) == dd.MAX_SERIES and result["truncated"] is True
+    with patch.object(dd, "_ds", return_value=CREDS), patch.object(dd, "_get", return_value={
+        "status": "ok", "series": [{"metric": "exact", "pointlist": [[1000, 1]] * dd.MAX_POINTS}],
+    }):
+        assert body(dd.datadog_query({"query": "avg:fixture.metric{*}"}))["truncated"] is False
+
+
+def test_custom_header_pair_is_not_overridden_by_stale_dedicated_keys():
+    creds = {**CREDS, "authType": "custom_header", "headerName": "DD-API-KEY", "headerValue": "fresh-api",
+             "headerName2": "DD-APPLICATION-KEY", "headerValue2": "fresh-app"}
+    headers = dd._headers(creds)
+    assert headers["DD-API-KEY"] == "fresh-api" and headers["DD-APPLICATION-KEY"] == "fresh-app"
diff --git a/agent/lambda/test_datasource_http.py b/agent/lambda/test_datasource_http.py
index f408ac75..7307e573 100644
--- a/agent/lambda/test_datasource_http.py
+++ b/agent/lambda/test_datasource_http.py
@@ -85,12 +85,13 @@ def test_not_connected(self):
                 dh.load_datasource("clickhouse")
 
     def test_load_datasource_instance_id_pure_arg(self):
-        # per-instance id key wins; falls back to the kind-mirror when the id key is absent/blank.
+        # an explicit id resolves only its own id key; the kind mirror serves only the no-id path.
         m = {"5": {"endpoint": "http://prom-a:9090"}, "prometheus": {"endpoint": "http://prom-default:9090"}}
         with mock.patch.object(dh, "_load_secret_map", return_value=m):
             self.assertEqual(dh.load_datasource("prometheus", instance_id=5)["endpoint"], "http://prom-a:9090")
             self.assertEqual(dh.load_datasource("prometheus")["endpoint"], "http://prom-default:9090")  # kind mirror
-            self.assertEqual(dh.load_datasource("prometheus", instance_id=999)["endpoint"], "http://prom-default:9090")  # missing id → mirror
+            with self.assertRaises(dh.NotConnected):
+                dh.load_datasource("prometheus", instance_id=999)
 
     def test_warm_container_no_bleed(self):
         # an inline conn set on one invocation must not bleed into a later instance_id invocation.
@@ -115,10 +116,14 @@ def test_no_redirect_follow(self):
         class _R:
             def __init__(self, code):
                 self._c = code
+                from io import BytesIO
+                self.body = BytesIO(b'{"ok":1}')
             def getcode(self):
                 return self._c
-            def read(self):
-                return b'{"ok":1}'
+            def read(self, size=-1):
+                return self.body.read(size)
+            def close(self):
+                self.body.close()
         with mock.patch.object(dh._opener, "open", return_value=_R(200)):
             status, data = dh.http_json("GET", "http://ch:8123/ping")
         self.assertEqual(status, 200)
diff --git a/agent/lambda/test_loki_mcp.py b/agent/lambda/test_loki_mcp.py
index 0750deee..ea2b2647 100644
--- a/agent/lambda/test_loki_mcp.py
+++ b/agent/lambda/test_loki_mcp.py
@@ -17,7 +17,7 @@ def setUp(self):
 class TestQuery(_Base):
     def test_range_ns_window_and_encoding(self):
         cap = {}
-        with mock.patch.object(lm, "http_json", side_effect=lambda m,u,headers=None,body=None,timeout=None:(cap.update(url=u,headers=headers) or (200,{"status":"success","data":{"resultType":"streams","result":[]}}))):
+        with mock.patch.object(lm, "http_json", side_effect=lambda m,u,headers=None,body=None,timeout=None,max_bytes=None:(cap.update(url=u,headers=headers) or (200,{"status":"success","data":{"resultType":"streams","result":[]}}))):
             out = lm.lambda_handler({"tool_name":"loki_query_range","arguments":{"query":'{app="x"} |= "err"'}}, None)
         self.assertEqual(out["statusCode"],200)
         q=_qs(cap["url"]); self.assertIn("/loki/api/v1/query_range",cap["url"])
@@ -29,7 +29,7 @@ def test_range_ns_window_and_encoding(self):
 
     def test_instant(self):
         cap={}
-        with mock.patch.object(lm,"http_json",side_effect=lambda m,u,headers=None,body=None,timeout=None:(cap.update(url=u) or (200,{"status":"success","data":{"resultType":"streams","result":[]}}))):
+        with mock.patch.object(lm,"http_json",side_effect=lambda m,u,headers=None,body=None,timeout=None,max_bytes=None:(cap.update(url=u) or (200,{"status":"success","data":{"resultType":"streams","result":[]}}))):
             lm.lambda_handler({"tool_name":"loki_query","arguments":{"query":"{app=\"x\"}"}}, None)
         self.assertIn("/loki/api/v1/query", cap["url"]); self.assertNotIn("query_range", cap["url"])
 
diff --git a/agent/lambda/test_mimir_mcp.py b/agent/lambda/test_mimir_mcp.py
index 8e089dce..7b86ad58 100644
--- a/agent/lambda/test_mimir_mcp.py
+++ b/agent/lambda/test_mimir_mcp.py
@@ -25,7 +25,7 @@ def test_execution_timeout_is_passed_and_clamped(self):
         # the connector bounded response SIZE only; a caller can now run model-written PromQL through it,
         # so the diag-signal dry run asks for an execution bound (review: it was applied to prometheus but
         # silently ignored here).
-        for given, expect in (("5s", "5s"), (5, "5s"), ("999", "60s"), ("3600s", "60s"), ("0", "1s"), (0, "1s")):
+        for given, expect in (("5s", "5s"), (5, "5s"), ("999", "10s"), ("3600s", "10s"), ("0", "1s"), (0, "1s")):
             cap={}
             with mock.patch.object(mm,"http_json",side_effect=lambda m,u,headers=None,body=None,timeout=None:(cap.update(url=u) or (200,{"status":"success","data":{"resultType":"vector","result":[]}}))):
                 mm.lambda_handler({"tool_name":"mimir_query","arguments":{"query":"up","timeout":given}},None)
@@ -33,7 +33,7 @@ def test_execution_timeout_is_passed_and_clamped(self):
         cap={}
         with mock.patch.object(mm,"http_json",side_effect=lambda m,u,headers=None,body=None,timeout=None:(cap.update(url=u) or (200,{"status":"success","data":{"resultType":"vector","result":[]}}))):
             mm.lambda_handler({"tool_name":"mimir_query","arguments":{"query":"up"}},None)
-        self.assertNotIn("timeout", _qs(cap["url"]))   # unchanged for callers that ask for no bound
+        self.assertEqual(_qs(cap["url"])["timeout"], ["10s"])
 
     def test_range_seconds_window(self):
         cap={}
diff --git a/agent/lambda/test_prometheus_mcp.py b/agent/lambda/test_prometheus_mcp.py
index e8d5abda..c59cab02 100644
--- a/agent/lambda/test_prometheus_mcp.py
+++ b/agent/lambda/test_prometheus_mcp.py
@@ -49,7 +49,7 @@ def fake(method, url, headers=None, body=None, timeout=None):
             cap.update(url=url)
             return 200, {"status": "success", "data": {"resultType": "vector", "result": []}}
 
-        for given, expect in (("5s", "5s"), (5, "5s"), ("999", "60s"), ("3600s", "60s"), ("0", "1s"), (0, "1s")):
+        for given, expect in (("5s", "5s"), (5, "5s"), ("999", "10s"), ("3600s", "10s"), ("0", "1s"), (0, "1s")):
             cap = {}
             with mock.patch.object(pm, "http_json", side_effect=fake):
                 pm.lambda_handler({"tool_name": "prometheus_query",
@@ -58,7 +58,7 @@ def fake(method, url, headers=None, body=None, timeout=None):
         cap = {}
         with mock.patch.object(pm, "http_json", side_effect=fake):
             pm.lambda_handler({"tool_name": "prometheus_query", "arguments": {"query": "up"}}, None)
-        self.assertNotIn("timeout", _qs(cap["url"]))   # unchanged for callers that ask for no bound
+        self.assertEqual(_qs(cap["url"])["timeout"], ["10s"])
 
     def test_range_defaults(self):
         cap = {}
diff --git a/agent/lambda/test_tempo_mcp.py b/agent/lambda/test_tempo_mcp.py
index 7375b1ca..a89a335d 100644
--- a/agent/lambda/test_tempo_mcp.py
+++ b/agent/lambda/test_tempo_mcp.py
@@ -30,6 +30,10 @@ def test_no_envelope_status_still_success(self):
             out=tm.lambda_handler({"tool_name":"tempo_search","arguments":{"query":"{}"}},None)
         self.assertEqual(out["statusCode"],200)
         self.assertEqual(len(json.loads(out["body"])["traces"]),1)
+    def test_health_accepts_bare_empty_but_search_does_not(self):
+        with mock.patch.object(tm,"http_json",return_value=(200,{})):
+            self.assertTrue(json.loads(tm.lambda_handler({"tool_name":"tempo_health","arguments":{}},None)["body"])["ok"])
+            self.assertEqual(json.loads(tm.lambda_handler({"tool_name":"tempo_search","arguments":{"query":"{}"}},None)["body"])["collectionStatus"],"unknown")
     def test_http_error(self):
         with mock.patch.object(tm,"http_json",return_value=(500,{"raw":"boom"})):
             self.assertEqual(tm.lambda_handler({"tool_name":"tempo_search","arguments":{"query":"{}"}},None)["statusCode"],400)
@@ -37,7 +41,7 @@ def test_http_error(self):
 class TestTrace(_Base):
     def test_get_trace_hex_path(self):
         cap={}
-        with mock.patch.object(tm,"http_json",side_effect=lambda m,u,headers=None,body=None,timeout=None:(cap.update(url=u) or (200,{"batches":[]}))):
+        with mock.patch.object(tm,"http_json",side_effect=lambda m,u,headers=None,body=None,timeout=None,max_bytes=None:(cap.update(url=u) or (200,{"batches":[]}))):
             tm.lambda_handler({"tool_name":"tempo_get_trace","arguments":{"trace_id":"a1B2c3"}},None)
         self.assertTrue(cap["url"].endswith("/api/traces/a1B2c3"))
     def test_non_hex_trace_id_rejected_before_request(self):
diff --git a/agent/tests/test_health_tools.py b/agent/tests/test_health_tools.py
index 84543f99..078105f8 100644
--- a/agent/tests/test_health_tools.py
+++ b/agent/tests/test_health_tools.py
@@ -21,44 +21,51 @@ class TestHealthTools(unittest.TestCase):
     def tearDown(self):
         dh.set_request_conn(None)
 
-    def test_prometheus_health_uses_inline_conn_and_probes_healthy(self):
+    def test_prometheus_health_uses_inline_conn_and_verifies_query(self):
         with mock.patch.object(prometheus_mcp, "assert_host_allowed"), \
-             mock.patch.object(dh, "assert_host_allowed"), \
-             mock.patch.object(dh, "http_json", return_value=(200, {})) as hj:
+             mock.patch.object(prometheus_mcp, "http_json", return_value=(200, {
+                 "status": "success", "data": {"resultType": "vector", "result": [{"metric": {}, "value": [1, "1"]}]},
+             })) as hj:
             resp = prometheus_mcp.lambda_handler(
                 {"tool_name": "prometheus_health", "arguments": {}, "conn_config": {"endpoint": "http://p:9090", "authType": "none"}},
                 None,
             )
         self.assertEqual(resp["statusCode"], 200)
         self.assertTrue(json.loads(resp["body"])["ok"])
-        self.assertTrue(hj.call_args[0][1].endswith("/-/healthy"))  # GET <endpoint>/-/healthy
+        self.assertIn("/api/v1/query?", hj.call_args[0][1])
+        self.assertIn("timeout=1s", hj.call_args[0][1])
 
-    def test_clickhouse_health_probes_ping(self):
-        with mock.patch.object(dh, "assert_host_allowed"), \
-             mock.patch.object(dh, "http_json", return_value=(200, {})) as hj:
+    def test_clickhouse_health_verifies_readonly_sql(self):
+        with mock.patch.object(clickhouse_mcp, "assert_host_allowed"), \
+             mock.patch.object(clickhouse_mcp, "http_json", return_value=(200, {
+                 "meta": [{"name": "probe", "type": "UInt8"}], "data": [{"probe": 1}], "rows": 1,
+             })) as hj:
             resp = clickhouse_mcp.lambda_handler(
                 {"tool_name": "clickhouse_health", "arguments": {}, "conn_config": {"endpoint": "http://ch:8123", "authType": "basic", "username": "u", "password": "p"}},
                 None,
             )
         self.assertEqual(resp["statusCode"], 200)
         self.assertTrue(json.loads(resp["body"])["ok"])
-        self.assertTrue(hj.call_args[0][1].endswith("/ping"))
+        self.assertIn("readonly=1", hj.call_args[0][1])
+        self.assertIn("max_execution_time=1", hj.call_args[0][1])
 
-    def test_loki_health_probes_ready(self):
-        with mock.patch.object(dh, "assert_host_allowed"), \
-             mock.patch.object(dh, "http_json", return_value=(200, {})) as hj:
+    def test_loki_health_verifies_tenant_query(self):
+        with mock.patch.object(loki_mcp, "assert_host_allowed"), \
+             mock.patch.object(loki_mcp, "http_json", return_value=(200, {
+                 "status": "success", "data": {"resultType": "streams", "result": []},
+             })) as hj:
             resp = loki_mcp.lambda_handler(
                 {"tool_name": "loki_health", "arguments": {}, "conn_config": {"endpoint": "http://loki:3100", "authType": "none", "org_id": "t1"}},
                 None,
             )
         self.assertEqual(resp["statusCode"], 200)
         self.assertTrue(json.loads(resp["body"])["ok"])
-        self.assertTrue(hj.call_args[0][1].endswith("/ready"))
+        self.assertIn("/loki/api/v1/query_range?", hj.call_args[0][1])
+        self.assertEqual(hj.call_args.kwargs["headers"]["X-Scope-OrgID"], "t1")
 
     def test_health_reports_failure_on_http_error(self):
         with mock.patch.object(prometheus_mcp, "assert_host_allowed"), \
-             mock.patch.object(dh, "assert_host_allowed"), \
-             mock.patch.object(dh, "http_json", return_value=(502, {})):
+             mock.patch.object(prometheus_mcp, "http_json", return_value=(502, {})):
             resp = prometheus_mcp.lambda_handler(
                 {"tool_name": "prometheus_health", "conn_config": {"endpoint": "http://p:9090"}}, None,
             )
diff --git a/docs-site/docs/monitoring/datasources.md b/docs-site/docs/monitoring/datasources.md
index d4707f87..b926fabb 100644
--- a/docs-site/docs/monitoring/datasources.md
+++ b/docs-site/docs/monitoring/datasources.md
@@ -58,7 +58,7 @@ AWSops 데이터소스 기능은 외부 관측성 플랫폼을 중앙에서 관
 | **Type** | O | 데이터소스 유형 (8종 중 선택) |
 | **URL** | O | 엔드포인트 URL (예: `http://prometheus:9090`) |
 | **Authentication** | - | 인증 방식 (None, Basic, Bearer Token, Custom Header) |
-| **Timeout** | - | 저장 범위 1–60초(기본 10초). ClickHouse는 모든 경로에서 `max_execution_time` 상한으로 적용하며 유효 최대는 55초(56–60초는 55초로 단축). Prometheus/Mimir는 Explore 경로에서만 API `timeout`으로 적용하며 최대 10초. 그 외 kind(Loki/Tempo/Jaeger/Dynatrace/Datadog)는 저장만 되고 현재 적용되지 않음 |
+| **Timeout** | - | 저장 범위 1–60초(기본 10초), 유효 최대 55초(56–60초는 55초로 단축). Prometheus/Mimir/ClickHouse는 모든 경로에서 서버 실행 제한으로 적용하고 커넥터 HTTP 타임아웃을 그 위로 정렬합니다. Explore의 Prometheus/Mimir 쿼리는 최대 10초로 더 짧게 제한합니다. 그 외 kind(Loki/Tempo/Jaeger/Dynatrace/Datadog)는 쿼리 도구의 HTTP 응답 제한(기본 12초)으로 적용하며 서버 쿼리 중단은 보장하지 않습니다. 스키마/디스커버리 호출은 커넥터의 고정 HTTP 예산을 유지합니다 |
 | **Database** | - | 기본 데이터베이스 이름 (ClickHouse 전용, 식별자만 허용) |
 
 :::note v1과의 차이
@@ -79,15 +79,15 @@ v1의 결과 캐시 TTL 설정은 v2에 없습니다 — v2의 질의 경로는
 
 | 데이터소스 | 테스트 엔드포인트 | 확인 내용 |
 |-----------|-----------------|----------|
-| Prometheus | `/-/healthy` | 서버 상태, 응답 시간 |
-| Loki | `/ready` | 서버 준비 상태, 응답 시간 |
-| Tempo | `/ready` | 서버 준비 상태, 응답 시간 |
-| ClickHouse | `SELECT 1` | 쿼리 실행 가능 여부, 응답 시간 |
-| Jaeger | `/api/services` | 서비스 목록 조회, 응답 시간 |
-| Dynatrace | `/api/v2/entities` | API 접근 가능 여부, 응답 시간 |
-| Datadog | `/api/v1/validate` | API 키 유효성, 응답 시간 |
+| Prometheus | `query?query=vector(1)` (서버 타임아웃 1초) | 쿼리 접근 |
+| Loki | `query_range`, 1분, `limit=1` | 테넌트 범위 쿼리 접근 |
+| Tempo | 트레이스 1건 검색, 1분 | 쿼리 접근 |
+| ClickHouse | `SELECT 1`, `readonly=1`, 실행 제한 1초 | 읽기 전용 SQL 접근 |
+| Jaeger | `/api/traces?limit=1`, 1분 | 쿼리 접근 |
+| Dynatrace | Metrics v2 `builtin:host.cpu.usage:avg`, 1분 | 메트릭 쿼리 접근 |
+| Datadog | 키 검증(키 쌍만) 후 `/api/v1/query`, 1분 | 메트릭 쿼리 접근 |
 
-테스트 결과에는 연결 성공/실패 상태와 응답 지연 시간(ms)이 표시됩니다.
+테스트 결과에는 성공/실패와 응답 지연 시간(ms)이 표시됩니다.
 
 ## 쿼리 실행
 
@@ -203,7 +203,7 @@ AI 어시스턴트는 등록된 데이터소스를 활용하여 분석을 수행
 
 | 설정 | 기본값 | 설명 |
 |------|--------|------|
-| **Timeout** | 10초 | 업스트림 쿼리 실행 제한(초, 1–60). ClickHouse는 모든 경로(Explore·서비스 그래프·에이전트)의 상한(ceiling)으로 적용되고(호출자는 더 짧게만 조정 가능) 커넥터가 자체 HTTP 타임아웃을 그 위로 정렬합니다(유효 최댓값 55초 — Lambda 60초 한도 아래 정렬을 위해 56–60초 설정은 55초로 단축됩니다). Prometheus/Mimir는 Explore 경로의 API `timeout` 파라미터로 적용되며 커넥터 HTTP 타임아웃(12초) 아래로 10초에 캡됩니다 |
+| **Timeout** | 10초 | 저장 범위 1–60초(기본 10초), 유효 최대 55초(56–60초는 55초로 단축). Prometheus/Mimir/ClickHouse는 모든 경로에서 서버 실행 제한으로 적용하고 커넥터 HTTP 타임아웃을 그 위로 정렬합니다. Explore의 Prometheus/Mimir 쿼리는 최대 10초로 더 짧게 제한합니다. 그 외 kind(Loki/Tempo/Jaeger/Dynatrace/Datadog)는 쿼리 도구의 HTTP 응답 제한(기본 12초)으로 적용하며 서버 쿼리 중단은 보장하지 않습니다. 스키마/디스커버리 호출은 커넥터의 고정 HTTP 예산을 유지합니다 |
 
 ### ClickHouse 전용
 
diff --git a/docs-site/i18n/en/docusaurus-plugin-content-docs/current/monitoring/datasources.md b/docs-site/i18n/en/docusaurus-plugin-content-docs/current/monitoring/datasources.md
index 69c277ec..e125dcff 100644
--- a/docs-site/i18n/en/docusaurus-plugin-content-docs/current/monitoring/datasources.md
+++ b/docs-site/i18n/en/docusaurus-plugin-content-docs/current/monitoring/datasources.md
@@ -54,7 +54,7 @@ Creating, editing, and deleting datasources requires an admin role. In v2, admin
 | **Type** | Yes | Datasource type (select from 8 types) |
 | **URL** | Yes | Endpoint URL (e.g., `http://prometheus:9090`) |
 | **Authentication** | No | Auth method (None, Basic, Bearer Token, Custom Header) |
-| **Timeout** | No | Stored range 1–60 seconds (default 10). ClickHouse applies the `max_execution_time` ceiling on every path, with an effective maximum of 55s (56–60 is shortened to 55). Prometheus/Mimir apply the API `timeout` only on the Explore path, capped at 10s. Other kinds (Loki/Tempo/Jaeger/Dynatrace/Datadog) store the value but do not apply it today |
+| **Timeout** | No | Stored range 1–60 seconds (default 10), effective maximum 55s (56–60 is shortened to 55). Prometheus/Mimir/ClickHouse apply it as the server execution limit on every path and align the connector HTTP timeout above it. Explore further caps Prometheus/Mimir queries at 10s. Other kinds (Loki/Tempo/Jaeger/Dynatrace/Datadog) apply it as the HTTP response limit of their query tools (default 12s); it does not guarantee server-side cancellation. Schema/discovery calls keep the connector's fixed HTTP budget |
 | **Database** | No | Default database name (ClickHouse only, identifier-only) |
 
 :::note Difference from v1
@@ -75,15 +75,15 @@ Clicking **Test Connection** performs the following checks per datasource type:
 
 | Datasource | Test Endpoint | Verification |
 |-----------|--------------|-------------|
-| Prometheus | `/-/healthy` | Server health, response time |
-| Loki | `/ready` | Server readiness, response time |
-| Tempo | `/ready` | Server readiness, response time |
-| ClickHouse | `SELECT 1` | Query execution capability, response time |
-| Jaeger | `/api/services` | Service list retrieval, response time |
-| Dynatrace | `/api/v2/entities` | API accessibility, response time |
-| Datadog | `/api/v1/validate` | API key validity, response time |
+| Prometheus | `query?query=vector(1)` (1s server timeout) | Query access |
+| Loki | `query_range`, 1 min, `limit=1` | Tenant-scoped query access |
+| Tempo | One-trace search, 1 min | Query access |
+| ClickHouse | `SELECT 1`, `readonly=1`, 1s limit | Read-only SQL access |
+| Jaeger | `/api/traces?limit=1`, 1 min | Query access |
+| Dynatrace | Metrics v2 `builtin:host.cpu.usage:avg`, 1 min | Metric-query access |
+| Datadog | Key validation (key pair only), then `/api/v1/query`, 1 min | Metric-query access |
 
-Test results display success/failure status and response latency in milliseconds.
+Test results display success/failure and latency in milliseconds.
 
 ## Query Execution
 
@@ -199,7 +199,7 @@ Datasource-related questions are processed through the `datasource` route. The A
 
 | Setting | Default | Description |
 |------|--------|------|
-| **Timeout** | 10s | Upstream query execution bound (seconds, 1–60). ClickHouse applies it as the CEILING on every path (Explore, service graph, agent — callers can only tighten it) with the connector aligning its own HTTP timeout above it (effective maximum 55s — values 56–60 are shortened to 55s to stay under the Lambda's 60s wall); Prometheus/Mimir apply it as the Explore-path API `timeout` param, capped at 10s under the connector's 12s HTTP timeout |
+| **Timeout** | 10s | Stored range 1–60 seconds (default 10), effective maximum 55s (56–60 is shortened to 55). Prometheus/Mimir/ClickHouse apply it as the server execution limit on every path and align the connector HTTP timeout above it. Explore further caps Prometheus/Mimir queries at 10s. Other kinds (Loki/Tempo/Jaeger/Dynatrace/Datadog) apply it as the HTTP response limit of their query tools (default 12s); it does not guarantee server-side cancellation. Schema/discovery calls keep the connector's fixed HTTP budget |
 
 ### ClickHouse only
 
diff --git a/docs-site/i18n/ja/docusaurus-plugin-content-docs/current/monitoring/datasources.md b/docs-site/i18n/ja/docusaurus-plugin-content-docs/current/monitoring/datasources.md
index 7a444b66..48d4165a 100644
--- a/docs-site/i18n/ja/docusaurus-plugin-content-docs/current/monitoring/datasources.md
+++ b/docs-site/i18n/ja/docusaurus-plugin-content-docs/current/monitoring/datasources.md
@@ -58,7 +58,7 @@ AWSops のデータソース機能は、外部オブザーバビリティプラ
 | **Type** | O | データソースのタイプ（8 種から選択） |
 | **URL** | O | エンドポイント URL（例: `http://prometheus:9090`） |
 | **Authentication** | - | 認証方式 (None, Basic, Bearer Token, Custom Header) |
-| **Timeout** | - | 保存範囲は 1–60 秒（デフォルト 10 秒）。ClickHouse はすべての経路で `max_execution_time` の上限として適用し、有効上限は 55 秒（56–60 秒は 55 秒に短縮）。Prometheus/Mimir は Explore 経路のみで API `timeout` として適用し、上限は 10 秒。その他の種類（Loki/Tempo/Jaeger/Dynatrace/Datadog）は保存のみで現在は適用されません |
+| **Timeout** | - | 保存範囲は 1–60 秒（デフォルト 10 秒）、有効上限は 55 秒（56–60 秒は 55 秒に短縮）。Prometheus/Mimir/ClickHouse はすべての経路でサーバー実行上限として適用し、コネクタの HTTP タイムアウトをその上に揃えます。Explore の Prometheus/Mimir クエリはさらに 10 秒に制限します。その他の種類（Loki/Tempo/Jaeger/Dynatrace/Datadog）はクエリツールの HTTP 応答上限（デフォルト 12 秒）として適用され、サーバー側のクエリ停止は保証しません。スキーマ/ディスカバリ呼び出しはコネクタ固定の HTTP 予算を維持します |
 | **Database** | - | デフォルトのデータベース名（ClickHouse 専用、識別子のみ） |
 
 :::note v1 との違い
@@ -79,15 +79,15 @@ v1 の結果キャッシュ TTL 設定は v2 にはありません — v2 のク
 
 | データソース | テストエンドポイント | 確認内容 |
 |-----------|-----------------|----------|
-| Prometheus | `/-/healthy` | サーバー状態、応答時間 |
-| Loki | `/ready` | サーバー準備状態、応答時間 |
-| Tempo | `/ready` | サーバー準備状態、応答時間 |
-| ClickHouse | `SELECT 1` | クエリ実行の可否、応答時間 |
-| Jaeger | `/api/services` | サービス一覧の取得、応答時間 |
-| Dynatrace | `/api/v2/entities` | API アクセスの可否、応答時間 |
-| Datadog | `/api/v1/validate` | API キーの有効性、応答時間 |
+| Prometheus | `query?query=vector(1)` (サーバータイムアウト 1 秒) | クエリアクセス |
+| Loki | `query_range`、1 分、`limit=1` | テナント範囲のクエリアクセス |
+| Tempo | トレース 1 件の検索、1 分 | クエリアクセス |
+| ClickHouse | `SELECT 1`、`readonly=1`、実行制限 1 秒 | 読み取り専用 SQL アクセス |
+| Jaeger | `/api/traces?limit=1`、1 分 | クエリアクセス |
+| Dynatrace | Metrics v2 `builtin:host.cpu.usage:avg`、1 分 | メトリッククエリアクセス |
+| Datadog | キー検証 (キーペアのみ) 後に `/api/v1/query`、1 分 | メトリッククエリアクセス |
 
-テスト結果には、接続の成功/失敗ステータスと応答遅延時間 (ms) が表示されます。
+テスト結果には、成功/失敗と応答遅延時間 (ms) が表示されます。
 
 ## クエリの実行
 
@@ -203,7 +203,7 @@ AI アシスタントは、登録されたデータソースを活用して分
 
 | 設定 | デフォルト | 説明 |
 |------|--------|------|
-| **Timeout** | 10 秒 | アップストリームのクエリ実行上限（秒、1–60）。ClickHouse はすべての経路（Explore・サービスグラフ・エージェント）の上限（ceiling）として適用され（呼び出し側は短くのみ調整可能）、コネクタは自身の HTTP タイムアウトをその上に揃えます（有効上限 55 秒 — Lambda の 60 秒制限内に収めるため 56–60 秒の設定は 55 秒に短縮されます）。Prometheus/Mimir は Explore 経路の API `timeout` パラメータとして適用され、コネクタの 12 秒 HTTP タイムアウトの下で 10 秒にキャップされます |
+| **Timeout** | 10 秒 | 保存範囲は 1–60 秒（デフォルト 10 秒）、有効上限は 55 秒（56–60 秒は 55 秒に短縮）。Prometheus/Mimir/ClickHouse はすべての経路でサーバー実行上限として適用し、コネクタの HTTP タイムアウトをその上に揃えます。Explore の Prometheus/Mimir クエリはさらに 10 秒に制限します。その他の種類（Loki/Tempo/Jaeger/Dynatrace/Datadog）はクエリツールの HTTP 応答上限（デフォルト 12 秒）として適用され、サーバー側のクエリ停止は保証しません。スキーマ/ディスカバリ呼び出しはコネクタ固定の HTTP 予算を維持します |
 
 ### ClickHouse 専用
 
diff --git a/docs-site/i18n/zh/docusaurus-plugin-content-docs/current/monitoring/datasources.md b/docs-site/i18n/zh/docusaurus-plugin-content-docs/current/monitoring/datasources.md
index 34bd705c..fad9a465 100644
--- a/docs-site/i18n/zh/docusaurus-plugin-content-docs/current/monitoring/datasources.md
+++ b/docs-site/i18n/zh/docusaurus-plugin-content-docs/current/monitoring/datasources.md
@@ -58,7 +58,7 @@ AWSops 数据源功能对外部可观测性平台进行集中管理。注册数
 | **Type** | O | 数据源类型（从 8 种中选择） |
 | **URL** | O | 端点 URL（例：`http://prometheus:9090`） |
 | **Authentication** | - | 认证方式（None、Basic、Bearer Token、Custom Header） |
-| **Timeout** | - | 存储范围为 1–60 秒（默认 10 秒）。ClickHouse 在所有路径应用 `max_execution_time` 上限，有效最大值为 55 秒（56–60 秒缩短为 55 秒）。Prometheus/Mimir 仅在 Explore 路径应用 API `timeout`，上限为 10 秒。其他类型（Loki/Tempo/Jaeger/Dynatrace/Datadog）仅存储该值，目前不生效 |
+| **Timeout** | - | 存储范围为 1–60 秒（默认 10 秒），有效最大值为 55 秒（56–60 秒缩短为 55 秒）。Prometheus/Mimir/ClickHouse 在所有路径作为服务器执行上限生效，并将连接器 HTTP 超时对齐到其上。Explore 中的 Prometheus/Mimir 查询进一步限制为 10 秒。其他类型（Loki/Tempo/Jaeger/Dynatrace/Datadog）作为查询工具的 HTTP 响应上限生效（默认 12 秒），不保证服务器端取消查询。模式/发现调用保持连接器固定的 HTTP 预算 |
 | **Database** | - | 默认数据库名称（仅 ClickHouse，仅允许标识符） |
 
 :::note 与 v1 的差异
@@ -79,15 +79,15 @@ v2 中没有 v1 的结果缓存 TTL 设置 — v2 查询路径刻意不做缓存
 
 | 数据源 | 测试端点 | 确认内容 |
 |-----------|-----------------|----------|
-| Prometheus | `/-/healthy` | 服务器状态、响应时间 |
-| Loki | `/ready` | 服务器就绪状态、响应时间 |
-| Tempo | `/ready` | 服务器就绪状态、响应时间 |
-| ClickHouse | `SELECT 1` | 查询是否可执行、响应时间 |
-| Jaeger | `/api/services` | 服务列表查询、响应时间 |
-| Dynatrace | `/api/v2/entities` | API 是否可访问、响应时间 |
-| Datadog | `/api/v1/validate` | API 密钥有效性、响应时间 |
+| Prometheus | `query?query=vector(1)`（服务器超时 1 秒） | 查询访问 |
+| Loki | `query_range`、1 分钟、`limit=1` | 租户范围的查询访问 |
+| Tempo | 单条 trace 搜索、1 分钟 | 查询访问 |
+| ClickHouse | `SELECT 1`、`readonly=1`、执行限制 1 秒 | 只读 SQL 访问 |
+| Jaeger | `/api/traces?limit=1`、1 分钟 | 查询访问 |
+| Dynatrace | Metrics v2 `builtin:host.cpu.usage:avg`、1 分钟 | 指标查询访问 |
+| Datadog | 密钥验证（仅密钥对）后 `/api/v1/query`、1 分钟 | 指标查询访问 |
 
-测试结果会显示连接成功/失败状态以及响应延迟时间 (ms)。
+测试结果会显示成功/失败以及响应延迟时间 (ms)。
 
 ## 查询执行
 
@@ -203,7 +203,7 @@ AI 助手可以利用已注册的数据源执行分析。
 
 | 设置 | 默认值 | 说明 |
 |------|--------|------|
-| **Timeout** | 10 秒 | 上游查询执行上限（秒，1–60）。ClickHouse 在所有路径（Explore、服务图、代理）作为上限生效（调用方只能调得更短），连接器会将自身 HTTP 超时对齐到该上限之上（有效上限 55 秒 — 为保持在 Lambda 60 秒限制之下，56–60 秒的设置会缩短为 55 秒）；Prometheus/Mimir 通过 Explore 路径的 API `timeout` 参数生效，并在连接器 12 秒 HTTP 超时之下封顶为 10 秒 |
+| **Timeout** | 10 秒 | 存储范围为 1–60 秒（默认 10 秒），有效最大值为 55 秒（56–60 秒缩短为 55 秒）。Prometheus/Mimir/ClickHouse 在所有路径作为服务器执行上限生效，并将连接器 HTTP 超时对齐到其上。Explore 中的 Prometheus/Mimir 查询进一步限制为 10 秒。其他类型（Loki/Tempo/Jaeger/Dynatrace/Datadog）作为查询工具的 HTTP 响应上限生效（默认 12 秒），不保证服务器端取消查询。模式/发现调用保持连接器固定的 HTTP 预算 |
 
 ### 仅 ClickHouse
 
diff --git a/docs/api-reference.md b/docs/api-reference.md
index 3e9bcc03..d5559117 100644
--- a/docs/api-reference.md
+++ b/docs/api-reference.md
@@ -381,7 +381,7 @@ reachability.
 | `/api/customization` | GET, POST, PUT | Admin skill/agent catalog CRUD; invalid tool policy → 400; unavailable validation/read → 503 (see below) | verifyUser |
 | `/api/datasources` | GET | 데이터소스 인스턴스 목록 — 크리덴셜 미노출 | verifyUser |
 | `/api/datasources/generate` | POST | 자연어 → 쿼리 초안 생성 (리뷰용 — 절대 실행 안 함) | verifyUser |
-| `/api/datasources/manage` | POST, PATCH | 인스턴스 생성/수정 + 크리덴셜 저장 (admin); `settings`(timeoutS 1–60[clickhouse 유효 최대 55]·clickhouse database)는 서버 측 sanitize 후 ds_settings JSONB에 저장 | verifyUser |
+| `/api/datasources/manage` | POST, PATCH | 인스턴스 생성/수정 + 크리덴셜 저장 (admin); `settings`(timeoutS 1–60[모든 kind 유효 최대 55 — Prometheus/Mimir/ClickHouse는 서버 실행 제한(Explore의 Prometheus/Mimir는 10초 상한), 그 외는 HTTP 응답 제한]·clickhouse database)는 서버 측 sanitize 후 ds_settings JSONB에 저장 | verifyUser |
 | `/api/datasources/query` | POST | 인스턴스 대상 read-only 쿼리 실행 (admin 아님 — 탐색용) | verifyUser |
 | `/api/datasources/test` | POST | 저장 전 연결 probe — SSRF 가드 (admin) | verifyUser |
 | `/api/datasources/[id]` | DELETE | 인스턴스 삭제 — 스키마 캐시/크리덴셜 cascade, 기본값 재선정 (admin) | verifyUser |
diff --git a/docs/runbooks/AGENTS.md b/docs/runbooks/AGENTS.md
index 6f2f8cee..2a447abb 100644
--- a/docs/runbooks/AGENTS.md
+++ b/docs/runbooks/AGENTS.md
@@ -13,6 +13,8 @@
 Web image receipt/ECR proof precedes private migrations, including reuse; promotion retains that digest. Automatic DDL is expand-only; contract cutovers require a merge freeze, drained queues and explicit operator coordination. Legacy images without receipts use the separately approved `legacy-web-image-recovery.md` path with trusted digest/source evidence and schema approval, never a fabricated receipt or mutable-tag fallback.
 
 Operational playbooks organized by scenario, each following symptoms → diagnosis → action.
+`datasource-credential-isolation.md` requires native-default credential backfill before stricter
+explicit-ID reads roll out; missing or foreign credentials fail closed (ADR-007/039).
 Graph collection/projection, freshness and retained-evidence rollout contracts live in `source-sync-observability.md`. See
 `docs/runbooks/CLAUDE.md`'s index for the current runbook list. v1-era runbooks have been
 removed; flag any change that reintroduces a v1 procedure.
diff --git a/docs/runbooks/CLAUDE.md b/docs/runbooks/CLAUDE.md
index b36aaf07..8e5611c0 100644
--- a/docs/runbooks/CLAUDE.md
+++ b/docs/runbooks/CLAUDE.md
@@ -19,6 +19,7 @@ erasing valid Read evidence. Source helper/tests and the role consumer catalog a
 | [istio-agent-eks-access.md](istio-agent-eks-access.md) | Granting `istio-read` MCP access to an EKS cluster (agent Lambda role Access Entry) |
 | [network-path-eks-access.md](network-path-eks-access.md) | Granting Network Path Check live-identity verification access to an EKS cluster's Nodes/Pods (worker/member-role Access Entry with minimal Nodes/Pods read RBAC) |
 | [k8sgpt-operator-install.md](k8sgpt-operator-install.md) | Out-of-band K8sGPT operator install (manual operator work, ADR-005 precedent) |
+| [datasource-credential-isolation.md](datasource-credential-isolation.md) | Native datasource credential identity and mandatory default backfill before rollout (ADR-007/039) |
 | [tempo-query-generation.md](tempo-query-generation.md) | Tempo query generation — connector/web deployment, admin API schema refresh, cached summaries, and recent-window limits |
 | [graph-read-contract.md](graph-read-contract.md) | Bounded graph reads/rebuilds, shared admission, infra dependency, publication outcomes, registry diagnostics and PostgreSQL fixtures |
 | [source-sync-observability.md](source-sync-observability.md) | Public source rollout — flow/infra/trace collection clocks, retention/budgets, reader projections, DX scope and deployment prerequisites |
diff --git a/docs/runbooks/datasource-credential-isolation.md b/docs/runbooks/datasource-credential-isolation.md
new file mode 100644
index 00000000..69de61cc
--- /dev/null
+++ b/docs/runbooks/datasource-credential-isolation.md
@@ -0,0 +1,73 @@
+# Datasource credential identity and rollout
+
+This runbook covers native datasource isolation under ADR-007 external-data governance
+and ADR-039 integration identity. It does not approve deployment or activate any gate.
+
+## Symptoms
+
+After the stricter connector rollout, an explicit-ID read for a migrated default whose
+per-instance credential key was never written fails closed. A sibling without its own
+credentials must fail.
+
+## Diagnosis
+
+Native datasource connectors store one credential blob per integration ID
+(`map[<id>]`). Their plain kind key is the managed mirror of the current default
+and serves only calls without an explicit ID. An explicit-ID read uses its own ID
+key only; it never falls back to the kind mirror. Missing credentials fail closed.
+
+## Action
+
+First run the idempotent instance backfill from the repository root:
+`node scripts/v2/backfill-datasource-instances.mjs --dry-run`, then again without
+`--dry-run`. It covers all eight native kinds (prometheus, mimir, loki, tempo,
+clickhouse, jaeger, dynatrace, datadog): for each kind credential it ensures an
+`integrations` row named after the kind and copies the kind credential to that
+row's ID key when the ID key is missing, that row is the current default and the
+row was just created or its endpoint origin matches the credential's; any other
+row named after the kind is reported as skipped. It does not touch rows with any
+other name, so still re-save every migrated default as follows.
+
+Before deploying the stricter connector producer, operators must re-save every
+migrated default through the existing authenticated administrator save flow:
+
+1. Identify the current default datasource in the Integrations interface. Keep
+   its endpoint and authentication type unchanged.
+2. Save that datasource's connection settings (Integrations → datasource → save
+   connection settings). Do not print credentials.
+3. Confirm the save succeeded. For a default, the save reads the ID key and falls
+   back to the kind mirror only when the mirror's endpoint origin matches the row;
+   otherwise stored authentication material is dropped and must be resupplied
+   through the same write-only form. It strips stale settings, writes the merged
+   blob under the instance ID and re-mirrors it to the kind key.
+4. Verify a credential-blind read for that exact instance ID before the connector
+   rollout, and verify a same-kind instance without its own credentials fails
+   closed. Stop deployment until all migrated defaults pass.
+
+A Datadog blob configured directly in Secrets Manager with `apiKey`/`appKey` and
+no `authType` keeps sending those keys, but the administrator save stamps the
+authentication type it submits. An explicit `none` transmits no stored keys, so
+re-save such a datasource with Custom header authentication and enter
+`DD-API-KEY`/`DD-APPLICATION-KEY` as the two header pairs.
+
+Default switching and deletion keep today's web behavior in this change. The
+successor's credential is resolved by ID key with a fallback to the kind mirror —
+which at that moment still holds the removed default's blob — so a successor
+without its own ID key is re-mirrored with the previous default's credential (a
+known residual); the mirror is cleared only when no instance of the kind remains.
+The datasource web contract change replaces this with an ID-stamped mirror and a
+409 guard. Secrets Manager and row writes are not one atomic cross-service
+transaction; a failed save requires inspection and retry through the administrator
+flow before rollout.
+
+The implementation is `web/app/api/datasources/manage/route.ts` and
+`agent/lambda/datasource_http.py`. Focused verification:
+
+```bash
+cd web && npx vitest run app/api/datasources/manage/route.test.ts
+cd agent/lambda && python3 -m pytest test_datasource_http.py test_collection_markers.py test_loki_mcp.py test_datadog_mcp.py test_dynatrace_mcp.py -q
+```
+
+Run the two commands from the repository root independently. These local tests
+mock Secrets Manager/provider reads and do not establish successful production
+backfill, producer deployment, IAM readiness or live provider access.
diff --git a/scripts/v2/agentcore/catalog.py b/scripts/v2/agentcore/catalog.py
index 9d691617..e756b039 100644
--- a/scripts/v2/agentcore/catalog.py
+++ b/scripts/v2/agentcore/catalog.py
@@ -406,8 +406,8 @@ def _p(t, d=""):
         "description": "Loki read-only — LogQL range/instant query, labels, label values (4 tools)",
         "tools": [
             {"name": "loki_schema", "description": "Introspect label names (cached)", "inputSchema": {"type": "object", "properties": {}}},
-            {"name": "loki_query_range", "description": "Range LogQL query over a time window (logs/metrics)", "inputSchema": {"type": "object", "properties": {"query": _p("string", "LogQL"), "start": _p("string", "1h/30m or unix/ISO (default now-1h)"), "end": _p("string", "unix/ISO (default now)"), "limit": _p("string", "Max entries (default 100)"), "direction": _p("string", "forward|backward (default backward)")}, "required": ["query"]}},
-            {"name": "loki_query", "description": "Instant LogQL query", "inputSchema": {"type": "object", "properties": {"query": _p("string", "LogQL"), "time": _p("string", "Eval time unix/ISO (default now)"), "limit": _p("string", "Max entries (default 100)")}, "required": ["query"]}},
+            {"name": "loki_query_range", "description": "Range LogQL query over a time window (logs/metrics); collectionStatus ok/empty denotes complete results; partial/unknown/error or truncated output must not support an absence claim. Reaching the requested limit sets limitSaturated and is partial.", "inputSchema": {"type": "object", "properties": {"query": _p("string", "LogQL"), "start": _p("string", "1h/30m or unix/ISO (default now-1h)"), "end": _p("string", "unix/ISO (default now)"), "limit": _p("string", "Max entries (default 100)"), "direction": _p("string", "forward|backward (default backward)")}, "required": ["query"]}},
+            {"name": "loki_query", "description": "Instant LogQL query; collectionStatus ok/empty denotes complete results; partial/unknown/error or truncated output must not support an absence claim. Reaching the requested limit sets limitSaturated and is partial.", "inputSchema": {"type": "object", "properties": {"query": _p("string", "LogQL"), "time": _p("string", "Eval time unix/ISO (default now)"), "limit": _p("string", "Max entries (default 100)")}, "required": ["query"]}},
             {"name": "loki_labels", "description": "List label names", "inputSchema": {"type": "object", "properties": {}}},
             {"name": "loki_label_values", "description": "List values of a label", "inputSchema": {"type": "object", "properties": {"label": _p("string", "Label name")}, "required": ["label"]}},
         ],
diff --git a/scripts/v2/backfill-datasource-instances.mjs b/scripts/v2/backfill-datasource-instances.mjs
index 74e1028d..5ea89a41 100644
--- a/scripts/v2/backfill-datasource-instances.mjs
+++ b/scripts/v2/backfill-datasource-instances.mjs
@@ -19,12 +19,13 @@ import {
   SecretsManagerClient, GetSecretValueCommand, PutSecretValueCommand,
 } from '@aws-sdk/client-secrets-manager';
 
-const DATASOURCE_KINDS = ['prometheus', 'mimir', 'loki', 'tempo', 'clickhouse'];
+const DATASOURCE_KINDS = ['prometheus', 'mimir', 'loki', 'tempo', 'clickhouse', 'jaeger', 'dynatrace', 'datadog']; // mirrors web/lib/integrations-category.ts DATASOURCE_KINDS
 const SECRET_NAME = process.env.INTEGRATIONS_SECRET_NAME || 'ops/awsops-v2/integrations/credentials';
 const REGION = process.env.AWS_REGION || 'ap-northeast-2';
 const LOCK_KEY = 472941143; // distinct from migrate.mjs / backfill-v1.mjs
 const die = (m) => { console.error(`\n✗ ${m}`); process.exit(1); };
 const redactDsn = (d) => { try { const u = new URL(d); if (u.password) u.password = '***'; return u.toString(); } catch { return '***'; } };
+const originOf = (u) => { try { return new URL(String(u ?? '')).origin; } catch { return null; } }; // scheme+host+port, as manage/route.ts
 
 function resolveDsn() {
   if (process.env.BACKFILL_DSN) return process.env.BACKFILL_DSN;
@@ -59,6 +60,7 @@ async function main() {
   console.log(`connected: ${redactDsn(resolveDsn())}`);
   const c = await pool.connect();
   let created = 0, mirrored = 0;
+  const skipped = [];
   try {
     await c.query('BEGIN');
     await c.query('SELECT pg_advisory_xact_lock($1)', [LOCK_KEY]);
@@ -73,11 +75,15 @@ async function main() {
          RETURNING id`,
         [slug],
       );
-      const { rows } = await c.query('SELECT id FROM integrations WHERE name = $1', [slug]);
+      const { rows } = await c.query('SELECT id, is_default, endpoint FROM integrations WHERE name = $1 AND kind = $1', [slug]);
       const id = rows[0]?.id;
       if (ins.rowCount > 0) created++;
-      // Copy the slug credential to the bigint id key if the id entry is missing (keep the kind mirror).
+      // The kind key mirrors the CURRENT default, but after a default switch/delete it can still hold another
+      // instance's blob. Copy it only into the slug-named default row it was just created from, or whose endpoint
+      // origin matches the mirror's (as the admin save flow does); otherwise report it for an administrator re-save.
       if (id != null && map[String(id)] === undefined) {
+        const sameOrigin = originOf(map[slug]?.endpoint) !== null && originOf(map[slug]?.endpoint) === originOf(rows[0].endpoint);
+        if (!rows[0].is_default || !(ins.rowCount > 0 || sameOrigin)) { skipped.push(slug); continue; }
         if (!dryRun) { map[String(id)] = map[slug]; secretDirty = true; }
         mirrored++;
       }
@@ -94,7 +100,7 @@ async function main() {
     c.release();
     await pool.end();
   }
-  console.log(`${dryRun ? '[dry-run] ' : ''}rows created: ${created}, id-credentials backfilled: ${mirrored}`);
+  console.log(`${dryRun ? '[dry-run] ' : ''}rows created: ${created}, id-credentials backfilled: ${mirrored}, skipped non-default or origin-mismatched rows (re-save via the admin flow): ${skipped.join(', ') || 'none'}`);
 }
 
 main().catch((e) => die(e?.message || String(e)));
diff --git a/web/lib/datasources.test.ts b/web/lib/datasources.test.ts
index 901ec1ce..922476f8 100644
--- a/web/lib/datasources.test.ts
+++ b/web/lib/datasources.test.ts
@@ -138,7 +138,7 @@ describe('resolveConnConfig', () => {
     expect(getCredentialById).not.toHaveBeenCalledWith(1, 'prometheus'); // never the kind fallback
   });
 
-  it('clickhouse settings ride the conn config (database + timeoutS); other kinds never set them', async () => {
+  it('timeoutS rides the conn config for every kind; database stays ClickHouse-only', async () => {
     getCredentialById.mockResolvedValueOnce(null);
     const ch = { ...row, kind: 'clickhouse', settings: { database: 'metrics_db', timeoutS: 30 } };
     expect(await resolveConnConfig(ch)).toMatchObject({ database: 'metrics_db', timeoutS: 30 });
@@ -146,7 +146,7 @@ describe('resolveConnConfig', () => {
     const prom = { ...row, settings: { database: 'metrics_db', timeoutS: 30 } }; // kind: prometheus
     const cc = await resolveConnConfig(prom);
     expect(cc).not.toHaveProperty('database');
-    expect(cc).not.toHaveProperty('timeoutS');
+    expect(cc).toMatchObject({ timeoutS: 30 });
   });
 
   it('a stale blob database/timeoutS never leaks through — the ROW settings are authoritative', async () => {
diff --git a/web/lib/datasources.ts b/web/lib/datasources.ts
index fb7d4857..196fcfef 100644
--- a/web/lib/datasources.ts
+++ b/web/lib/datasources.ts
@@ -185,11 +185,11 @@ export async function resolveConnConfig(ds: DatasourceRow): Promise<ConnConfig>
     ...(ds.endpoint ? { endpoint: ds.endpoint } : {}),
     ...(ds.authType ? { authType: ds.authType } : {}),
     // gap L203: the ClickHouse settings ride the conn config — database (identifier-validated
-    // on write AND read) becomes &database=, and timeoutS becomes the connector's DEFAULT
-    // max_execution_time, so the Explore route, the service-graph sources, and the agent
-    // path all get the same bound from one mechanism.
+    // on write AND read) becomes &database=; timeoutS is the connector's query bound for every kind
+    // (execution ceiling or transport deadline), so the Explore route, the service-graph sources and
+    // the agent path all get the same bound from one mechanism.
     ...(ds.kind === 'clickhouse' && ds.settings?.database ? { database: ds.settings.database } : {}),
-    ...(ds.kind === 'clickhouse' && ds.settings?.timeoutS ? { timeoutS: ds.settings.timeoutS } : {}),
+    ...(ds.settings?.timeoutS ? { timeoutS: ds.settings.timeoutS } : {}),
   } as ConnConfig;
 }