diff --git a/tests/api-testing/tests/dashboards/test_concurrent_create_default_folder.py b/tests/api-testing/tests/dashboards/test_concurrent_create_default_folder.py
new file mode 100644
index 0000000000..c99c5b69b1
--- /dev/null
+++ b/tests/api-testing/tests/dashboards/test_concurrent_create_default_folder.py
@@ -0,0 +1,78 @@
+"""Concurrent dashboard create in a fresh org must not 500  [regression #14163].
+
+In a brand-new org the `default` dashboards folder does not exist yet. Several
+concurrent `POST /dashboards` all passed the folder-existence check and then
+raced to create it; the losers' INSERT violated the folder unique index and
+surfaced as a raw HTTP 500. Retrying worked, because by then the folder existed.
+PR #14186 made the creation idempotent -- `folders::get_or_create` attempts the
+insert and only reports failure if the row is still absent afterwards.
+
+This is a check-then-act race, so it only shows up when the requests genuinely
+overlap; issuing them sequentially would pass on the unfixed build. The requests
+are therefore fired from a thread pool against a freshly created org.
+
+Storage contention is NOT the defect. On SQLite a concurrent burst can answer
+"database is locked", which is a 5xx that says nothing about folder creation, so
+those are counted and tolerated separately -- the defect's own signature is a
+unique-constraint failure, which does not mention a lock. Treating every 5xx as
+the bug made this test fail on CI even against a fixed build.
+"""
+
+import logging
+import os
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+CONCURRENCY = 6
+LOCK_MARKERS = ("database is locked", "is locked", "locked")
+
+
+def _is_storage_contention(resp) -> bool:
+    body = (resp.text or "").lower()
+    return any(m in body for m in LOCK_MARKERS)
+
+
+@pytest.fixture(scope="module")
+def fresh_org(create_session, base_url):
+    """A brand-new org, so the default dashboards folder is guaranteed absent.
+
+    Module-scoped deliberately: `DELETE /api/organizations/{id}` answers 404 on
+    this build, so an org cannot be cleaned up and a function-scoped fixture
+    would leak one per test. One org per module bounds that to a single row.
+    """
+    session = create_session
+    org = f"e2e_dash_race_{uuid.uuid4().hex[:8]}"
+    resp = session.post(f"{base_url}api/organizations", json={"name": org})
+    if resp.status_code not in (200, 201):
+        pytest.skip(f"could not create a fresh org ({resp.status_code}): {resp.text[:200]}")
+    yield resp.json().get("identifier", org)
+
+
+def test_concurrent_dashboard_create_does_not_500(create_session, base_url, fresh_org):
+    session = create_session
+
+    def create(i: int):
+        return session.post(
+            f"{base_url}api/{fresh_org}/dashboards",
+            json={"title": f"race_{i}", "description": "concurrent create race", "panels": []},
+        )
+
+    with ThreadPoolExecutor(max_workers=CONCURRENCY) as pool:
+        responses = list(pool.map(create, range(CONCURRENCY)))
+
+    codes = [r.status_code for r in responses]
+    contention = [r for r in responses if r.status_code >= 500 and _is_storage_contention(r)]
+    defects = [r for r in responses if r.status_code >= 500 and not _is_storage_contention(r)]
+    logger.info("status codes: %s (storage contention: %d)", codes, len(contention))
+
+    assert not defects, (
+        "concurrent dashboard creates raced default-folder creation and returned "
+        f"{[r.status_code for r in defects]}; first body: {defects[0].text[:300]}"
+    )
+    assert any(c in (200, 201) for c in codes), \
+        f"no concurrent create succeeded at all, got {codes}"
diff --git a/tests/api-testing/tests/ingest/test_otlp_log_severity.py b/tests/api-testing/tests/ingest/test_otlp_log_severity.py
new file mode 100644
index 0000000000..5c75f06df7
--- /dev/null
+++ b/tests/api-testing/tests/ingest/test_otlp_log_severity.py
@@ -0,0 +1,87 @@
+"""OTLP log severity must be consistent  [regression #14639].
+
+`severity` was stored as `severityText` when the record set it and as the raw
+`severityNumber` otherwise, so the same level landed as `"ERROR"` from one
+exporter and `17` from another. The column type then depended on which record
+arrived first, and the log viewer read a numeric severity as a syslog level.
+PR #14791 derives the text from the number and keeps the number alongside.
+
+Both records carry severityNumber 17, differing only in whether severityText is
+set, so any divergence between them is the defect.
+"""
+
+import logging
+import os
+import time
+import uuid
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+ORG_ID = os.environ.get("TEST_ORG_ID", "default")
+SEVERITY_NUMBER = 17          # OTLP ERROR
+
+
+def test_severity_is_derived_when_only_the_number_is_sent(create_session, base_url):
+    session = create_session
+    tag_with = f"withtext_{uuid.uuid4().hex[:6]}"
+    tag_without = f"numonly_{uuid.uuid4().hex[:6]}"
+    t_ns = int(time.time() * 1e9) - 180 * 10**9
+
+    def record(offset_ns, tag, with_text):
+        rec = {
+            "timeUnixNano": str(t_ns + offset_ns),
+            "severityNumber": SEVERITY_NUMBER,
+            "body": {"stringValue": f"probe {tag}"},
+            "attributes": [{"key": "probe_tag", "value": {"stringValue": tag}}],
+        }
+        if with_text:
+            rec["severityText"] = "ERROR"
+        return rec
+
+    payload = {"resourceLogs": [{
+        "resource": {"attributes": [{"key": "service.name",
+                                     "value": {"stringValue": "pytest_severity"}}]},
+        "scopeLogs": [{"scope": {"name": "pytest"}, "logRecords": [
+            record(0, tag_with, True),
+            record(10**6, tag_without, False),
+        ]}],
+    }]}
+    r = session.post(f"{base_url}api/{ORG_ID}/v1/logs", json=payload)
+    assert r.status_code == 200, f"OTLP log ingest failed: {r.status_code} {r.text[:300]}"
+
+    now_us = int(time.time() * 1_000_000)
+    rows = []
+    for _ in range(30):
+        q = session.post(
+            f"{base_url}api/{ORG_ID}/_search?type=logs",
+            json={"query": {
+                "sql": 'SELECT probe_tag, severity, severity_number FROM "default" '
+                       f"WHERE probe_tag IN ('{tag_with}','{tag_without}')",
+                "start_time": now_us - 600_000_000, "end_time": now_us + 60_000_000,
+                "from": 0, "size": 10}},
+        )
+        if q.status_code == 200 and len(q.json().get("hits", [])) >= 2:
+            rows = q.json()["hits"]
+            break
+        time.sleep(2)
+    assert len(rows) >= 2, f"both log records never became searchable, got {rows}"
+
+    by_tag = {r_["probe_tag"]: r_ for r_ in rows}
+    logger.info("severity by tag: %s",
+                {k: (v.get("severity"), v.get("severity_number")) for k, v in by_tag.items()})
+
+    sev_with = by_tag[tag_with].get("severity")
+    sev_without = by_tag[tag_without].get("severity")
+
+    # The defect stored the raw number here, giving two representations of one level.
+    assert sev_without == sev_with, \
+        (f"severity must not depend on whether severityText was sent: "
+         f"with-text={sev_with!r} number-only={sev_without!r}")
+    assert str(sev_without).upper() == "ERROR", \
+        f"severityNumber {SEVERITY_NUMBER} should derive ERROR, got {sev_without!r}"
+
+    # The number must survive alongside the derived text.
+    for tag, row in by_tag.items():
+        assert row.get("severity_number") == SEVERITY_NUMBER, \
+            f"{tag} lost its severity_number: {row.get('severity_number')!r}"
diff --git a/tests/api-testing/tests/ingest/test_trace_resource_attribute_collision.py b/tests/api-testing/tests/ingest/test_trace_resource_attribute_collision.py
new file mode 100644
index 0000000000..4734c24d32
--- /dev/null
+++ b/tests/api-testing/tests/ingest/test_trace_resource_attribute_collision.py
@@ -0,0 +1,79 @@
+"""Resource attributes must survive span flattening  [regression #12448].
+
+When a span attribute used the same key as a resource-derived field, flattening
+let the span value overwrite it -- so a span carrying its own `service.name`
+replaced the resource's, and the trace was attributed to the wrong service.
+PR #12456 keeps the resource value canonical and moves the colliding span
+attribute aside.
+
+The control span (no collision) is what makes this meaningful: without it a
+build that simply ignored span attributes entirely would also pass.
+"""
+
+import logging
+import os
+import time
+import uuid
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+ORG_ID = os.environ.get("TEST_ORG_ID", "default")
+RESOURCE_SERVICE = "pytest_resource_svc"
+SPAN_SERVICE = "pytest_span_override"
+
+
+def test_span_attribute_does_not_overwrite_resource_service_name(create_session, base_url):
+    session = create_session
+    now_ns = int(time.time() * 1e9) - 120 * 10**9
+    colliding_id = uuid.uuid4().hex
+    control_id = uuid.uuid4().hex
+
+    def span(trace_id, name, attrs):
+        return {
+            "traceId": trace_id, "spanId": uuid.uuid4().hex[:16], "parentSpanId": "",
+            "flags": 1, "name": name, "kind": 2,
+            "startTimeUnixNano": str(now_ns), "endTimeUnixNano": str(now_ns + 5 * 10**6),
+            "attributes": attrs, "status": {"code": 0},
+        }
+
+    payload = {"resourceSpans": [{
+        "resource": {"attributes": [
+            {"key": "service.name", "value": {"stringValue": RESOURCE_SERVICE}}]},
+        "scopeSpans": [{"scope": {"name": "pytest"}, "spans": [
+            # Collides with the resource-derived service_name.
+            span(colliding_id, "colliding",
+                 [{"key": "service.name", "value": {"stringValue": SPAN_SERVICE}}]),
+            span(control_id, "control",
+                 [{"key": "http.route", "value": {"stringValue": "/control"}}]),
+        ]}],
+    }]}
+
+    r = session.post(f"{base_url}api/{ORG_ID}/v1/traces", json=payload)
+    assert r.status_code == 200, f"trace ingest failed: {r.status_code} {r.text[:300]}"
+
+    start_us, end_us = int(now_ns / 1000) - 60_000_000, int(time.time() * 1_000_000)
+    rows = []
+    for _ in range(40):
+        q = session.post(
+            f"{base_url}api/{ORG_ID}/_search?type=traces",
+            json={"query": {
+                "sql": 'SELECT trace_id, service_name, operation_name FROM "default" '
+                       f"WHERE trace_id IN ('{colliding_id}','{control_id}')",
+                "start_time": start_us, "end_time": end_us, "from": 0, "size": 20}},
+        )
+        if q.status_code == 200 and len(q.json().get("hits", [])) >= 2:
+            rows = q.json()["hits"]
+            break
+        time.sleep(2)
+    assert len(rows) >= 2, f"seeded spans never became searchable, got {rows}"
+
+    by_trace = {r_["trace_id"]: r_.get("service_name") for r_ in rows}
+    logger.info("service_name by trace: %s", by_trace)
+
+    assert by_trace.get(control_id) == RESOURCE_SERVICE, \
+        f"control span should carry the resource service, got {by_trace.get(control_id)}"
+    # The defect let the span attribute win here.
+    assert by_trace.get(colliding_id) == RESOURCE_SERVICE, \
+        (f"a span attribute must not overwrite the resource service_name: "
+         f"got {by_trace.get(colliding_id)!r}, expected {RESOURCE_SERVICE!r}")
diff --git a/tests/api-testing/tests/metrics/test_otlp_datapoint_flags_and_exp_histogram.py b/tests/api-testing/tests/metrics/test_otlp_datapoint_flags_and_exp_histogram.py
new file mode 100644
index 0000000000..d562a87444
--- /dev/null
+++ b/tests/api-testing/tests/metrics/test_otlp_datapoint_flags_and_exp_histogram.py
@@ -0,0 +1,141 @@
+"""OTLP data-point flags and exponential histograms  [#14637, #14634].
+
+#14637 (PR #14727): bit 0 of `flags` is NO_RECORDED_VALUE -- the point marks a
+gap, e.g. a Prometheus staleness marker translated by the Collector. It was
+turned into a label and the point's value stored as if it were real data, so
+gaps became readings.
+
+#14634 (PR #14724): exponential histograms were flattened into classic buckets
+incorrectly -- each bucket carried its own count rather than the cumulative
+count, `zero_count` was never written so zero-bucket observations were lost, and
+there was no `le="inf"` bucket, which makes `histogram_quantile` return NaN.
+All three are asserted separately because each fails on its own.
+"""
+
+import time
+import uuid
+
+import pytest
+
+from support.wait import wait_until
+
+
+@pytest.fixture(scope="module")
+def flagged_metric(create_session, base_url, org_id):
+    """One point flagged NO_RECORDED_VALUE, one clean point a second later."""
+    session = create_session
+    metric = f"pytest_flags_{uuid.uuid4().hex[:6]}"
+    t_ns = int(time.time() * 1e9) - 180 * 10**9
+
+    payload = {"resourceMetrics": [{
+        "resource": {"attributes": [{"key": "service.name",
+                                     "value": {"stringValue": "pytest_flags"}}]},
+        "scopeMetrics": [{"scope": {"name": "pytest"}, "metrics": [{
+            "name": metric,
+            "gauge": {"dataPoints": [
+                {"asDouble": 99.0, "timeUnixNano": str(t_ns), "flags": 1},
+                {"asDouble": 7.0, "timeUnixNano": str(t_ns + 10**9), "flags": 0},
+            ]},
+        }]}],
+    }]}
+    r = session.post(f"{base_url}api/{org_id}/v1/metrics", json=payload)
+    assert r.status_code == 200, f"OTLP ingest failed: {r.status_code} {r.text[:300]}"
+
+    at = int(t_ns / 1e9) + 60
+
+    def visible():
+        q = session.get(f"{base_url}api/{org_id}/prometheus/api/v1/query",
+                        params={"query": metric, "time": str(at)})
+        return q.status_code == 200 and bool(q.json().get("data", {}).get("result"))
+
+    wait_until(visible, timeout=120, interval=2, msg=f"{metric} never became queryable")
+    yield metric, at
+    session.delete(f"{base_url}api/{org_id}/streams/{metric}?type=metrics")
+
+
+@pytest.fixture(scope="module")
+def exp_histogram(create_session, base_url, org_id):
+    """An exponential histogram with a non-empty zero bucket."""
+    session = create_session
+    metric = f"pytest_exphist_{uuid.uuid4().hex[:6]}"
+    t_ns = int(time.time() * 1e9) - 180 * 10**9
+    zero_count, positive = 3, ["2", "3", "2"]      # total = 3 + 7 = 10
+
+    payload = {"resourceMetrics": [{
+        "resource": {"attributes": [{"key": "service.name",
+                                     "value": {"stringValue": "pytest_exphist"}}]},
+        "scopeMetrics": [{"scope": {"name": "pytest"}, "metrics": [{
+            "name": metric,
+            "exponentialHistogram": {"aggregationTemporality": 2, "dataPoints": [{
+                "startTimeUnixNano": str(t_ns), "timeUnixNano": str(t_ns),
+                "count": "10", "sum": 55.0, "scale": 0,
+                "zeroCount": str(zero_count),
+                "positive": {"offset": 0, "bucketCounts": positive},
+            }]},
+        }]}],
+    }]}
+    r = session.post(f"{base_url}api/{org_id}/v1/metrics", json=payload)
+    assert r.status_code == 200, f"OTLP ingest failed: {r.status_code} {r.text[:300]}"
+
+    at = int(t_ns / 1e9) + 60
+
+    def visible():
+        q = session.get(f"{base_url}api/{org_id}/prometheus/api/v1/query",
+                        params={"query": f"{metric}_bucket", "time": str(at)})
+        return q.status_code == 200 and bool(q.json().get("data", {}).get("result"))
+
+    wait_until(visible, timeout=120, interval=2, msg=f"{metric}_bucket never appeared")
+    yield metric, at, zero_count, 10
+    for suffix in ("", "_count", "_sum", "_bucket"):
+        session.delete(f"{base_url}api/{org_id}/streams/{metric}{suffix}?type=metrics")
+
+
+def _buckets(session, base_url, org_id, metric, at):
+    q = session.get(f"{base_url}api/{org_id}/prometheus/api/v1/query",
+                    params={"query": f"{metric}_bucket", "time": str(at)})
+    assert q.status_code == 200, f"bucket query failed: {q.status_code} {q.text[:300]}"
+    out = {}
+    for s in q.json()["data"]["result"]:
+        out[s["metric"].get("le")] = float(s["value"][1])
+    return out
+
+
+def test_no_recorded_value_point_is_dropped(create_session, base_url, org_id, flagged_metric):
+    """The flagged point marks a gap and must not be stored as a reading."""
+    metric, at = flagged_metric
+    q = create_session.get(f"{base_url}api/{org_id}/prometheus/api/v1/query_range",
+                           params={"query": metric, "start": str(at - 600),
+                                   "end": str(at), "step": "30"})
+    assert q.status_code == 200, f"range query failed: {q.status_code} {q.text[:300]}"
+    values = {v[1] for s in q.json()["data"]["result"] for v in s.get("values", [])}
+
+    assert "7" in values or 7.0 in {float(v) for v in values}, \
+        f"the clean point must survive, got {values}"
+    assert "99" not in values and 99.0 not in {float(v) for v in values}, \
+        f"a NO_RECORDED_VALUE point was stored as real data: {values}"
+
+
+def test_exponential_histogram_buckets_are_cumulative(create_session, base_url, org_id, exp_histogram):
+    metric, at, _zero, total = exp_histogram
+    buckets = _buckets(create_session, base_url, org_id, metric, at)
+
+    finite = sorted(((float(le), v) for le, v in buckets.items() if le not in (None, "inf", "+Inf")),
+                    key=lambda p: p[0])
+    counts = [v for _le, v in finite]
+    assert counts == sorted(counts), \
+        f"bucket counts must be cumulative (non-decreasing), got {finite}"
+    assert counts[-1] == total, \
+        f"the widest finite bucket must reach the total {total}, got {counts[-1]}"
+
+
+def test_exponential_histogram_has_zero_and_inf_buckets(create_session, base_url, org_id, exp_histogram):
+    metric, at, zero_count, total = exp_histogram
+    buckets = _buckets(create_session, base_url, org_id, metric, at)
+
+    inf = next((v for le, v in buckets.items() if le in ("inf", "+Inf")), None)
+    assert inf is not None, f"an le=inf bucket must exist, got {sorted(buckets)}"
+    assert inf == total, f"the inf bucket must carry the total {total}, got {inf}"
+
+    # Zero-bucket observations were silently lost before the fix.
+    assert buckets.get("0") == zero_count, \
+        f"the zero bucket must carry zeroCount {zero_count}, got {buckets.get('0')}"
diff --git a/tests/api-testing/tests/metrics/test_otlp_histogram_optional_stats.py b/tests/api-testing/tests/metrics/test_otlp_histogram_optional_stats.py
new file mode 100644
index 0000000000..a945aa3d11
--- /dev/null
+++ b/tests/api-testing/tests/metrics/test_otlp_histogram_optional_stats.py
@@ -0,0 +1,87 @@
+"""OTLP histogram optional stats must be skipped when absent  [regression #13019].
+
+OTLP's `HistogramDataPoint` defines `sum`, `min` and `max` as optional. The
+ingestion path turned a missing optional into JSON `null` and still emitted the
+`_sum` / `_min` / `_max` record, so those metrics existed with no real value.
+PR #13082 skips the record entirely when the stat is absent, while `_count` and
+the bucket records are always emitted.
+
+Asserting only that `_count` exists would pass on the unfixed build, so the
+contract here is the ABSENCE of `_sum` / `_min` / `_max` for a point that
+carries none -- proven against a second histogram in the same payload that does
+carry them, so a build emitting nothing at all cannot pass either.
+"""
+
+import time
+import uuid
+
+import pytest
+
+from support.wait import wait_until
+
+
+@pytest.fixture(scope="module")
+def histogram_names(create_session, base_url, org_id):
+    """Two histograms: one without sum/min/max, one with, ingested together."""
+    session = create_session
+    bare = f"pytest_hist_bare_{uuid.uuid4().hex[:6]}"
+    full = f"pytest_hist_full_{uuid.uuid4().hex[:6]}"
+    now_ns = int(time.time() * 1e9) - 120 * 10**9
+
+    def point(with_stats):
+        p = {
+            "startTimeUnixNano": str(now_ns),
+            "timeUnixNano": str(now_ns),
+            "count": "6",
+            "bucketCounts": ["2", "3", "1"],
+            "explicitBounds": [1.0, 5.0],
+        }
+        if with_stats:
+            p.update({"sum": 21.0, "min": 0.5, "max": 9.0})
+        return p
+
+    payload = {"resourceMetrics": [{
+        "resource": {"attributes": [
+            {"key": "service.name", "value": {"stringValue": "pytest_hist"}}]},
+        "scopeMetrics": [{"scope": {"name": "pytest"}, "metrics": [
+            {"name": bare, "unit": "1",
+             "histogram": {"dataPoints": [point(False)], "aggregationTemporality": 2}},
+            {"name": full, "unit": "1",
+             "histogram": {"dataPoints": [point(True)], "aggregationTemporality": 2}},
+        ]}],
+    }]}
+
+    resp = session.post(f"{base_url}api/{org_id}/v1/metrics", json=payload)
+    assert resp.status_code == 200, f"OTLP metrics ingest failed: {resp.status_code} {resp.text[:400]}"
+
+    def names():
+        r = session.get(f"{base_url}api/{org_id}/prometheus/api/v1/label/__name__/values")
+        return set(r.json().get("data", [])) if r.status_code == 200 else set()
+
+    wait_until(lambda: f"{full}_sum" in names(), timeout=120, interval=2,
+               msg="the control histogram's _sum never appeared")
+    yield bare, full, names()
+    for base in (bare, full):
+        for suffix in ("", "_count", "_sum", "_min", "_max", "_bucket"):
+            session.delete(f"{base_url}api/{org_id}/streams/{base}{suffix}?type=metrics")
+
+
+def test_absent_optional_stats_emit_no_record(create_session, base_url, org_id, histogram_names):
+    bare, _full, present = histogram_names
+    for suffix in ("sum", "min", "max"):
+        assert f"{bare}_{suffix}" not in present, \
+            f"{bare}_{suffix} must not exist for a point that carries no {suffix}"
+
+
+def test_count_and_buckets_survive(create_session, base_url, org_id, histogram_names):
+    bare, _full, present = histogram_names
+    assert f"{bare}_count" in present, f"{bare}_count must still be emitted"
+    assert f"{bare}_bucket" in present, f"{bare}_bucket must still be emitted"
+
+
+def test_present_optional_stats_are_still_emitted(create_session, base_url, org_id, histogram_names):
+    """Control: a point that DOES carry the stats keeps its _sum/_min/_max."""
+    _bare, full, present = histogram_names
+    for suffix in ("sum", "min", "max"):
+        assert f"{full}_{suffix}" in present, \
+            f"{full}_{suffix} must exist for a point that carries {suffix}"
diff --git a/tests/api-testing/tests/metrics/test_promql_label_metadata.py b/tests/api-testing/tests/metrics/test_promql_label_metadata.py
new file mode 100644
index 0000000000..356469bf25
--- /dev/null
+++ b/tests/api-testing/tests/metrics/test_promql_label_metadata.py
@@ -0,0 +1,127 @@
+"""Prometheus label metadata endpoints  [regression #14635].
+
+Two separate defects, fixed in two PRs, and the shipped behaviour for the second
+is NOT what the issue asked for:
+
+  - #14729 made `/labels` and `/label/__name__/values` honour `start`/`end`.
+    They previously returned an empty list whenever a time range was supplied,
+    which is exactly what Grafana's metric browser sends, so it showed nothing.
+
+  - #14814 made `/label/{name}/values` REJECT a request that does not identify a
+    single metric, with HTTP 400 and a Prometheus `bad_data` error. The issue
+    asked for these to work without `match[]` the way Prometheus does; the fix
+    deliberately chose an explicit error over a cross-stream query, because
+    querying across all metric streams is unsupported. A successful empty list
+    was the bug: it hid that nothing had been queried.
+
+So the asserted contract here is the error, not the values. `/labels` and
+`/label/__name__/values` keep working without `match[]`.
+"""
+import time
+import uuid
+
+import pytest
+
+from support.wait import wait_until
+
+LABEL_NAME = "job"
+JOB_VALUE = "pytest_label_metadata"
+
+
+@pytest.fixture(scope="module")
+def labelled_metric(create_session, base_url, org_id):
+    """Ingest one gauge carrying a non-__name__ label, and wait for it to be queryable."""
+    session = create_session
+    metric = f"pytest_labels_{uuid.uuid4().hex[:8]}"
+    now_sec = int(time.time())
+
+    payload = [
+        {
+            "__name__": metric,
+            "__type__": "gauge",
+            LABEL_NAME: JOB_VALUE,
+            "_timestamp": (now_sec - 60 + i) * 1000,
+            "value": float(i),
+        }
+        for i in range(5)
+    ]
+    resp = session.post(f"{base_url}api/{org_id}/ingest/metrics/_json", json=payload)
+    assert resp.status_code == 200, f"metrics ingest failed: {resp.status_code} {resp.text[:400]}"
+
+    def visible():
+        r = session.get(
+            f"{base_url}api/{org_id}/prometheus/api/v1/label/__name__/values",
+        )
+        return r.status_code == 200 and metric in r.json().get("data", [])
+
+    wait_until(visible, timeout=90, interval=1,
+               msg=f"metric {metric} never appeared in /label/__name__/values")
+    yield metric, now_sec
+    session.delete(f"{base_url}api/{org_id}/streams/{metric}?type=metrics")
+
+
+def test_labels_honours_a_time_range(create_session, base_url, org_id, labelled_metric):
+    """`/labels` must return names with start/end, not the empty list it used to."""
+    session = create_session
+    _, now_sec = labelled_metric
+
+    resp = session.get(
+        f"{base_url}api/{org_id}/prometheus/api/v1/labels",
+        params={"start": str(now_sec - 3600), "end": str(now_sec + 60)},
+    )
+    assert resp.status_code == 200, f"/labels failed: {resp.status_code} {resp.text[:400]}"
+    names = resp.json().get("data", [])
+    assert names, "/labels returned nothing for a range that contains ingested data"
+    assert LABEL_NAME in names, f"expected {LABEL_NAME!r} among label names, got {names}"
+
+
+def test_name_values_honours_a_time_range(create_session, base_url, org_id, labelled_metric):
+    """`/label/__name__/values` must return metric names with start/end."""
+    session = create_session
+    metric, now_sec = labelled_metric
+
+    resp = session.get(
+        f"{base_url}api/{org_id}/prometheus/api/v1/label/__name__/values",
+        params={"start": str(now_sec - 3600), "end": str(now_sec + 60)},
+    )
+    assert resp.status_code == 200, \
+        f"/label/__name__/values failed: {resp.status_code} {resp.text[:400]}"
+    values = resp.json().get("data", [])
+    assert metric in values, \
+        f"{metric} missing from /label/__name__/values over a range containing it, got {values[:20]}"
+
+
+def test_label_values_without_a_metric_is_rejected(create_session, base_url, org_id, labelled_metric):
+    """An unscoped label-values request must error, not answer an empty success."""
+    session = create_session
+    _, now_sec = labelled_metric
+
+    resp = session.get(
+        f"{base_url}api/{org_id}/prometheus/api/v1/label/{LABEL_NAME}/values",
+        params={"start": str(now_sec - 3600), "end": str(now_sec + 60)},
+    )
+    assert resp.status_code == 400, \
+        f"unscoped label values should be rejected with 400, got {resp.status_code}: {resp.text[:400]}"
+    body = resp.json()
+    assert body.get("status") == "error", f"expected a Prometheus error envelope, got {body}"
+    assert "match[]" in body.get("error", ""), \
+        f"the error should tell the caller match[] is required, got {body.get('error')!r}"
+
+
+def test_label_values_with_a_metric_returns_values(create_session, base_url, org_id, labelled_metric):
+    """Scoped by match[], the same request returns the label's values."""
+    session = create_session
+    metric, now_sec = labelled_metric
+
+    resp = session.get(
+        f"{base_url}api/{org_id}/prometheus/api/v1/label/{LABEL_NAME}/values",
+        params={
+            "match[]": metric,
+            "start": str(now_sec - 3600),
+            "end": str(now_sec + 60),
+        },
+    )
+    assert resp.status_code == 200, \
+        f"scoped label values failed: {resp.status_code} {resp.text[:400]}"
+    values = resp.json().get("data", [])
+    assert JOB_VALUE in values, f"expected {JOB_VALUE!r} among {LABEL_NAME} values, got {values}"
diff --git a/tests/api-testing/tests/metrics/test_promql_selectors_and_offset.py b/tests/api-testing/tests/metrics/test_promql_selectors_and_offset.py
new file mode 100644
index 0000000000..0456192843
--- /dev/null
+++ b/tests/api-testing/tests/metrics/test_promql_selectors_and_offset.py
@@ -0,0 +1,100 @@
+"""PromQL offset and label_replace regressions  [#11322, #11321].
+
+Both fixed in PR #11335: an instant query with `offset` returned nothing, and a
+label generated by `label_replace` was not visible to an aggregation grouping by
+it.
+
+#13892 (mixed-case `__name__` selectors) is deliberately NOT covered here: it
+only reproduces with `ZO_FORMAT_STREAM_NAME_TO_LOWERCASE=false`, which is not the
+default, so on a stock server the name is lowercased at ingest and the selector
+legitimately finds nothing. `tests/api-testing/tests/regression/test_promql_mixed_case_metric_name.py`
+already covers the related #13926 -- though note that folder is excluded from the
+integration run, so it never executes.
+"""
+
+import time
+import uuid
+
+import pytest
+
+from support.wait import wait_until
+
+SAMPLES = 6
+SPACING_SEC = 30
+
+
+@pytest.fixture(scope="module")
+def seeded_metric(create_session, base_url, org_id):
+    """A gauge sampled every 30s, ~20 minutes in the past, on two instances."""
+    session = create_session
+    metric = f"pytest_selector_{uuid.uuid4().hex[:6]}"
+    t0 = (int(time.time()) // SPACING_SEC) * SPACING_SEC - 20 * 60
+
+    payload = [
+        {
+            "__name__": metric,
+            "__type__": "gauge",
+            "job": "pytest_selectors",
+            "instance": f"host{i % 2}",
+            "_timestamp": (t0 + i * SPACING_SEC) * 1000,
+            "value": float(i + 1),
+        }
+        for i in range(SAMPLES)
+    ]
+    resp = session.post(f"{base_url}api/{org_id}/ingest/metrics/_json", json=payload)
+    assert resp.status_code == 200, f"metrics ingest failed: {resp.status_code} {resp.text[:400]}"
+
+    def visible():
+        r = session.get(
+            f"{base_url}api/{org_id}/prometheus/api/v1/query",
+            params={"query": f'{{__name__="{metric}"}}', "time": str(t0 + (SAMPLES - 1) * SPACING_SEC)},
+        )
+        return r.status_code == 200 and bool(r.json().get("data", {}).get("result"))
+
+    wait_until(visible, timeout=120, interval=2,
+               msg=f"metric {metric} never became queryable by __name__ selector")
+    yield metric, t0
+    session.delete(f"{base_url}api/{org_id}/streams/{metric}?type=metrics")
+
+
+def _instant(session, base_url, org_id, query, at):
+    r = session.get(f"{base_url}api/{org_id}/prometheus/api/v1/query",
+                    params={"query": query, "time": str(at)})
+    assert r.status_code == 200, f"instant query failed: {r.status_code} {r.text[:300]}"
+    body = r.json()
+    assert body.get("status") == "success", body
+    return body["data"]["result"]
+
+
+def test_instant_query_with_offset_returns_the_earlier_sample(create_session, base_url, org_id, seeded_metric):
+    """`offset` must shift the evaluation back, not return an empty result."""
+    metric, t0 = seeded_metric
+    at = t0 + (SAMPLES - 1) * SPACING_SEC
+
+    now_res = _instant(create_session, base_url, org_id, f'{{__name__="{metric}"}}', at)
+    off_res = _instant(create_session, base_url, org_id,
+                       f'{{__name__="{metric}"}} offset {SPACING_SEC}s', at)
+
+    assert off_res, f"offset query returned nothing (the defect); plain query returned {len(now_res)}"
+    now_val = float(now_res[0]["value"][1])
+    off_val = float(off_res[0]["value"][1])
+    # Values ascend with time, so the offset sample must be strictly earlier/smaller.
+    assert off_val < now_val, \
+        f"offset must evaluate an earlier sample: offset={off_val} current={now_val}"
+
+
+def test_label_replace_output_is_available_to_aggregation(create_session, base_url, org_id, seeded_metric):
+    """A label generated by `label_replace` must be groupable by an aggregation."""
+    metric, t0 = seeded_metric
+    at = t0 + (SAMPLES - 1) * SPACING_SEC
+
+    query = (
+        f'sum by (derived) (label_replace({{__name__="{metric}"}}, '
+        f'"derived", "$1", "instance", "(.*)"))'
+    )
+    result = _instant(create_session, base_url, org_id, query, at)
+    assert result, "aggregation over a label_replace-generated label returned nothing"
+
+    derived = {r["metric"].get("derived") for r in result}
+    assert derived and all(d for d in derived), \
+        f"the generated label must reach the aggregation output, got {derived}"
diff --git a/tests/api-testing/tests/orgs/test_user_and_token_validation.py b/tests/api-testing/tests/orgs/test_user_and_token_validation.py
new file mode 100644
index 0000000000..c8304cbf66
--- /dev/null
+++ b/tests/api-testing/tests/orgs/test_user_and_token_validation.py
@@ -0,0 +1,82 @@
+"""User listing, org validation and default ingestion tokens  [#9993, #1679, #12407].
+
+#9993 (PR #10004): the root user was omitted from the users list on an OSS
+build, so an operator could not see the account they were signed in as.
+
+#1679 (PR #13806): `POST /api/{org_id}/users` accepted an empty or
+whitespace-only org_id path segment, creating user records against a
+meaningless organization.
+
+#12407 (PR #12412): organizations created before ingestion tokens existed had
+none, and no migration backfilled them. The fix creates the default token
+lazily when the list endpoint is first called, so simply asking must yield one.
+"""
+
+import logging
+import os
+import uuid
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+ORG_ID = os.environ.get("TEST_ORG_ID", "default")
+ROOT_EMAIL = os.environ.get("ZO_ROOT_USER_EMAIL", "root@example.com")
+
+
+def test_root_user_is_listed(create_session, base_url):
+    """The signed-in root account must appear in the org's user list."""
+    session = create_session
+    resp = session.get(f"{base_url}api/{ORG_ID}/users")
+    assert resp.status_code == 200, f"user list failed: {resp.status_code} {resp.text[:300]}"
+
+    body = resp.json()
+    rows = body if isinstance(body, list) else body.get("data", body.get("list", []))
+    emails = {(r.get("email") or "").lower() for r in rows if isinstance(r, dict)}
+    logger.info("users listed: %s", sorted(emails))
+    assert ROOT_EMAIL.lower() in emails, \
+        f"root user {ROOT_EMAIL} must be listed, got {sorted(emails)}"
+
+
+def test_ingestion_tokens_are_created_on_demand(create_session, base_url):
+    """Listing tokens must yield the default one even if the org never had it."""
+    session = create_session
+    resp = session.get(f"{base_url}api/{ORG_ID}/ingestion-tokens")
+    assert resp.status_code == 200, \
+        f"ingestion-tokens list failed: {resp.status_code} {resp.text[:300]}"
+
+    body = resp.json()
+    rows = body if isinstance(body, list) else body.get("data", body.get("list", []))
+    logger.info("ingestion tokens returned: %s", len(rows))
+    # The defect returned an empty list for orgs predating the feature.
+    assert rows, "listing ingestion tokens must create and return the default token"
+
+
+def test_blank_org_id_does_not_create_a_user(create_session, base_url):
+    """A whitespace-only org_id must not produce a user record.
+
+    The status alone is a weak signal: on this build org resolution answers 404
+    before `validate_org_id` can return its 400, and either layer refusing is
+    acceptable. What the defect actually produced was a user account against a
+    meaningless organization, so the assertion is that no such account exists.
+    """
+    session = create_session
+    email = f"e2e_blankorg_{uuid.uuid4().hex[:8]}@example.com"
+    payload = {
+        "email": email,
+        "password": "Complexpass#123",
+        "role": "admin",
+        "first_name": "blank",
+        "last_name": "org",
+    }
+    resp = session.post(f"{base_url}api/%20/users", json=payload)
+    logger.info("blank org_id user create answered %s", resp.status_code)
+    assert resp.status_code >= 400, \
+        f"a whitespace org_id must be rejected, got {resp.status_code}: {resp.text[:300]}"
+
+    listing = session.get(f"{base_url}api/{ORG_ID}/users")
+    assert listing.status_code == 200, f"user list failed: {listing.status_code}"
+    body = listing.json()
+    rows = body if isinstance(body, list) else body.get("data", body.get("list", []))
+    emails = {(r.get("email") or "").lower() for r in rows if isinstance(r, dict)}
+    assert email.lower() not in emails, \
+        f"{email} was created despite the blank org_id being rejected"
diff --git a/tests/api-testing/tests/pipelines/test_invalid_function_error.py b/tests/api-testing/tests/pipelines/test_invalid_function_error.py
new file mode 100644
index 0000000000..b8df9736eb
--- /dev/null
+++ b/tests/api-testing/tests/pipelines/test_invalid_function_error.py
@@ -0,0 +1,58 @@
+"""A malformed function must error, not panic  [regression #7708].
+
+Adding an incorrect function surfaced a panic rather than a usable error, so the
+caller saw a failure with no indication of what was wrong -- and a panicking
+request path is a liveness risk.
+
+Asserting only the status would be weak: the contract is a *usable* error (a
+message naming the problem) AND a server still answering afterwards, since a
+panic can take the request down while the process restarts behind it.
+"""
+
+import logging
+import os
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+ORG_ID = os.environ.get("TEST_ORG_ID", "default")
+BAD_VRL = ".x = this is not valid vrl ((("
+
+
+def _alive(session, base_url):
+    return session.get(f"{base_url}api/{ORG_ID}/streams", timeout=30).status_code == 200
+
+
+def test_malformed_function_returns_a_usable_error(create_session, base_url):
+    session = create_session
+    assert _alive(session, base_url), "server was already unhealthy before the request"
+
+    resp = session.post(
+        f"{base_url}api/{ORG_ID}/functions/test",
+        json={"function": BAD_VRL, "events": [{"a": 1}]},
+    )
+    logger.info("malformed function answered %s", resp.status_code)
+
+    assert 400 <= resp.status_code < 500, \
+        f"a malformed function should be a client error, got {resp.status_code}: {resp.text[:300]}"
+
+    body = resp.text.lower()
+    assert "error" in body or "syntax" in body, \
+        f"the response should explain what is wrong, got {resp.text[:300]}"
+    assert "panic" not in body, f"a panic leaked into the response: {resp.text[:300]}"
+
+    # A panicking path can answer and still have taken the worker down.
+    assert _alive(session, base_url), "server stopped answering after the malformed function"
+
+
+def test_valid_function_still_works(create_session, base_url):
+    """Control: the rejection is about the function, not all requests."""
+    session = create_session
+    resp = session.post(
+        f"{base_url}api/{ORG_ID}/functions/test",
+        json={"function": '.added = "ok"\n.', "events": [{"a": 1}]},
+    )
+    assert resp.status_code == 200, f"a valid function must succeed: {resp.status_code} {resp.text[:300]}"
+    results = resp.json().get("results", [])
+    assert results and results[0].get("event", {}).get("added") == "ok", \
+        f"the valid function did not apply: {resp.text[:300]}"
diff --git a/tests/api-testing/tests/pipelines/test_vrl_parse_json_precision.py b/tests/api-testing/tests/pipelines/test_vrl_parse_json_precision.py
new file mode 100644
index 0000000000..76de17634a
--- /dev/null
+++ b/tests/api-testing/tests/pipelines/test_vrl_parse_json_precision.py
@@ -0,0 +1,63 @@
+"""VRL parse_json must not leak serde_json's private number wrapper  [#10919].
+
+With `serde_json` built with `arbitrary_precision`, a high-precision number
+deserialises into an object keyed `$serde_json::private::Number` rather than a
+scalar. `parse_json` in VRL passed that object straight through, so flattening
+produced a field literally named `__serde_json__private__number` instead of the
+value. PR #10920 detects the wrapper and converts it to a float.
+
+The assertion is therefore twofold: the parsed value must be numeric, AND no
+field carrying the private wrapper's name may appear anywhere in the event --
+checking only the value would miss the wrapper leaking alongside it.
+"""
+
+import logging
+import os
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+ORG_ID = os.environ.get("TEST_ORG_ID", "default")
+
+# Enough digits that serde_json cannot hold it as a plain f64/i64.
+HIGH_PRECISION = "12345678901234567890.123456789"
+
+
+def _run_vrl(session, base_url, function, events):
+    resp = session.post(
+        f"{base_url}api/{ORG_ID}/functions/test",
+        json={"function": function, "events": events},
+    )
+    assert resp.status_code == 200, \
+        f"function test failed: {resp.status_code} {resp.text[:300]}"
+    results = resp.json().get("results", [])
+    assert results, f"no results returned: {resp.text[:300]}"
+    return results[0].get("event", {})
+
+
+def test_parse_json_yields_a_number_for_high_precision_input(create_session, base_url):
+    event = _run_vrl(
+        create_session, base_url,
+        ".parsed = parse_json!(.payload)\n.",
+        [{"payload": '{"big": ' + HIGH_PRECISION + '}'}],
+    )
+    logger.info("parsed event keys: %s", sorted(event.keys()))
+
+    assert "parsed_big" in event, f"the parsed number is missing: {event}"
+    assert isinstance(event["parsed_big"], (int, float)), \
+        f"parse_json must yield a number, got {type(event['parsed_big']).__name__}: {event['parsed_big']!r}"
+
+
+def test_no_serde_json_private_wrapper_leaks_into_fields(create_session, base_url):
+    """The wrapper object must not survive as a field name."""
+    event = _run_vrl(
+        create_session, base_url,
+        ".parsed = parse_json!(.payload)\n.",
+        [{"payload": '{"big": ' + HIGH_PRECISION + ', "nested": {"inner": ' + HIGH_PRECISION + '}}'}],
+    )
+    leaked = [k for k in event if "serde_json" in k.lower() or "private" in k.lower()]
+    logger.info("event keys: %s", sorted(event.keys()))
+    assert not leaked, f"serde_json's private number wrapper leaked as field(s): {leaked}"
+
+    assert isinstance(event.get("parsed_nested_inner"), (int, float)), \
+        f"a nested high-precision number must also parse as a number: {event.get('parsed_nested_inner')!r}"
diff --git a/tests/api-testing/tests/rum/test_rum_forwarded_client_ip.py b/tests/api-testing/tests/rum/test_rum_forwarded_client_ip.py
new file mode 100644
index 0000000000..e84424e31e
--- /dev/null
+++ b/tests/api-testing/tests/rum/test_rum_forwarded_client_ip.py
@@ -0,0 +1,80 @@
+"""RUM must record the real client IP, not the proxy's  [regression #11377].
+
+Behind a proxy every RUM record was attributed to the proxy (or loopback), so
+per-user geo and IP were meaningless. PR #11378 resolves the address through
+`RealIp`, which honours the forwarded header, and stores it as `ip`.
+
+The test sends the beacon with an explicit `X-Forwarded-For` and asserts the
+stored value is that address rather than a loopback one -- asserting merely that
+`ip` is non-empty would pass on the unfixed build, which happily stored the
+proxy address.
+"""
+
+import logging
+import os
+import time
+import uuid
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+ORG_ID = os.environ.get("TEST_ORG_ID", "default")
+FORWARDED_IP = "203.0.113.7"          # TEST-NET-3, never a real client
+LOOPBACK = {"127.0.0.1", "::1"}
+
+
+def _rum_token(session, base_url):
+    r = session.get(f"{base_url}api/{ORG_ID}/rumtoken")
+    assert r.status_code == 200, f"rum token fetch failed: {r.status_code} {r.text[:200]}"
+    body = r.json()
+    token = (body.get("data") or {}).get("rum_token") or body.get("rum_token")
+    assert token, f"no rum token in response: {body}"
+    return token
+
+
+def test_forwarded_client_ip_is_recorded(create_session, base_url):
+    session = create_session
+    token = _rum_token(session, base_url)
+    service = f"pytest_realip_{uuid.uuid4().hex[:8]}"
+
+    beacon = {
+        "type": "view",
+        "date": int(time.time() * 1000),
+        "service": service,
+        "application": {"id": "pytest_realip"},
+        "session": {"id": uuid.uuid4().hex},
+        "view": {"id": uuid.uuid4().hex, "url": "http://example.test/"},
+    }
+    resp = session.post(
+        f"{base_url}rum/v1/{ORG_ID}/rum",
+        params={"oo-api-key": token},
+        headers={"X-Forwarded-For": FORWARDED_IP},
+        json=beacon,
+    )
+    assert resp.status_code in (200, 202), \
+        f"rum ingest failed: {resp.status_code} {resp.text[:300]}"
+
+    # RUM records carry their own `date`, so query a wide window rather than
+    # assuming the row lands near "now".
+    now_us = int(time.time() * 1_000_000)
+    row = None
+    for _ in range(30):
+        q = session.post(
+            f"{base_url}api/{ORG_ID}/_search?type=logs",
+            json={"query": {
+                "sql": f"SELECT ip, service FROM \"_rumdata\" WHERE service = '{service}'",
+                "start_time": now_us - 86_400_000_000, "end_time": now_us + 60_000_000,
+                "from": 0, "size": 5}},
+        )
+        if q.status_code == 200 and q.json().get("hits"):
+            row = q.json()["hits"][0]
+            break
+        time.sleep(2)
+
+    assert row, f"the RUM beacon for {service} never became searchable"
+    logger.info("stored RUM ip: %s", row.get("ip"))
+
+    assert row.get("ip") not in LOOPBACK, \
+        f"the proxy/loopback address was recorded instead of the client: {row.get('ip')}"
+    assert row.get("ip") == FORWARDED_IP, \
+        f"expected the forwarded client IP {FORWARDED_IP}, got {row.get('ip')!r}"
diff --git a/tests/api-testing/tests/rum/test_sourcemap_api.py b/tests/api-testing/tests/rum/test_sourcemap_api.py
index ab8c38858a..f3ba8b5aec 100644
--- a/tests/api-testing/tests/rum/test_sourcemap_api.py
+++ b/tests/api-testing/tests/rum/test_sourcemap_api.py
@@ -11,13 +11,12 @@ Test Coverage:
 - Delete sourcemaps
 
 Prerequisites:
-- OpenObserve ENTERPRISE build running on ZO_BASE_URL
+- OpenObserve running on ZO_BASE_URL (either edition; the sourcemaps routes are not enterprise-gated)
 - RUM enabled (ZO_RUM_ENABLED=true)
 - Static sourcemap fixtures at tests/api-testing/fixtures/sourcemaps/
 
 Note:
 - Tests use pre-built sourcemaps (no build step required)
-- Tests marked with @pytest.mark.skip for OSS CI (enterprise-only feature)
 - Tests must run serially due to shared module-scoped fixture
 - Use: pytest test_sourcemap_api.py -v (no -n flag for parallel)
 """
@@ -25,10 +24,11 @@ Note:
 import pytest
 import logging
 
-# Mark all tests in this module to run serially and skip in OSS (enterprise-only feature)
+# Serial: these cases share one sourcemap store. They run on OSS too -- the
+# sourcemaps routes are outside the enterprise cfg block in router/mod.rs, and
+# the previous blanket skip (which never ran them on ANY build) was mistaken.
 pytestmark = [
     pytest.mark.order(1),
-    pytest.mark.skip(reason="Sourcemaps is an enterprise feature and cannot be tested in CI with non-enterprise build")
 ]
 from helpers.sourcemap_helpers import (
     load_static_sourcemaps,
diff --git a/tests/api-testing/tests/rum/test_sourcemap_rbac.py b/tests/api-testing/tests/rum/test_sourcemap_rbac.py
index a9c4a1c23c..29657cb5b1 100644
--- a/tests/api-testing/tests/rum/test_sourcemap_rbac.py
+++ b/tests/api-testing/tests/rum/test_sourcemap_rbac.py
@@ -19,7 +19,8 @@ Prerequisites:
 - OpenFGA enabled for RBAC enforcement
 
 Note:
-- Tests marked with @pytest.mark.skip for OSS CI (enterprise-only feature)
+- Skipped at runtime by `_require_custom_roles` when the build has no custom
+  roles (GET /roles answers 403 on OSS); they run on enterprise
 - Tests must run serially due to shared module-scoped fixture
 - Uses session-based authentication (matches enterprise RBAC test pattern)
 """
@@ -28,10 +29,8 @@ import pytest
 import logging
 import requests
 
-# Mark all tests in this module to run serially and skip in OSS (enterprise-only feature)
 pytestmark = [
     pytest.mark.order(2),
-    pytest.mark.skip(reason="Sourcemaps is an enterprise feature and cannot be tested in CI with non-enterprise build")
 ]
 
 logger = logging.getLogger(__name__)
@@ -40,6 +39,32 @@ logger = logging.getLogger(__name__)
 test_users = {}
 
 
+@pytest.fixture(scope="module", autouse=True)
+def _require_custom_roles(create_session, base_url):
+    """Skip on builds without custom roles, rather than skipping everywhere.
+
+    These cases need viewer/editor/admin accounts. An OSS build has no custom
+    roles, so the users cannot exist and every case errors at login.
+
+    This replaces an unconditional `pytest.mark.skip`, which skipped the module
+    on EVERY build -- enterprise included -- so these cases had never run
+    anywhere. The capability is probed rather than assumed from the edition, in
+    the same way `test_anomaly_conditions.py` probes its endpoint.
+
+    The probe is deliberately READ-ONLY. Creating and deleting a probe user
+    would mutate the shared `default` org from an autouse fixture, in parallel
+    (xdist) with the RBAC suite that tests user permissions -- `GET /roles`
+    answers 403 "Not Supported" on OSS and lists roles on enterprise, which is
+    the same signal without touching anything.
+    """
+    probe = create_session.get(f"{base_url}api/default/roles")
+    if probe.status_code != 200:
+        pytest.skip(
+            f"custom roles are not available on this build "
+            f"(GET /roles -> {probe.status_code})"
+        )
+
+
 @pytest.fixture(scope="module", autouse=True)
 def setup_rbac_users(create_session, base_url, org_id):
     """
@@ -61,7 +86,10 @@ def setup_rbac_users(create_session, base_url, org_id):
 
     for role in roles:
         email = f"{role}@sourcemap-test.local"
-        password = "TestPass123!"
+        # The enterprise password policy requires at least 15 characters; the
+        # previous 12-character value was rejected with HTTP 400, so every case
+        # then failed at login.
+        password = "TestPass123!Secure"
 
         logger.info(f"Creating user: {email} with role: {role}")
 
@@ -85,8 +113,18 @@ def setup_rbac_users(create_session, base_url, org_id):
             # Create authenticated session via login endpoint
             user_session = create_user_session(email, password, url)
             if not user_session:
+                # Fail loudly rather than skip. The module gate already
+                # established that this build supports custom roles, so failing
+                # to provision one here is a real defect -- skipping would hide
+                # it, which is how the 15-character password policy went
+                # unnoticed. The create response is included so the cause is
+                # visible without re-running.
                 logger.error(f"Failed to create session for {email}")
-                raise Exception(f"Login failed for {email}")
+                raise AssertionError(
+                    f"could not provision the {role} user ({email}) on a build that "
+                    f"reports custom-role support; create returned "
+                    f"{response.status_code}: {response.text[:200]}"
+                )
 
             # Store user info
             test_users[role] = {
diff --git a/tests/api-testing/tests/rum/test_sourcemap_zip_validation.py b/tests/api-testing/tests/rum/test_sourcemap_zip_validation.py
new file mode 100644
index 0000000000..4ccf1755ad
--- /dev/null
+++ b/tests/api-testing/tests/rum/test_sourcemap_zip_validation.py
@@ -0,0 +1,99 @@
+"""A sourcemap ZIP with no valid pairs must error  [regression #12951].
+
+Uploading a ZIP that contained no usable source-map files reported success, so
+an operator believed their maps were in place while error stack traces stayed
+unminified. PR #12962 makes the upload fail instead.
+
+The valid-ZIP control is what stops a build that rejects every upload from
+passing: the error must be specific to there being nothing usable inside.
+"""
+
+import io
+import logging
+import os
+import uuid
+import zipfile
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+ORG_ID = os.environ.get("TEST_ORG_ID", "default")
+
+
+def _zip(entries: dict[str, str]) -> bytes:
+    buf = io.BytesIO()
+    with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
+        for name, body in entries.items():
+            zf.writestr(name, body)
+    return buf.getvalue()
+
+
+def _upload(session, base_url, data: bytes, filename: str, service: str):
+    """Upload under an explicit service so the group can be deleted afterwards."""
+    return session.post(
+        f"{base_url}api/{ORG_ID}/sourcemaps",
+        files={"file": (filename, data, "application/zip")},
+        data={"service": service, "env": "pytest", "version": "1.0.0"},
+    )
+
+
+def _delete_group(session, base_url, service: str):
+    return session.delete(
+        f"{base_url}api/{ORG_ID}/sourcemaps",
+        params={"service": service, "env": "pytest", "version": "1.0.0"},
+    )
+
+
+def _listed_count(session, base_url) -> int:
+    r = session.get(f"{base_url}api/{ORG_ID}/sourcemaps")
+    if r.status_code != 200:
+        return -1
+    body = r.json()
+    rows = body if isinstance(body, list) else body.get("data", body.get("list", []))
+    return len(rows) if isinstance(rows, list) else -1
+
+
+def test_zip_without_any_sourcemap_is_rejected(create_session, base_url):
+    session = create_session
+    data = _zip({"README.txt": "no maps here", "notes/other.md": "# nothing"})
+    service = f"pytest_nomaps_{uuid.uuid4().hex[:6]}"
+    before = _listed_count(session, base_url)
+    resp = _upload(session, base_url, data, "nomaps.zip", service)
+    logger.info("no-sourcemap ZIP answered %s: %s", resp.status_code, resp.text[:160])
+
+    # Reporting success for a ZIP with nothing usable inside was the defect.
+    assert resp.status_code >= 400, \
+        f"a ZIP with no sourcemap must be rejected, got {resp.status_code}: {resp.text[:300]}"
+    assert "sourcemap" in resp.text.lower(), \
+        f"the error should say what was missing, got {resp.text[:300]}"
+
+    # A status alone would not catch a build that errored yet still stored something.
+    after = _listed_count(session, base_url)
+    assert after == before, \
+        f"a rejected upload must store nothing: sourcemap count went {before} -> {after}"
+
+
+def test_zip_with_a_valid_sourcemap_is_accepted(create_session, base_url):
+    """Control: the rejection must be about the contents, not all uploads."""
+    session = create_session
+    smap = (
+        '{"version":3,"file":"app.min.js","sources":["app.js"],'
+        '"names":[],"mappings":"AAAA","sourcesContent":["console.log(1)"]}'
+    )
+    data = _zip({"app.min.js.map": smap, "app.min.js": "console.log(1)\n"})
+    service = f"pytest_withmap_{uuid.uuid4().hex[:6]}"
+    before = _listed_count(session, base_url)
+    try:
+        resp = _upload(session, base_url, data, "withmap.zip", service)
+        logger.info("valid ZIP answered %s: %s", resp.status_code, resp.text[:160])
+
+        assert resp.status_code < 400, \
+            f"a ZIP containing a real sourcemap must be accepted, got {resp.status_code}: {resp.text[:300]}"
+
+        # Accepting and then storing nothing would be the same failure wearing a 2xx.
+        after = _listed_count(session, base_url)
+        assert after > before, \
+            f"an accepted upload must store the map: sourcemap count went {before} -> {after}"
+    finally:
+        # This test is the only one here that stores anything, so it reaps its own group.
+        _delete_group(session, base_url, service)
diff --git a/tests/api-testing/tests/search/test_cache_respects_query_size.py b/tests/api-testing/tests/search/test_cache_respects_query_size.py
new file mode 100644
index 0000000000..25d5d066a2
--- /dev/null
+++ b/tests/api-testing/tests/search/test_cache_respects_query_size.py
@@ -0,0 +1,72 @@
+"""The result cache must key on query size  [regression #10382].
+
+`query.size` was missing from the cache hash, so a second search for the same
+SQL over the same range with a DIFFERENT size was served the first size's cached
+rows. PR #10384 includes size in the hash.
+
+Order matters: the small query must run first so its result is what a
+size-blind cache would wrongly replay for the larger one.
+"""
+
+import logging
+import os
+import time
+import uuid
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+ORG_ID = os.environ.get("TEST_ORG_ID", "default")
+
+def _drop(session, base_url, stream, stream_type="logs"):
+    """Follows the suite convention: the test that seeds a stream drops it."""
+    session.delete(f"{base_url}api/{ORG_ID}/streams/{stream}?type={stream_type}")
+
+ROW_COUNT = 40
+SMALL, LARGE = 5, 25
+
+
+def _seed(session, base_url, stream):
+    now = int(time.time() * 1_000_000)
+    rows = [{"_timestamp": now - i * 1_000_000, "job": "pytest_cache_size", "n": i}
+            for i in range(ROW_COUNT)]
+    r = session.post(f"{base_url}api/{ORG_ID}/{stream}/_json", json=rows)
+    assert r.status_code == 200, f"ingest failed: {r.status_code} {r.text[:300]}"
+    start, end = now - 3_600_000_000, now + 60_000_000
+    for _ in range(30):
+        q = session.post(f"{base_url}api/{ORG_ID}/_search?type=logs",
+                         json={"query": {"sql": f'SELECT count(*) AS n FROM "{stream}"',
+                                         "start_time": start, "end_time": end,
+                                         "from": 0, "size": 1}})
+        if q.status_code == 200 and (q.json().get("hits") or [{}])[0].get("n", 0) >= ROW_COUNT:
+            return start, end
+        time.sleep(1)
+    raise AssertionError(f"stream {stream} never became searchable")
+
+
+def test_cached_result_is_not_replayed_for_a_different_size(create_session, base_url):
+    session = create_session
+    stream = f"e2e_cachesize_{uuid.uuid4().hex[:8]}"
+    try:
+        start, end = _seed(session, base_url, stream)
+        sql = f'SELECT * FROM "{stream}" ORDER BY _timestamp DESC'
+
+        def run(size):
+            r = session.post(
+                f"{base_url}api/{ORG_ID}/_search?type=logs&use_cache=true",
+                json={"query": {"sql": sql, "start_time": start, "end_time": end,
+                                "from": 0, "size": size}},
+            )
+            assert r.status_code == 200, f"search failed: {r.status_code} {r.text[:300]}"
+            return len(r.json().get("hits", []))
+
+        small = run(SMALL)
+        large = run(LARGE)
+        logger.info("size %s -> %s rows; size %s -> %s rows", SMALL, small, LARGE, large)
+
+        assert small == SMALL, f"first query must return {SMALL} rows, got {small}"
+        # A size-blind cache replays the first result here, returning SMALL again.
+        assert large == LARGE, \
+            f"second query must return {LARGE} rows, got {large} (cache ignored query size)"
+    finally:
+        _drop(session, base_url, stream)
diff --git a/tests/api-testing/tests/search/test_match_all_disjunction_stability.py b/tests/api-testing/tests/search/test_match_all_disjunction_stability.py
new file mode 100644
index 0000000000..e3370c41b2
--- /dev/null
+++ b/tests/api-testing/tests/search/test_match_all_disjunction_stability.py
@@ -0,0 +1,106 @@
+"""Many OR-ed match_all() conditions must not abort the process  [regression #12398].
+
+A single search that ORs together several `match_all()` terms aborted the whole
+node with `fatal runtime error: stack overflow` -- one query could take the
+server down, and it then restarted. Fixed in PR #14469 by hardening the leader
+and follower against deeply nested full-text disjunctions.
+
+The contract the issue states is "the query runs, or fails gracefully with an
+error -- it must not abort". So the assertion is deliberately NOT that the query
+succeeds: a well-formed error is an acceptable outcome. What must hold is that
+the server is still answering afterwards, which is the half a status-code check
+alone would miss -- a crashed node restarts, so a later request can succeed
+while the query still killed the process.
+"""
+
+import logging
+import os
+import time
+import uuid
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+ORG_ID = os.environ.get("TEST_ORG_ID", "default")
+
+def _drop(session, base_url, stream, stream_type="logs"):
+    """Follows the suite convention: the test that seeds a stream drops it."""
+    session.delete(f"{base_url}api/{ORG_ID}/streams/{stream}?type={stream_type}")
+
+ROW_COUNT = 60
+TERMS = ["error", "Error", "exception", "Exception",
+         "Traceback", "FATAL", "panic", "CRITICAL"]
+
+
+def _seed(session, base_url, stream):
+    now = int(time.time() * 1_000_000)
+    rows = [
+        {
+            "_timestamp": now - i * 1_000_000,
+            "container_name": f"c{i % 4}",
+            "project": f"p{i % 3}",
+            "message": f"{TERMS[i % len(TERMS)]} something happened {i}",
+        }
+        for i in range(ROW_COUNT)
+    ]
+    resp = session.post(f"{base_url}api/{ORG_ID}/{stream}/_json", json=rows)
+    assert resp.status_code == 200, f"ingest failed: {resp.status_code} {resp.text[:400]}"
+
+    for _ in range(30):
+        r = session.post(
+            f"{base_url}api/{ORG_ID}/_search?type=logs",
+            json={"query": {
+                "sql": f'SELECT count(*) AS n FROM "{stream}"',
+                "start_time": now - 3_600_000_000, "end_time": now + 60_000_000,
+                "from": 0, "size": 1,
+            }},
+        )
+        if r.status_code == 200 and (r.json().get("hits") or [{}])[0].get("n", 0) >= ROW_COUNT:
+            return now
+        time.sleep(1)
+    raise AssertionError(f"stream {stream} never became searchable")
+
+
+def _server_is_alive(session, base_url) -> bool:
+    r = session.get(f"{base_url}api/{ORG_ID}/streams", timeout=30)
+    return r.status_code == 200
+
+
+def test_ored_match_all_terms_do_not_abort_the_server(create_session, base_url):
+    session = create_session
+    stream = f"e2e_matchall_or_{uuid.uuid4().hex[:8]}"
+    try:
+        now = _seed(session, base_url, stream)
+
+        assert _server_is_alive(session, base_url), "server was already unhealthy before the query"
+
+        disjunction = " OR ".join(f"match_all('{t}')" for t in TERMS)
+        sql = (
+            f'SELECT project AS proj, container_name, count(*) AS n FROM "{stream}" '
+            f"WHERE ({disjunction}) GROUP BY proj, container_name ORDER BY n DESC"
+        )
+
+        resp = session.post(
+            f"{base_url}api/{ORG_ID}/_search?type=logs",
+            json={"query": {
+                "sql": sql,
+                "start_time": now - 3_600_000_000, "end_time": now + 60_000_000,
+                "from": 0, "size": 100,
+            }},
+        )
+        logger.info("disjunction query answered %s", resp.status_code)
+
+        # A graceful error is acceptable; a dead connection is not.
+        assert resp.status_code < 500 or resp.text, \
+            f"query returned {resp.status_code} with no body, which suggests the node died"
+
+        # The real contract: the process is still serving.
+        assert _server_is_alive(session, base_url), \
+            "server stopped answering after the OR-ed match_all query, i.e. the query aborted it"
+
+        if resp.status_code == 200:
+            hits = resp.json().get("hits", [])
+            assert isinstance(hits, list), f"unexpected result shape: {resp.text[:300]}"
+            logger.info("query succeeded with %d grouped rows", len(hits))
+    finally:
+        _drop(session, base_url, stream)
diff --git a/tests/api-testing/tests/search/test_result_cache_no_duplicates.py b/tests/api-testing/tests/search/test_result_cache_no_duplicates.py
new file mode 100644
index 0000000000..e1ad4a9dd2
--- /dev/null
+++ b/tests/api-testing/tests/search/test_result_cache_no_duplicates.py
@@ -0,0 +1,67 @@
+"""The result cache must not return duplicate records  [regression #4274].
+
+Overlapping cached ranges replayed rows that the fresh half of the query also
+returned, so a search spanning a cached window came back with duplicates. PR
+#4293 dedupes on timestamp.
+
+The first query must be cached and the second must OVERLAP it -- a
+non-overlapping range exercises nothing, and running only one query cannot
+produce the duplication at all.
+"""
+
+import logging
+import os
+import time
+import uuid
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+ORG_ID = os.environ.get("TEST_ORG_ID", "default")
+
+def _drop(session, base_url, stream, stream_type="logs"):
+    """Follows the suite convention: the test that seeds a stream drops it."""
+    session.delete(f"{base_url}api/{ORG_ID}/streams/{stream}?type={stream_type}")
+
+ROW_COUNT = 40
+
+
+def test_overlapping_cached_ranges_do_not_duplicate_rows(create_session, base_url):
+    session = create_session
+    stream = f"e2e_cachedup_{uuid.uuid4().hex[:8]}"
+    try:
+        now = int(time.time() * 1_000_000)
+        rows = [{"_timestamp": now - i * 1_000_000, "job": "pytest_cache_dup", "n": i}
+                for i in range(ROW_COUNT)]
+        r = session.post(f"{base_url}api/{ORG_ID}/{stream}/_json", json=rows)
+        assert r.status_code == 200, f"ingest failed: {r.status_code} {r.text[:300]}"
+
+        oldest = now - (ROW_COUNT - 1) * 1_000_000
+        sql = f'SELECT _timestamp, n FROM "{stream}" ORDER BY _timestamp DESC'
+
+        def run(start, end):
+            resp = session.post(
+                f"{base_url}api/{ORG_ID}/_search?type=logs&use_cache=true",
+                json={"query": {"sql": sql, "start_time": start, "end_time": end,
+                                "from": 0, "size": 200}},
+            )
+            assert resp.status_code == 200, f"search failed: {resp.status_code} {resp.text[:300]}"
+            return resp.json().get("hits", [])
+
+        for _ in range(30):
+            if len(run(oldest - 60_000_000, now + 60_000_000)) >= ROW_COUNT:
+                break
+            time.sleep(1)
+
+        # Warm the cache over the inner half, then query a window that overlaps it.
+        inner = run(oldest + 10 * 1_000_000, now - 5 * 1_000_000)
+        outer = run(oldest - 60_000_000, now + 60_000_000)
+        logger.info("inner=%s outer=%s rows", len(inner), len(outer))
+
+        stamps = [h["_timestamp"] for h in outer]
+        dupes = {t for t in stamps if stamps.count(t) > 1}
+        assert not dupes, f"{len(dupes)} timestamps were returned more than once: {sorted(dupes)[:5]}"
+        assert len(stamps) == len(set(stamps)) == ROW_COUNT, \
+            f"expected {ROW_COUNT} unique rows, got {len(stamps)} ({len(set(stamps))} unique)"
+    finally:
+        _drop(session, base_url, stream)
diff --git a/tests/api-testing/tests/search/test_sql_limit_and_stream_names.py b/tests/api-testing/tests/search/test_sql_limit_and_stream_names.py
new file mode 100644
index 0000000000..8117686f2a
--- /dev/null
+++ b/tests/api-testing/tests/search/test_sql_limit_and_stream_names.py
@@ -0,0 +1,125 @@
+"""SQL LIMIT precedence and awkward stream names  [regressions #8372, #10378].
+
+#8372 (PR #14473): an explicit SQL `LIMIT` must win over the request's `size`.
+`_search` already honoured it because `AddSortAndLimit` leaves an existing Limit
+node alone; the streaming path did not. The contract asserted here is the one
+both paths must agree on -- LIMIT decides, size does not override it.
+
+#10378 (PR #10394): a stream whose name contains `:` produced a ParserError
+because the name was interpolated into SQL unquoted. Quoting it fixes the parse,
+so the assertion is that such a stream is queryable at all.
+"""
+
+import logging
+import os
+import time
+import uuid
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+ORG_ID = os.environ.get("TEST_ORG_ID", "default")
+
+def _drop(session, base_url, stream, stream_type="logs"):
+    """Follows the suite convention: the test that seeds a stream drops it."""
+    session.delete(f"{base_url}api/{ORG_ID}/streams/{stream}?type={stream_type}")
+
+ROW_COUNT = 30
+SQL_LIMIT = 7
+REQUEST_SIZE = 100
+
+
+def _window():
+    now = int(time.time() * 1_000_000)
+    return now - 3_600_000_000, now + 60_000_000
+
+
+def _seed(session, base_url, stream, rows=ROW_COUNT):
+    now = int(time.time() * 1_000_000)
+    payload = [
+        {"_timestamp": now - i * 1_000_000, "job": "pytest_sql_limit", "n": i,
+         "message": f"row {i}"}
+        for i in range(rows)
+    ]
+    resp = session.post(f"{base_url}api/{ORG_ID}/{stream}/_json", json=payload)
+    assert resp.status_code == 200, f"ingest failed: {resp.status_code} {resp.text[:400]}"
+
+    start, end = _window()
+    for _ in range(30):
+        r = session.post(
+            f"{base_url}api/{ORG_ID}/_search?type=logs",
+            json={"query": {"sql": f'SELECT count(*) AS n FROM "{stream}"',
+                            "start_time": start, "end_time": end, "from": 0, "size": 1}},
+        )
+        if r.status_code == 200 and (r.json().get("hits") or [{}])[0].get("n", 0) >= rows:
+            return
+        time.sleep(1)
+    raise AssertionError(f"stream {stream} never became searchable")
+
+
+def test_explicit_sql_limit_wins_over_request_size(create_session, base_url):
+    """`LIMIT 7` with `size: 100` must return 7 rows, not 100."""
+    session = create_session
+    stream = f"e2e_sqllimit_{uuid.uuid4().hex[:8]}"
+    try:
+        _seed(session, base_url, stream)
+        start, end = _window()
+
+        resp = session.post(
+            f"{base_url}api/{ORG_ID}/_search?type=logs",
+            json={"query": {
+                "sql": f'SELECT * FROM "{stream}" ORDER BY _timestamp DESC LIMIT {SQL_LIMIT}',
+                "start_time": start, "end_time": end, "from": 0, "size": REQUEST_SIZE,
+            }},
+        )
+        assert resp.status_code == 200, f"search failed: {resp.status_code} {resp.text[:400]}"
+        hits = resp.json().get("hits", [])
+        logger.info("LIMIT %s with size %s returned %s rows", SQL_LIMIT, REQUEST_SIZE, len(hits))
+
+        # There are more rows than the LIMIT, so a build that ignores it returns more.
+        assert len(hits) == SQL_LIMIT, \
+            f"explicit SQL LIMIT {SQL_LIMIT} must decide the row count, got {len(hits)}"
+    finally:
+        _drop(session, base_url, stream)
+
+def test_request_size_still_applies_without_an_explicit_limit(create_session, base_url):
+    """A negative control: with no SQL LIMIT, `size` is what caps the result."""
+    session = create_session
+    stream = f"e2e_sqllimit_{uuid.uuid4().hex[:8]}"
+    try:
+        _seed(session, base_url, stream)
+        start, end = _window()
+
+        size = 5
+        resp = session.post(
+            f"{base_url}api/{ORG_ID}/_search?type=logs",
+            json={"query": {"sql": f'SELECT * FROM "{stream}"',
+                            "start_time": start, "end_time": end, "from": 0, "size": size}},
+        )
+        assert resp.status_code == 200, f"search failed: {resp.status_code} {resp.text[:400]}"
+        hits = resp.json().get("hits", [])
+        logger.info("no SQL LIMIT, size %s returned %s rows", size, len(hits))
+        assert len(hits) == size, f"size must cap the result when the SQL has no LIMIT, got {len(hits)}"
+    finally:
+        _drop(session, base_url, stream)
+
+def test_stream_name_containing_a_colon_is_queryable(create_session, base_url):
+    """A `:` in the stream name must be quoted into the SQL, not raise a ParserError."""
+    session = create_session
+    stream = f"e2e:colon:{uuid.uuid4().hex[:6]}"
+    try:
+        _seed(session, base_url, stream, rows=5)
+        start, end = _window()
+
+        resp = session.post(
+            f"{base_url}api/{ORG_ID}/_search?type=logs",
+            json={"query": {"sql": f'SELECT * FROM "{stream}"',
+                            "start_time": start, "end_time": end, "from": 0, "size": 10}},
+        )
+        logger.info("colon-named stream answered %s", resp.status_code)
+        assert resp.status_code == 200, \
+            f"a colon in the stream name must not break parsing: {resp.status_code} {resp.text[:400]}"
+        assert "ParserError" not in resp.text, f"ParserError surfaced: {resp.text[:300]}"
+        assert len(resp.json().get("hits", [])) == 5, "all seeded rows must come back"
+    finally:
+        _drop(session, base_url, stream)
diff --git a/tests/api-testing/tests/search/test_traces_latest_ordering.py b/tests/api-testing/tests/search/test_traces_latest_ordering.py
new file mode 100644
index 0000000000..114f2ff29e
--- /dev/null
+++ b/tests/api-testing/tests/search/test_traces_latest_ordering.py
@@ -0,0 +1,153 @@
+"""Traces list ordering and duration filtering  [regressions #10986, #11339, #11505].
+
+#10986 (PR #10993): the traces list did not come back ordered by `_timestamp`
+descending, so the newest trace was not first.
+
+#11339 (PR #11334): partition ordering ignored the requested sort direction --
+partitions stay as-is for `zo_sql_timestamp DESC` and must be reversed for ASC.
+Asserting both directions is the point: a build that always returns one fixed
+order satisfies a single-direction test.
+
+#11505 (PR #11514): filtering the list by duration produced blank rows, because
+the per-trace enrichment query was unbounded. The contract is that a filtered
+row still carries its trace identity, not that some number of rows comes back.
+"""
+
+import logging
+import os
+import time
+import uuid
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+ORG_ID = os.environ.get("TEST_ORG_ID", "default")
+TRACE_COUNT = 6
+STREAM = "default"
+
+
+def _span(trace_id, span_id, start_ns, dur_ns, name):
+    return {
+        "traceId": trace_id, "spanId": span_id, "parentSpanId": "", "flags": 1,
+        "name": name, "kind": 2,
+        "startTimeUnixNano": str(start_ns), "endTimeUnixNano": str(start_ns + dur_ns),
+        "attributes": [{"key": "http.route", "value": {"stringValue": "/probe"}}],
+        "status": {"code": 0},
+    }
+
+
+def _seed_traces(session, base_url):
+    """Six traces a second apart, with deliberately different durations."""
+    base_ns = int(time.time() * 1e9) - 300 * 10**9
+    ids = []
+    spans = []
+    for i in range(TRACE_COUNT):
+        tid = uuid.uuid4().hex
+        ids.append(tid)
+        spans.append(_span(tid, uuid.uuid4().hex[:16], base_ns + i * 10**9,
+                           (i + 1) * 5 * 10**6, f"op_{i}"))
+    payload = {"resourceSpans": [{
+        "resource": {"attributes": [{"key": "service.name",
+                                     "value": {"stringValue": "pytest_traces_order"}}]},
+        "scopeSpans": [{"scope": {"name": "pytest"}, "spans": spans}],
+    }]}
+    r = session.post(f"{base_url}api/{ORG_ID}/v1/traces", json=payload)
+    assert r.status_code == 200, f"trace ingest failed: {r.status_code} {r.text[:300]}"
+    return ids, base_ns
+
+
+def _seed_window(base_ns):
+    """Bound the query to the seeded span only.
+
+    The traces stream is shared and earlier runs leave data behind. With a wide
+    window `sort_order=asc` returns the oldest traces in the whole stream and
+    `size` truncates before this module's trace_id filter can apply, so the two
+    directions would compare different populations.
+    """
+    start_us = int(base_ns / 1000) - 1_000_000
+    end_us = int(base_ns / 1000) + (TRACE_COUNT + 2) * 1_000_000
+    return start_us, end_us
+
+
+def _latest(session, base_url, start_us, end_us, **extra):
+    params = {"start_time": start_us, "end_time": end_us, "from": 0, "size": 20}
+    params.update(extra)
+    r = session.get(f"{base_url}api/{ORG_ID}/{STREAM}/traces/latest", params=params)
+    assert r.status_code == 200, f"latest traces failed: {r.status_code} {r.text[:300]}"
+    return r.json()
+
+
+def _wait_for_traces(session, base_url, ids, start_us, end_us):
+    for _ in range(40):
+        body = _latest(session, base_url, start_us, end_us)
+        got = {h.get("trace_id") for h in body.get("hits", [])}
+        if len([i for i in ids if i in got]) >= TRACE_COUNT:
+            return body
+        time.sleep(2)
+    raise AssertionError("seeded traces never became visible in the latest-traces list")
+
+
+def test_latest_traces_are_ordered_newest_first(create_session, base_url):
+    session = create_session
+    ids, base_ns = _seed_traces(session, base_url)
+    start_us, end_us = _seed_window(base_ns)
+
+    body = _wait_for_traces(session, base_url, ids, start_us, end_us)
+
+    # Restrict to this run's traces; the stream is shared and older ones linger.
+    mine = [h for h in body["hits"] if h.get("trace_id") in set(ids)]
+    assert len(mine) == TRACE_COUNT, f"expected {TRACE_COUNT} seeded traces, got {len(mine)}"
+
+    stamps = [h["start_time"] for h in mine]
+    logger.info("latest-traces start_times: %s", stamps)
+    # Seeded one second apart, so a stable sort cannot mask a wrong direction.
+    assert stamps == sorted(stamps, reverse=True), \
+        f"traces must be newest-first, got {stamps}"
+
+
+def test_duration_filtered_rows_are_not_blank(create_session, base_url):
+    session = create_session
+    ids, base_ns = _seed_traces(session, base_url)
+    start_us, end_us = _seed_window(base_ns)
+    _wait_for_traces(session, base_url, ids, start_us, end_us)
+
+    # Durations are (i+1)*5ms, so this threshold must exclude the shortest traces
+    # rather than matching everything -- a filter that is a no-op proves nothing.
+    threshold_us = 15_000
+    body = _latest(session, base_url, start_us, end_us, filter=f"duration > {threshold_us}")
+    mine = [h for h in body.get("hits", []) if h.get("trace_id") in set(ids)]
+    logger.info("duration-filtered rows: %s of %s seeded", len(mine), TRACE_COUNT)
+
+    assert mine, "a duration filter matching some traces must still return rows"
+    assert len(mine) < TRACE_COUNT, \
+        f"the filter must exclude the shorter traces, got all {len(mine)}"
+
+    # The defect rendered rows with no trace identity, so populated rows are the contract.
+    for h in mine:
+        assert h.get("trace_id"), f"filtered row came back blank: {h}"
+        assert h.get("duration", 0) > threshold_us, \
+            f"row below the duration threshold leaked through: {h.get('duration')}"
+
+
+def test_sort_order_asc_reverses_the_list(create_session, base_url):
+    """`sort_order=asc` must reverse the order, not return the same fixed list."""
+    session = create_session
+    ids, base_ns = _seed_traces(session, base_url)
+    start_us, end_us = _seed_window(base_ns)
+    _wait_for_traces(session, base_url, ids, start_us, end_us)
+
+    seeded = set(ids)
+    desc = [h["start_time"] for h in
+            _latest(session, base_url, start_us, end_us, sort_order="desc")["hits"]
+            if h.get("trace_id") in seeded]
+    asc = [h["start_time"] for h in
+           _latest(session, base_url, start_us, end_us, sort_order="asc")["hits"]
+           if h.get("trace_id") in seeded]
+    logger.info("desc=%s asc=%s", desc, asc)
+
+    assert desc == sorted(desc, reverse=True), f"desc must be newest-first, got {desc}"
+    assert asc == sorted(asc), f"asc must be oldest-first, got {asc}"
+    # The defect ignored the requested direction, so the two must actually differ.
+    assert asc == list(reversed(desc)), \
+        f"asc must be the reverse of desc; asc={asc} desc={desc}"
+
diff --git a/tests/api-testing/tests/streams/test_schema_type_evolution.py b/tests/api-testing/tests/streams/test_schema_type_evolution.py
new file mode 100644
index 0000000000..92bd23f0cc
--- /dev/null
+++ b/tests/api-testing/tests/streams/test_schema_type_evolution.py
@@ -0,0 +1,75 @@
+"""Boolean to numeric schema evolution  [regression #10475].
+
+A field first seen as a boolean and later as a number was rejected, because
+`is_widening_conversion` did not permit boolean widening. PR #10478 expanded the
+allowed conversions so the schema evolves instead of failing the write.
+
+The second ingest is the contract. Asserting only that the first succeeds would
+pass on any build, so the test checks the later records are accepted AND that
+both generations are queryable afterwards -- a build that accepted the write but
+dropped the rows would otherwise look fixed.
+"""
+
+import logging
+import os
+import time
+import uuid
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+ORG_ID = os.environ.get("TEST_ORG_ID", "default")
+
+def _drop(session, base_url, stream, stream_type="logs"):
+    """Follows the suite convention: the test that seeds a stream drops it."""
+    session.delete(f"{base_url}api/{ORG_ID}/streams/{stream}?type={stream_type}")
+
+
+
+def _window():
+    now = int(time.time() * 1_000_000)
+    return now - 3_600_000_000, now + 60_000_000
+
+
+def _search(session, base_url, sql):
+    start, end = _window()
+    return session.post(
+        f"{base_url}api/{ORG_ID}/_search?type=logs",
+        json={"query": {"sql": sql, "start_time": start, "end_time": end,
+                        "from": 0, "size": 100}},
+    )
+
+
+def test_boolean_field_widens_to_numeric(create_session, base_url):
+    session = create_session
+    stream = f"e2e_evolve_{uuid.uuid4().hex[:8]}"
+    try:
+        now = int(time.time() * 1_000_000)
+
+        bool_rows = [{"_timestamp": now - 5_000_000 + i, "job": "pytest_evolve",
+                      "flag": True, "tag": "boolgen"} for i in range(3)]
+        r1 = session.post(f"{base_url}api/{ORG_ID}/{stream}/_json", json=bool_rows)
+        assert r1.status_code == 200, f"boolean ingest failed: {r1.status_code} {r1.text[:300]}"
+
+        num_rows = [{"_timestamp": now + i, "job": "pytest_evolve",
+                     "flag": 7, "tag": "numgen"} for i in range(3)]
+        r2 = session.post(f"{base_url}api/{ORG_ID}/{stream}/_json", json=num_rows)
+        assert r2.status_code == 200, f"numeric ingest failed: {r2.status_code} {r2.text[:300]}"
+        body = r2.json()
+        failed = sum(s.get("failed", 0) for s in body.get("status", []))
+        logger.info("numeric generation ingest status: %s", body.get("status"))
+        assert failed == 0, f"numeric records were rejected after a boolean field: {body}"
+
+        for _ in range(30):
+            resp = _search(session, base_url, f'SELECT tag, flag FROM "{stream}"')
+            if resp.status_code == 200 and len(resp.json().get("hits", [])) >= 6:
+                break
+            time.sleep(1)
+        assert resp.status_code == 200, f"search failed: {resp.status_code} {resp.text[:300]}"
+
+        tags = [h.get("tag") for h in resp.json()["hits"]]
+        logger.info("rows visible after evolution: %s", sorted(tags))
+        assert tags.count("boolgen") == 3, f"boolean-generation rows lost: {tags}"
+        assert tags.count("numgen") == 3, f"numeric-generation rows lost: {tags}"
+    finally:
+        _drop(session, base_url, stream)
diff --git a/tests/ui-testing/ci-matrix/ci_matrix_regression.json b/tests/ui-testing/ci-matrix/ci_matrix_regression.json
index a8de3a13da..c6be97f506 100644
--- a/tests/ui-testing/ci-matrix/ci_matrix_regression.json
+++ b/tests/ui-testing/ci-matrix/ci_matrix_regression.json
@@ -21,7 +21,8 @@
       "logs-multistream-matchall-pagination.spec.js",
       "logs-14303-streaming-render.spec.js",
       "logs-multistream-share-url.spec.js",
-      "logs-histogram-severity.spec.js"
+      "logs-histogram-severity.spec.js",
+      "logs-field-value-rendering.spec.js"
     ],
     "disabled": []
   },
@@ -102,7 +103,8 @@
     "browser": "chrome",
     "run_files": [
       "ui-regression.spec.js",
-      "landing-regression.spec.js"
+      "landing-regression.spec.js",
+      "internal-route-guards.spec.js"
     ],
     "disabled": []
   },
diff --git a/tests/ui-testing/pages/generalPages/routeGuardPage.js b/tests/ui-testing/pages/generalPages/routeGuardPage.js
new file mode 100644
index 0000000000..2d167b7d25
--- /dev/null
+++ b/tests/ui-testing/pages/generalPages/routeGuardPage.js
@@ -0,0 +1,45 @@
+import { expect } from '@playwright/test';
+
+/** Internal pages that must not be reachable in a shipped build. */
+export class RouteGuardPage {
+    constructor(page) {
+        this.page = page;
+        // The removed demo rendered its own heading; the URL alone is not the tell.
+        this.emptyStateDemoHeading = page.getByText('Empty states', { exact: false });
+        this.notFoundHeading = page.getByText('Page not found', { exact: false });
+    }
+
+    /** Navigate to an app route under /web, carrying the org identifier. */
+    async gotoAppRoute(routePath, orgIdentifier) {
+        const org = orgIdentifier || process.env.ORGNAME || 'default';
+        const base = process.env.ZO_BASE_URL.replace(/\/$/, '');
+        await this.page.goto(`${base}/web/${routePath}?org_identifier=${org}`);
+        await this.page.waitForLoadState('networkidle', { timeout: 15000 }).catch(() => {});
+    }
+
+    /** build_type as the server reports it, so an edition-specific guard can be skipped elsewhere. */
+    async getBuildType() {
+        const base = process.env.ZO_BASE_URL.replace(/\/$/, '');
+        return await this.page.evaluate(async (url) => {
+            try {
+                const r = await fetch(`${url}/config`, { credentials: 'include' });
+                return (await r.json()).build_type || 'unknown';
+            } catch {
+                return 'unknown';
+            }
+        }, base);
+    }
+
+    async expectEmptyStateDemoNotRendered() {
+        await expect(this.emptyStateDemoHeading).toHaveCount(0, { timeout: 10000 });
+    }
+
+    /** A removed route falls through to the 404 view rather than changing the URL. */
+    async expectNotFoundRendered() {
+        await expect(this.notFoundHeading.first()).toBeVisible({ timeout: 15000 });
+    }
+
+    currentUrl() {
+        return this.page.url();
+    }
+}
diff --git a/tests/ui-testing/pages/logsPages/logsPage.js b/tests/ui-testing/pages/logsPages/logsPage.js
index 72143d5730..196ef3cc0a 100644
--- a/tests/ui-testing/pages/logsPages/logsPage.js
+++ b/tests/ui-testing/pages/logsPages/logsPage.js
@@ -511,6 +511,8 @@ export class LogsPage {
         // The "=" icon is revealed on hover over its sidebar field row.
         this.fieldListItemPrefix = 'logs-field-list-item-';
         this.logDetailRowPrefix = 'log-detail-row-';
+        this.fieldValuesList = '[data-test="field-values-panel-values-list"]';
+        this.fieldValueLabelPrefix = (field) => `[data-test^="logs-search-subfield-add-${field}-"]`;
         this.fieldEqualsButton = (field) => `[data-test="log-search-index-list-filter-${field}-field-btn"]`;
         this.logDetailRow = (field) => `[data-test="log-detail-row-${field}"]`;
         this.logDetailFieldMenuTrigger = '[data-test="log-details-include-exclude-field-btn"]';
@@ -2045,6 +2047,28 @@ export class LogsPage {
             .toContainText(`1 to ${size}`, { timeout: 30000 });
     }
 
+    /**
+     * Wait until the banner reports the expected total.
+     *
+     * Ingest acknowledges before every row is searchable, so a query fired
+     * immediately after seeding sees an arbitrary prefix of the rows -- the same
+     * seed produced 12, then 6, then 3 across consecutive runs. Re-running the
+     * query while polling lets the index catch up.
+     */
+    async waitForResultTotal(expected, attempts = 12) {
+        for (let i = 0; i < attempts; i++) {
+            const title = await this.getResultTitleText();
+            const total = Number((title.match(/out of\s+([\d.]+)/) || [])[1]);
+            if (total === expected) return title;
+            await this.page.waitForTimeout(2000);
+            await this.clickRefreshButton();
+            await this.expectResultsGridSettledWithRows();
+        }
+        throw new Error(
+            `banner never reported ${expected} events; last was "${(await this.getResultTitleText()).replace(/\n/g, ' | ')}"`,
+        );
+    }
+
     /** The "1 to N out of M events" banner, used to compare the claim against rendered rows. */
     async getResultTitleText() {
         return (await this.page.locator(this.paginationRowCountTitle).first().innerText()).trim();
@@ -2059,6 +2083,25 @@ export class LogsPage {
         return await this.page.locator(this.highlightedMatch).count();
     }
 
+    /**
+     * Rendered text and wrapping of a sidebar field's first value row.
+     *
+     * The wrapping lives on an inner value div carrying `title`, not on the
+     * labelled element itself, so reading the label's own computed style
+     * reports the wrapper's `normal` and misses the fix entirely.
+     */
+    async getFirstFieldValueRendering(field) {
+        const label = this.page.locator(this.fieldValueLabelPrefix(field)).first();
+        await label.waitFor({ state: 'visible', timeout: 20000 });
+        const value = label.locator('div[title]').first();
+        await value.waitFor({ state: 'visible', timeout: 10000 });
+        return await value.evaluate((el) => ({
+            whiteSpace: getComputedStyle(el).whiteSpace,
+            text: el.innerText,
+            newlineCount: (el.innerText.match(/\n/g) || []).length,
+        }));
+    }
+
     /** Sidebar fields that offer an "=" filter, in render order; the set varies by dataset. */
     async getFilterableSidebarFields(count = 2) {
         const fields = await this.page
diff --git a/tests/ui-testing/pages/page-manager.js b/tests/ui-testing/pages/page-manager.js
index 5c62ab9a12..77db79960c 100644
--- a/tests/ui-testing/pages/page-manager.js
+++ b/tests/ui-testing/pages/page-manager.js
@@ -51,6 +51,7 @@ import { IngestionPage } from "./generalPages/ingestionPage.js";
 import { CloudLoginPage } from "./cloudPages/cloudLoginPage.js";
 import { isCloudEnvironment } from "./cloudPages/cloud-env.js";
 import { IngestionConfigPage } from "./generalPages/ingestionConfigPage.js";
+import { RouteGuardPage } from "./generalPages/routeGuardPage.js";
 import { SplunkHecPage } from "./generalPages/splunkHecPage.js";
 
 // ===== GENERAL TESTS ADDITIONAL PAGE OBJECTS =====
@@ -201,6 +202,7 @@ class PageManager {
     this.pipelinesFormValidation = new PipelinesFormValidationPage(page);
     this.loginPage = isCloudEnvironment() ? new CloudLoginPage(page) : new LoginPage(page);
     this.ingestionPage = new IngestionPage(page);
+    this.routeGuardPage = new RouteGuardPage(page);
     this.ingestionConfigPage = new IngestionConfigPage(page);
     this.splunkHecPage = new SplunkHecPage(page);
 
diff --git a/tests/ui-testing/playwright-tests/RegressionSet/General/internal-route-guards.spec.js b/tests/ui-testing/playwright-tests/RegressionSet/General/internal-route-guards.spec.js
new file mode 100644
index 0000000000..95ad1d10dd
--- /dev/null
+++ b/tests/ui-testing/playwright-tests/RegressionSet/General/internal-route-guards.spec.js
@@ -0,0 +1,25 @@
+const { test, navigateToBase } = require('../../utils/enhanced-baseFixtures.js');
+const testLogger = require('../../utils/test-logger.js');
+const PageManager = require('../../../pages/page-manager.js');
+
+test.describe("Internal Route Guards Regression", () => {
+  test.describe.configure({ mode: 'parallel' });
+  let pm;
+
+  test.beforeEach(async ({ page }, testInfo) => {
+    testLogger.testStart(testInfo.title, testInfo.file);
+    await navigateToBase(page);
+    pm = new PageManager(page);
+  });
+
+  test("the internal empty-state demo page is not shipped @bug-14375 @P2 @regression @routeGuards", async () => {
+    await pm.routeGuardPage.gotoAppRoute('empty-state-demo');
+
+    // The route and its component were removed outright. The router keeps the
+    // requested URL and falls through to the 404 view, so the demo's own heading
+    // and the 404 view are the contract -- not the address bar.
+    await pm.routeGuardPage.expectEmptyStateDemoNotRendered();
+    await pm.routeGuardPage.expectNotFoundRendered();
+    testLogger.info(`empty-state-demo fell through to the 404 view`);
+  });
+});
diff --git a/tests/ui-testing/playwright-tests/RegressionSet/Logs/logs-bugs.spec.js b/tests/ui-testing/playwright-tests/RegressionSet/Logs/logs-bugs.spec.js
index ba738550fb..49092710eb 100644
--- a/tests/ui-testing/playwright-tests/RegressionSet/Logs/logs-bugs.spec.js
+++ b/tests/ui-testing/playwright-tests/RegressionSet/Logs/logs-bugs.spec.js
@@ -1970,6 +1970,45 @@ test.describe("Logs Regression Bug Fixes", () => {
   // Bug #14228: Builder saved view does not apply when already on the Build tab
   // https://github.com/openobserve/openobserve/issues/14228
   // ==========================================================================
+  // ==========================================================================
+  // Bug #13990: Builder saved view restores the wrong stream
+  // https://github.com/openobserve/openobserve/issues/13990
+  // ==========================================================================
+  test('Builder saved view restores the stream it was saved with @bug-13990 @P1 @regression @logsRegression @savedViews @queryBuilder', {
+    tag: ['@bug-13990', '@P1', '@regression', '@logsRegression', '@savedViews', '@queryBuilder']
+  }, async ({ page }) => {
+    testLogger.info('Test: Builder saved view restores its own stream (Bug #13990)');
+
+    const savedViewName = `streamslog_build_13990_${Date.now()}`;
+    const otherStream = 'e2e_13990_other_' + Math.random().toString(36).slice(2, 7);
+
+    // A second stream to move away to; without it the assertion would hold
+    // trivially, since the page would never have left the saved stream.
+    const orgId = getOrgIdentifier() || 'default';
+    await sendRequest(page, getIngestionUrl(orgId, otherStream), [{
+      level: 'info', job: 'test_13990', log: 'second stream for saved view restore', e2e: '1',
+    }], getHeaders());
+    fieldCacheStreamsToCleanup.push(otherStream);
+
+    await page.goto(`${logData.logsUrl}?org_identifier=${orgId}`);
+    await page.waitForLoadState('networkidle', { timeout: 15000 }).catch(() => {});
+    await pm.logsPage.selectStream('e2e_automate');
+    await pm.logsPage.expectLogsSearchIndexListContainsText('e2e_automate');
+
+    await createBuilderSavedView(pm, page, savedViewName, 'line');
+
+    await pm.logsPage.selectStream(otherStream);
+    await pm.logsPage.expectLogsSearchIndexListContainsText(otherStream);
+    testLogger.info(`Moved to ${otherStream} before applying the saved view`);
+
+    await applySavedViewByName(pm, page, savedViewName);
+
+    // The defect restored an unrelated stream, so the stream is the contract here
+    // rather than the chart type that #14228 covers.
+    await pm.logsPage.expectLogsSearchIndexListContainsText('e2e_automate');
+    testLogger.info('Saved view restored its original stream');
+  });
+
   test("should apply a builder saved view while already on the build tab", {
     tag: ['@bug-14228', '@P1', '@regression', '@logsRegression', '@savedViews', '@queryBuilder']
   }, async ({ page }) => {
diff --git a/tests/ui-testing/playwright-tests/RegressionSet/Logs/logs-field-value-rendering.spec.js b/tests/ui-testing/playwright-tests/RegressionSet/Logs/logs-field-value-rendering.spec.js
new file mode 100644
index 0000000000..24e3bda5a5
--- /dev/null
+++ b/tests/ui-testing/playwright-tests/RegressionSet/Logs/logs-field-value-rendering.spec.js
@@ -0,0 +1,45 @@
+const { test, expect, navigateToBase } = require('../../utils/enhanced-baseFixtures.js');
+const testLogger = require('../../utils/test-logger.js');
+const PageManager = require('../../../pages/page-manager.js');
+const { sendRequest, getHeaders, getIngestionUrl } = require('../../utils/data-ingestion.js');
+const { getOrgIdentifier } = require('../../utils/cloud-auth.js');
+
+const MULTILINE_FIELD = 'stacktrace';
+const MULTILINE_VALUE = 'Traceback (most recent call last):\n  File "a.py", line 1\n    raise ValueError';
+
+test.describe("Logs Field Value Rendering Regression", () => {
+  test.describe.configure({ mode: 'parallel' });
+  let pm;
+  let stream;
+
+  test.beforeEach(async ({ page }, testInfo) => {
+    testLogger.testStart(testInfo.title, testInfo.file);
+    await navigateToBase(page);
+    pm = new PageManager(page);
+
+    // A field whose value genuinely contains newlines; the shared fixture has none.
+    stream = 'e2e_13354_' + Math.random().toString(36).slice(2, 7);
+    const orgId = getOrgIdentifier() || 'default';
+    await sendRequest(page, getIngestionUrl(orgId, stream), [{
+      level: 'error', job: 'test_13354', [MULTILINE_FIELD]: MULTILINE_VALUE,
+    }], getHeaders());
+
+    await pm.logsPage.selectStream(stream);
+    await pm.logsPage.clickRefreshButton();
+  });
+
+  test("multiline field values keep their newlines in the values panel @bug-13354 @P2 @regression @logsRegression", async () => {
+    await pm.logsPage.clickFieldExpandButton(MULTILINE_FIELD);
+
+    const rendering = await pm.logsPage.getFirstFieldValueRendering(MULTILINE_FIELD);
+    testLogger.info(`white-space=${rendering.whiteSpace} newlines=${rendering.newlineCount}`);
+
+    // The label collapsed newlines because it did not carry pre-wrap, so the
+    // wrapping mode is the contract -- the text alone can look right while the
+    // browser still renders it on one line.
+    expect(rendering.whiteSpace, 'the value label must preserve newlines')
+      .toMatch(/pre-wrap|pre-line|pre/);
+    expect(rendering.newlineCount, 'the rendered value must still contain its newlines')
+      .toBeGreaterThan(0);
+  });
+});
diff --git a/tests/ui-testing/playwright-tests/RegressionSet/Logs/logs-result-display.spec.js b/tests/ui-testing/playwright-tests/RegressionSet/Logs/logs-result-display.spec.js
index 191ff86ad3..4f2f4b76da 100644
--- a/tests/ui-testing/playwright-tests/RegressionSet/Logs/logs-result-display.spec.js
+++ b/tests/ui-testing/playwright-tests/RegressionSet/Logs/logs-result-display.spec.js
@@ -1,10 +1,13 @@
 const { test, expect, navigateToBase } = require('../../utils/enhanced-baseFixtures.js');
 const testLogger = require('../../utils/test-logger.js');
 const PageManager = require('../../../pages/page-manager.js');
-const { ingestTestData } = require('../../utils/data-ingestion.js');
+const { ingestTestData, sendRequest, getHeaders, getIngestionUrl } = require('../../utils/data-ingestion.js');
+const { getOrgIdentifier } = require('../../utils/cloud-auth.js');
+const logData = require('../../../fixtures/log.json');
 
 const STREAM = 'e2e_automate';
 const PAGE_SIZE = 100;
+const DENSE_ROWS = 12;
 
 test.describe("Logs Result Display Regression", () => {
   test.describe.configure({ mode: 'parallel' });
@@ -52,4 +55,49 @@ test.describe("Logs Result Display Regression", () => {
     expect(rendered, 'rendered rows must match the page size the banner claims')
       .toBe(PAGE_SIZE);
   });
+
+  test("the event total matches the range, not the histogram bucket @bug-13896 @P1 @regression @logsRegression", async ({ page }) => {
+    // The counter was summed from histogram buckets, so it only diverges when the
+    // selected range is NARROWER than the histogram interval -- hence one dense
+    // second of its own stream rather than the shared fixture.
+    const denseStream = 'e2e_13896_' + Math.random().toString(36).slice(2, 7);
+    const orgId = getOrgIdentifier() || 'default';
+    const secondStartMs = Date.now() - 5 * 60 * 1000;
+
+    const rows = Array.from({ length: DENSE_ROWS }, (_, i) => ({
+      _timestamp: (secondStartMs + i * 20) * 1000,
+      level: 'info',
+      job: 'test_13896',
+      log: `dense row ${i}`,
+    }));
+    await sendRequest(page, getIngestionUrl(orgId, denseStream), rows, getHeaders());
+
+    await pm.logsPage.waitForStreamAvailable(denseStream);
+    await page.goto(
+      `${logData.logsUrl}?org_identifier=${orgId}&stream=${denseStream}&stream_type=logs` +
+      `&from=${secondStartMs}&to=${secondStartMs + 1000}`,
+    );
+    await pm.logsPage.clickRefreshButton();
+    await pm.logsPage.expectResultsGridSettledWithRows();
+
+    // Ingest acks before every row is searchable, so settle on the full count
+    // first -- otherwise the banner reports whatever prefix happens to be
+    // indexed and the comparison below is against the wrong number.
+    const title = await pm.logsPage.waitForResultTotal(DENSE_ROWS);
+    const rendered = await pm.logsPage.getLogRowCount();
+    testLogger.info(`Banner: ${title.replace(/\n/g, ' | ')} | rendered rows: ${rendered}`);
+
+    const claimedTotal = Number((title.match(/out of\s+([\d.]+)/) || [])[1]);
+    expect(Number.isFinite(claimedTotal), `could not read a total from: ${title}`).toBeTruthy();
+
+    // The defect overstated the total by an order of magnitude, counting the whole
+    // histogram bucket while the log query correctly filtered to the range.
+    // Asserted as EQUAL, not <=: a `<=` here also passes when the window is wrong
+    // and only part of the seeded second is matched, which is a weaker test than
+    // it looks. from/to are milliseconds (see logsUtils), while ingest takes
+    // microseconds -- both denote the same instant.
+    expect(claimedTotal, 'the banner total must equal the rows seeded in the range')
+      .toBe(DENSE_ROWS);
+    expect(rendered, 'all rows in the one-second window must render').toBe(claimedTotal);
+  });
 });
diff --git a/tests/ui-testing/playwright-tests/cleanup.spec.js b/tests/ui-testing/playwright-tests/cleanup.spec.js
index e04abde08a..c02f52c449 100644
--- a/tests/ui-testing/playwright-tests/cleanup.spec.js
+++ b/tests/ui-testing/playwright-tests/cleanup.spec.js
@@ -90,6 +90,7 @@ test.describe("Pre-Test Cleanup", () => {
     // them. Harmless on CI's throwaway server, but they pile up on a shared dev env.
     await pm.apiCleanup.cleanupStreams([/^e2e_14238_/], [], { streamType: 'metrics' });
 
+
     // Clean up all reports owned by automation user
     await pm.apiCleanup.cleanupReports();
 
@@ -297,6 +298,9 @@ test.describe("Pre-Test Cleanup", () => {
     await pm.apiCleanup.cleanupStreams(
       [
         /^e2e_slo_14269_/,             // slo-error-messages.spec.js (#14269) seeded stream
+        /^e2e_13354_/,                 // logs-field-value-rendering.spec.js (#13354)
+        /^e2e_13896_/,                 // logs-result-display.spec.js (#13896)
+        /^e2e_13990_other_/,           // logs-bugs.spec.js (#13990)
         /^e2e_sev_str_/,               // logs-histogram-severity.spec.js (#11353)
         /^e2e_sev_num_/,               // logs-histogram-severity.spec.js (#11441)
         /^e2e_ms7332_/,                // logs-multistream-share-url.spec.js (#7332)