test: close the metric drilldown coverage gaps

src/backend/services/analytics/Cargo.toml

@@ -83,3 +83,6 @@ tower = { version = "0.5", features = ["util"] }
 # `clickhouse::test::Mock` for exercising ClickHouse read paths (the
 # contract-version sweep) against an in-process server in unit tests.
 clickhouse = { workspace = true, features = ["test-util"] }
+# Paused-time `#[tokio::test]` for the semaphore acquire-timeout refusals —
+# the 2 s acquire window elapses instantly instead of being slept through.
+tokio = { workspace = true, features = ["test-util"] }

src/backend/services/analytics/src/api/metric_drilldown.rs

@@ -152,6 +152,20 @@ async fn acquire_export_permit() -> Result<tokio::sync::SemaphorePermit<'static>
         .map_err(|_| export_busy())
 }
 
+async fn acquire_query_permit() -> Result<tokio::sync::SemaphorePermit<'static>, CanonicalError> {
+    tokio::time::timeout(QUERY_ACQUIRE_TIMEOUT, QUERY_SEMAPHORE.acquire())
+        .await
+        .map_err(|_| {
+            tracing::warn!(
+                capacity = MAX_CONCURRENT_QUERIES,
+                available = QUERY_SEMAPHORE.available_permits(),
+                "metric drilldown query capacity exhausted"
+            );
+            query_busy()
+        })?
+        .map_err(|_| query_busy())
+}
+
 async fn collect_export_rows(
     state: &Arc<AppState>,
     validated: &ValidatedMetricDrilldown,
@@ -166,12 +180,17 @@ async fn collect_export_rows(
         .map_err(|_| export_limit("Export exceeded the execution time limit."))??;
     verify_evidence_snapshot(&state.ch, &validated.plan.relation, &validated.snapshot_id).await?;
 
-    if rows.len() > MAX_EXPORT_ROWS {
+    enforce_export_row_limit(rows.len())?;
+    Ok(rows)
+}
+
+fn enforce_export_row_limit(rows: usize) -> Result<(), CanonicalError> {
+    if rows > MAX_EXPORT_ROWS {
         return Err(export_limit(format!(
             "Export exceeds the {MAX_EXPORT_ROWS} row limit."
         )));
     }
-    Ok(rows)
+    Ok(())
 }
 
 async fn serialize_export(
@@ -230,10 +249,7 @@ async fn fetch_rows(
 ) -> Result<Vec<EvidenceQueryRow>, CanonicalError> {
     // INVARIANT: the permit is held across the awaited ClickHouse execution and
     // byte collection below — the hold is the MAX_CONCURRENT_QUERIES cap.
-    let _permit = tokio::time::timeout(QUERY_ACQUIRE_TIMEOUT, QUERY_SEMAPHORE.acquire())
-        .await
-        .map_err(|_| query_busy())?
-        .map_err(|_| query_busy())?;
+    let _permit = acquire_query_permit().await?;
     let (sql, params) = compile_query(req)?;
     let base = state
         .ch
@@ -333,4 +349,54 @@ mod tests {
             axum::http::StatusCode::INTERNAL_SERVER_ERROR
         );
     }
+
+    #[test]
+    fn the_export_row_limit_refuses_only_past_the_cap() {
+        assert!(enforce_export_row_limit(MAX_EXPORT_ROWS).is_ok());
+        let refused = enforce_export_row_limit(MAX_EXPORT_ROWS + 1)
+            .err()
+            .map(|error| error.status_code());
+        assert_eq!(
+            refused,
+            Some(axum::http::StatusCode::TOO_MANY_REQUESTS.as_u16())
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn export_permits_refuse_past_the_concurrency_cap() {
+        let mut held = Vec::new();
+        for _ in 0..MAX_CONCURRENT_EXPORTS {
+            held.extend(acquire_export_permit().await.ok());
+        }
+        assert_eq!(held.len(), MAX_CONCURRENT_EXPORTS);
+        let refused = acquire_export_permit()
+            .await
+            .err()
+            .map(|error| error.status_code());
+        assert_eq!(
+            refused,
+            Some(axum::http::StatusCode::TOO_MANY_REQUESTS.as_u16())
+        );
+        drop(held);
+        assert!(acquire_export_permit().await.is_ok());
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn query_permits_refuse_past_the_concurrency_cap() {
+        let mut held = Vec::new();
+        for _ in 0..MAX_CONCURRENT_QUERIES {
+            held.extend(acquire_query_permit().await.ok());
+        }
+        assert_eq!(held.len(), MAX_CONCURRENT_QUERIES);
+        let refused = acquire_query_permit()
+            .await
+            .err()
+            .map(|error| error.status_code());
+        assert_eq!(
+            refused,
+            Some(axum::http::StatusCode::TOO_MANY_REQUESTS.as_u16())
+        );
+        drop(held);
+        assert!(acquire_query_permit().await.is_ok());
+    }
 }

src/ingestion/tools/seed/PROFILE.md

@@ -17,7 +17,7 @@ builder that writes `manifest.json`, so the two cannot disagree.
 | realm | `insight` |
 | anchor_date | `2026-06-30` |
 | data_window | `2026-05-02..2026-06-30` |
-| seed_revision | `07365a1509ac7db6` |
+| seed_revision | `ef801f8b54393aa7` |
 | manifest_version | 1 |
 
 `anchor_date` is the last day carrying seeded activity. It is resolved

src/ingestion/tools/seed/insight_seed/generators/git.py

@@ -12,7 +12,7 @@
 from collections.abc import Sequence
 from typing import TYPE_CHECKING
 
-from ..profiles import TEAM_PROFILES, Person
+from ..profiles import DEV_LEAD_UUID, TEAM_PROFILES, Person
 from .base import (
     bulk_insert,
     clamp,
@@ -52,6 +52,24 @@
 # change_type_label multiIf; anything else renders as the raw value.
 CHANGE_TYPES = ("added", "modified", "renamed", "deleted")
 
+# Deliberately hostile — and clearly synthetic — commit messages for the
+# drilldown-export escaping scenario (#1603 scenario 11). The gold evidence
+# model surfaces a commit's message as the drilldown "Title" cell, so these
+# cover every value class a spreadsheet consumer can mishandle: the four
+# formula-prefix bytes ('=', '+', '-', '@') and a value with an embedded tab
+# and an embedded newline (which must stay inside one CSV cell). Only the
+# FIRST few commits the dev lead generates carry one; every other commit
+# keeps the column default, so no row count, metric value, or other
+# person's evidence changes.
+HOSTILE_COMMIT_MESSAGES = (
+    "=SUM(A1:A9) synthetic title",
+    "+A1 synthetic title",
+    "-2+3 synthetic title",
+    "@macro synthetic title",
+    "tab\tinside synthetic title",
+    "newline\ninside synthetic title",
+)
+
 
 def _eligible(roster: Sequence[Person]) -> list[Person]:
     """Persons whose team profile has any git weight."""
@@ -96,10 +114,18 @@ def seed_class_git_commits(
         "lines_added",
         "lines_removed",
         "data_source",
+        # Appended after the long-standing columns: the link-parity tests read
+        # this table positionally, so a mid-tuple insertion moves their fields.
+        "message",
         "_version",
     ]
     rows: list[tuple[object, ...]] = []
     version = 1
+    # Dealt to the dev lead's first commits in generation order (a merge
+    # commit is skipped — gold's evidence model filters those out, so a
+    # message on one would never reach the drilldown). Deterministic across
+    # re-seeds because the commit stream itself is.
+    hostile_messages = list(HOSTILE_COMMIT_MESSAGES)
     for p in _eligible(roster):
         persona = persona_multiplier(p.uuid)
         weight = TEAM_PROFILES[p.team or ""].weights["github"]
@@ -113,6 +139,9 @@ def seed_class_git_commits(
                 # LOC per commit capped at ≤200 by construction.
                 added = float(rng.randint(2, 180))
                 removed = float(rng.randint(0, 80))
+                message = ""
+                if p.uuid == DEV_LEAD_UUID and not is_merge and hostile_messages:
+                    message = hostile_messages.pop(0)
                 rows.append(
                     (
                         tenant_uuid,
@@ -128,6 +157,7 @@ def seed_class_git_commits(
                         added,
                         removed,
                         "insight_github",
+                        message,
                         version,
                     )
                 )

tests/pyproject.toml

@@ -82,6 +82,7 @@ markers = [
     "requires_ingestion: needs a stand whose manifest declares the 'ingestion' capability; skipped with a reason when it does not",
     "requires_catalogue(*parts): needs rows the analytics seed writes ('table_columns', 'definition_override'); skipped with a reason on a stand seeded without that step",
     "requires_service_principal: needs a stand whose authenticator token listener this runner can reach, so a service principal can be obtained; skipped with a reason when it cannot",
+    "rebuild_lane: the serialized rebuild lane — the test triggers a scoped dbt rebuild of one gold evidence relation through the stand's own seed image, so it needs docker beside the local compose stand and a run nothing else shares. OPT-IN: skipped unless --rebuild-lane is passed (tests/stand/conftest.py)",
     # Quality vectors — every stand api/ui test carries exactly one: a module-level
     # pytestmark where the whole module shares a vector, per-test markers throughout
     # a mixed module (never both — markers are additive, so a default plus an