Merge pull request #144 from seqeralabs/cursor/issue-72-api-preflight-bc67

Improve Seqera API preflight errors

Author: edmundmiller

.seqera/context/ERRORS.md

@@ -22,37 +22,45 @@ Both runs `1cF65l2PDvxNd5` (maniac_celsius) and `5VGC0gjEmghOaz` (clever_kalman)
 
 **Fix:** Verify the org/workspace names in the input CSV match the Platform exactly (case-sensitive).
 
-### 3. API Rate Limiting / Transient Failures
+### 3. API Endpoint / Token Preflight Failure
+
+**When:** The API endpoint is wrong, unreachable, or the access token is invalid.
+
+**Error:** `RuntimeException: Seqera Platform API preflight failed at '<endpoint>/service-info' ...` or `RuntimeException: Authentication failed at '<endpoint>/user-info' ...`
+
+**Fix:** Verify `--seqera_api_endpoint` (or the samplesheet `platform` column), network/truststore settings, and the token stored in `TOWER_ACCESS_TOKEN` (or the per-row `token_env` override).
+
+### 4. API Rate Limiting / Transient Failures
 
 **When:** Fetching data from busy Platform instances with many concurrent runs.
 
 **Mitigation:** Built-in retry with exponential backoff (3 attempts, 1s/2s/4s delays). After 3 failures, the pipeline aborts with the original error.
 
-### 4. Wave Container Build Failures
+### 5. Wave Container Build Failures
 
 **When:** Using `wave` profile with `spack` strategy.
 
 **Error:** Wave freeze builds fail with spack.
 
 **Fix:** The wave profile explicitly uses `strategy = ['conda', 'container', 'dockerfile']` — no spack. Don't add spack to the strategy list.
 
-### 5. CGROUPv2 Docker Failures (Cloud VM / Firecracker)
+### 6. CGROUPv2 Docker Failures (Cloud VM / Firecracker)
 
 **When:** Running with Docker in Cloud VMs where cgroup resource delegation is restricted.
 
 **Error:** `cannot enter cgroupv2 ... with domain controllers`
 
 **Fix:** Apply the runc wrapper documented in AGENTS.md that strips `linux.resources` from the OCI spec.
 
-### 6. Empty Benchmark Report
+### 7. Empty Benchmark Report
 
 **When:** API runs are provided but `--generate_benchmark_report` is not set.
 
 **Warning:** `Found N API run(s) but --generate_benchmark_report is not enabled. API runs will not produce any output.`
 
 **Fix:** Add `--generate_benchmark_report` to the run command.
 
-### 7. Scheduling Overhead (Observed Pattern)
+### 8. Scheduling Overhead (Observed Pattern)
 
 **Not an error per se**, but both current runs show 4–8 minute gaps between task submit and task start times. This is expected with AWS Batch spot instances — EC2 instances must be provisioned and containers pulled before execution begins. Not actionable unless overhead exceeds ~15 minutes consistently.
 

.seqera/context/PIPELINE.md

@@ -89,7 +89,7 @@ results/
 
 ## Data Flow Details
 
-1. **API path:** `SeqeraApi.fetchRunData()` runs in Groovy process memory (not a Nextflow task). It resolves workspace name → ID, then fetches `/workflow/{id}`, `/workflow/{id}/metrics`, `/workflow/{id}/tasks` (paginated), and `/workflow/{id}/progress`. Results are written to temp JSON files.
+1. **API path:** `SeqeraApi.fetchRunData()` runs in Groovy process memory (not a Nextflow task). It first validates the configured Platform endpoint with `/service-info` and the bearer token with `/user-info`, then resolves workspace name → ID and fetches `/workflow/{id}`, `/workflow/{id}/metrics`, `/workflow/{id}/tasks` (paginated), and `/workflow/{id}/progress`. Results are written to temp JSON files.
 2. **External path:** EXTRACT_TARBALL unpacks `.tar.gz` into a directory of JSON files. Directories are used directly.
 3. **All JSON files** are collected into a single temp directory and passed to the 3-stage Python pipeline: normalize → aggregate → render.
 4. The Python stages are separate Nextflow processes sharing one Wave container image (`python_duckdb_jinja2_typer_pruned`).

CHANGELOG.md

@@ -15,6 +15,7 @@ Thank you to everyone else that has contributed by reporting bugs, enhancements
 
 ### Enhancements & fixes
 
+- Improve Seqera API preflight errors for invalid API endpoints and access tokens
 - [PR #88](https://github.com/seqeralabs/nf-aggregate/pull/88) - Update tw cli container version to 0.11.2 and allow .nextflow.log to be missing from tw call
 - [PR #89](https://github.com/seqeralabs/nf-aggregate/pull/89) - Enable usage of external run dumps with nf-aggregate & update devcontainer specifications
 - [PR #91](https://github.com/seqeralabs/nf-aggregate/pull/91) - Update benchmark report image to include a fix causing large memory footprint for reshaping large AWS cost report files

README.md

@@ -15,6 +15,8 @@
 
 The pipeline fetches run data from the Seqera Platform API and generates benchmark reports comparing pipeline runs.
 
+Before the pipeline resolves workspaces or fetches run details, it performs API preflight checks against `/service-info` and `/user-info`. This helps surface misconfigured API endpoints, missing truststore configuration, and invalid access tokens with explicit error messages.
+
 ## Prerequisites
 
 - [Nextflow](https://www.nextflow.io/docs/latest/getstarted.html#installation) >=25.10.0

bin/benchmark_report_fetch.py

@@ -4,6 +4,8 @@
 from __future__ import annotations
 
 import json
+from functools import lru_cache
+from urllib.error import HTTPError, URLError
 from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
 from urllib.request import Request, urlopen
 
@@ -21,6 +23,62 @@ def _api_get(url: str, headers: dict[str, str], params: dict[str, str] | None =
         return json.loads(resp.read())
 
 
+def _format_http_error(exc: HTTPError) -> str:
+    return f"API request failed: {exc.url} → HTTP {exc.code}"
+
+
+@lru_cache(maxsize=None)
+def _validate_api_access_cached(api_endpoint: str, authorization: str, token_env_var: str) -> None:
+    headers = {"Authorization": authorization}
+
+    try:
+        _api_get(f"{api_endpoint}/service-info", headers=headers)
+    except HTTPError as exc:
+        raise RuntimeError(
+            "Seqera Platform API preflight failed at "
+            f"'{api_endpoint}/service-info'. Check the API endpoint URL from "
+            "--api-endpoint or the input samplesheet platform column. "
+            f"Original error: {_format_http_error(exc)}"
+        ) from exc
+    except URLError as exc:
+        raise RuntimeError(
+            "Could not reach Seqera Platform API preflight endpoint "
+            f"'{api_endpoint}/service-info'. Check the API endpoint URL, network "
+            f"access, and TLS configuration. Original error: {exc.reason}"
+        ) from exc
+
+    try:
+        _api_get(f"{api_endpoint}/user-info", headers=headers)
+    except HTTPError as exc:
+        if exc.code in {401, 403}:
+            raise RuntimeError(
+                "Authentication failed at "
+                f"'{api_endpoint}/user-info'. Check the access token stored in "
+                f"'{token_env_var}'. Original error: {_format_http_error(exc)}"
+            ) from exc
+        raise RuntimeError(
+            "Seqera Platform API auth preflight failed at "
+            f"'{api_endpoint}/user-info'. Check the API endpoint URL and the "
+            f"access token stored in '{token_env_var}'. Original error: "
+            f"{_format_http_error(exc)}"
+        ) from exc
+    except URLError as exc:
+        raise RuntimeError(
+            "Could not complete Seqera Platform auth preflight at "
+            f"'{api_endpoint}/user-info'. Check network access and the access "
+            f"token stored in '{token_env_var}'. Original error: {exc.reason}"
+        ) from exc
+
+
+def validate_api_access(
+    api_endpoint: str,
+    headers: dict[str, str],
+    token_env_var: str = "TOWER_ACCESS_TOKEN",
+) -> None:
+    authorization = headers.get("Authorization", "")
+    _validate_api_access_cached(api_endpoint, authorization, token_env_var)
+
+
 def resolve_workspace_id(workspace: str, api_endpoint: str, headers: dict[str, str]) -> int:
     org_name, workspace_name = workspace.split("/", 1)
 
@@ -67,6 +125,7 @@ def fetch_all_tasks(base_url: str, headers: dict[str, str]) -> list[dict]:
 
 def fetch_run_data(run_id: str, workspace: str, api_endpoint: str, token: str) -> dict:
     headers = {"Authorization": f"Bearer {token}"}
+    validate_api_access(api_endpoint, headers=headers)
     ws_id = resolve_workspace_id(workspace, api_endpoint, headers)
 
     workflow_data = _api_get(

bin/test_benchmark_report_fetch.py

@@ -1,8 +1,9 @@
+from urllib.error import HTTPError
 from unittest.mock import patch
 
 import pytest
 
-from benchmark_report_fetch import fetch_all_tasks, fetch_run_data, resolve_workspace_id
+from benchmark_report_fetch import fetch_all_tasks, fetch_run_data, resolve_workspace_id, validate_api_access
 
 
 def test_resolve_workspace_id():
@@ -33,12 +34,32 @@ def test_fetch_all_tasks_paginates():
 
 def test_fetch_run_data_keys():
     with patch("benchmark_report_fetch.resolve_workspace_id", return_value=10):
-        with patch("benchmark_report_fetch._api_get") as mock_get:
-            with patch("benchmark_report_fetch.fetch_all_tasks", return_value=[{"task": {"id": 1}}]):
-                mock_get.side_effect = [
-                    {"workflow": {"id": "run1"}},
-                    {"metrics": []},
-                    {"progress": {"workflowProgress": {}}},
-                ]
-                data = fetch_run_data("run1", "org/ws", "https://api.example.com", "tok")
-                assert set(data.keys()) == {"workflow", "metrics", "tasks", "progress"}
+        with patch("benchmark_report_fetch.validate_api_access") as mock_validate:
+            with patch("benchmark_report_fetch._api_get") as mock_get:
+                with patch("benchmark_report_fetch.fetch_all_tasks", return_value=[{"task": {"id": 1}}]):
+                    mock_get.side_effect = [
+                        {"workflow": {"id": "run1"}},
+                        {"metrics": []},
+                        {"progress": {"workflowProgress": {}}},
+                    ]
+                    data = fetch_run_data("run1", "org/ws", "https://api.example.com", "tok")
+                    assert set(data.keys()) == {"workflow", "metrics", "tasks", "progress"}
+                    mock_validate.assert_called_once_with("https://api.example.com", headers={"Authorization": "Bearer tok"})
+
+
+def test_validate_api_access_bad_token():
+    error = HTTPError("https://api.example.com/user-info", 401, "Unauthorized", hdrs=None, fp=None)
+
+    with patch("benchmark_report_fetch._api_get") as mock_get:
+        mock_get.side_effect = [{}, error]
+
+        with pytest.raises(RuntimeError, match="Authentication failed at 'https://api.example.com/user-info'"):
+            validate_api_access("https://api.example.com", headers={"Authorization": "Bearer tok"})
+
+
+def test_validate_api_access_bad_endpoint():
+    error = HTTPError("https://bad.example.com/service-info", 404, "Not Found", hdrs=None, fp=None)
+
+    with patch("benchmark_report_fetch._api_get", side_effect=error):
+        with pytest.raises(RuntimeError, match="preflight failed at 'https://bad.example.com/service-info'"):
+            validate_api_access("https://bad.example.com", headers={"Authorization": "Bearer tok"})

lib/AGENTS.md

@@ -13,10 +13,16 @@ Plain `java.net.URL.openConnection()` HTTP client for Seqera Platform API. Used
 | `apiGet(url, headers)`                                | Single GET request, returns parsed JSON map                                      |
 | `apiGetAllTasks(baseUrl, headers)`                    | Paginated GET for `/tasks` endpoint (100/page)                                   |
 | `resolveWorkspaceId(workspace, apiEndpoint, headers)` | "org/workspace" string → numeric workspace ID                                    |
-| `fetchRunData(meta, apiEndpoint)`                     | Orchestrator: calls 4 endpoints per run → `{workflow, metrics, tasks, progress}` |
+| `fetchRunData(meta, apiEndpoint)`                     | Orchestrator: preflights `/service-info` + `/user-info`, then calls 4 run endpoints → `{workflow, metrics, tasks, progress}` |
 
-### API Endpoints Called (per run)
+### API Endpoints Called
 
+Preflight once per endpoint/token pair:
+
+1. `GET /service-info` — validates the API endpoint is reachable
+2. `GET /user-info` — validates the bearer token before workspace resolution
+
+Per run:
 1. `GET /workflow/{id}?workspaceId={wsId}` — run metadata
 2. `GET /workflow/{id}/metrics?workspaceId={wsId}` — resource metrics
 3. `GET /workflow/{id}/tasks?workspaceId={wsId}` — all tasks (paginated)
@@ -35,3 +41,4 @@ Calls `/orgs` → finds org by name → calls `/orgs/{orgId}/workspaces` → fin
 - Uses plain `java.net.URL.openConnection()` for HTTP requests — no external plugin dependency
 - Runs in the Nextflow head JVM, not in a container process
 - Network errors throw `RuntimeException` which will fail the pipeline
+- Bad endpoints fail during `/service-info` preflight; bad or expired tokens fail during `/user-info` preflight

lib/SeqeraApi.groovy

@@ -1,15 +1,27 @@
 // Seqera Platform API client using plain java.net HTTP
 
 class SeqeraApi {
+    private static final Set<String> validatedApiSessions = Collections.synchronizedSet(new HashSet<String>())
+
+    static class ApiRequestException extends RuntimeException {
+        final String url
+        final int statusCode
+
+        ApiRequestException(String url, int statusCode) {
+            super("API request failed: ${url} → HTTP ${statusCode}")
+            this.url = url
+            this.statusCode = statusCode
+        }
+    }
 
     static Map apiGet(String url, Map headers) {
-        def conn = new URL(url).openConnection()
+        def conn = (HttpURLConnection) new URL(url).openConnection()
         try {
             conn.setRequestMethod('GET')
             headers.each { k, v -> conn.setRequestProperty(k, v) }
             def code = conn.getResponseCode()
             if (code != 200) {
-                throw new RuntimeException("API request failed: ${url} → HTTP ${code}")
+                throw new ApiRequestException(url, code)
             }
             def stream = conn.getInputStream()
             try {
@@ -23,6 +35,69 @@ class SeqeraApi {
         }
     }
 
+    static void validateApiAccess(String apiEndpoint, Map headers, String tokenEnvVar, String token) {
+        def tokenDigest = java.security.MessageDigest.getInstance('SHA-256')
+            .digest(token.getBytes('UTF-8'))
+            .collect { String.format('%02x', it) }
+            .join()
+        def validationKey = "${apiEndpoint}|${tokenEnvVar}|${tokenDigest}"
+        if (validatedApiSessions.contains(validationKey)) {
+            return
+        }
+
+        synchronized (validatedApiSessions) {
+            if (validatedApiSessions.contains(validationKey)) {
+                return
+            }
+
+            try {
+                apiGet("${apiEndpoint}/service-info", headers)
+            } catch (ApiRequestException e) {
+                throw new RuntimeException(
+                    "Seqera Platform API preflight failed at '${apiEndpoint}/service-info'. " +
+                    "Check the API endpoint URL from --seqera_api_endpoint or the input samplesheet platform column. " +
+                    "Original error: ${e.message}",
+                    e
+                )
+            } catch (Exception e) {
+                throw new RuntimeException(
+                    "Could not reach Seqera Platform API preflight endpoint '${apiEndpoint}/service-info'. " +
+                    "Check the API endpoint URL, network access, and JVM truststore settings. " +
+                    "Original error: ${e.message}",
+                    e
+                )
+            }
+
+            try {
+                apiGet("${apiEndpoint}/user-info", headers)
+            } catch (ApiRequestException e) {
+                if (e.statusCode in [401, 403]) {
+                    throw new RuntimeException(
+                        "Authentication failed at '${apiEndpoint}/user-info'. " +
+                        "Check the access token stored in '${tokenEnvVar}'. " +
+                        "Original error: ${e.message}",
+                        e
+                    )
+                }
+                throw new RuntimeException(
+                    "Seqera Platform API auth preflight failed at '${apiEndpoint}/user-info'. " +
+                    "Check the API endpoint URL and the access token stored in '${tokenEnvVar}'. " +
+                    "Original error: ${e.message}",
+                    e
+                )
+            } catch (Exception e) {
+                throw new RuntimeException(
+                    "Could not complete Seqera Platform auth preflight at '${apiEndpoint}/user-info'. " +
+                    "Check network access and the access token stored in '${tokenEnvVar}'. " +
+                    "Original error: ${e.message}",
+                    e
+                )
+            }
+
+            validatedApiSessions.add(validationKey)
+        }
+    }
+
     /**
      * Paginate through /tasks endpoint. Returns flat list of all tasks.
      */
@@ -79,6 +154,7 @@ class SeqeraApi {
             )
         }
         def headers = ["Authorization": "Bearer ${token}"]
+        validateApiAccess(effectiveEndpoint, headers, tokenEnvVar, token)
         def wsId = resolveWorkspaceId(meta.workspace, effectiveEndpoint, headers)
         def base = "${effectiveEndpoint}/workflow/${meta.id}?workspaceId=${wsId}"
 

tests/AGENTS.md

@@ -2,7 +2,7 @@
 
 ## Layout
 
-Each pipeline-level scenario lives in its own directory:
+Pipeline-level scenarios live in their own directories:
 
 - `pipeline_api_only/`
 - `pipeline_mixed_no_benchmark/`
@@ -16,13 +16,21 @@ Each scenario directory contains:
 - `main.nf.test.snap` — the snapshot for that scenario
 - `AGENTS.md` — scenario-specific guidance for future edits
 
+Function-level nf-test coverage for `lib/` helpers lives separately under:
+
+- `lib/`
+
+Reference: nf-test Function Testing docs — https://www.nf-test.com/docs/testcases/nextflow_function/
+
 ## Conventions
 
 - One scenario per directory.
 - Keep assertions local and explicit rather than heavily abstracted.
 - Prefer stable fixture files under `workflows/nf_aggregate/assets/`.
 - If routing behavior changes, update the scenario-specific `AGENTS.md` along with the test.
-- This directory is for pipeline routing/integration scenarios only; stage-specific pytest tests now live beside the relevant module under `modules/local/*/tests/`.
+- Pipeline routing/integration scenarios live under `tests/pipeline_*/`.
+- Function-level nf-test coverage for Groovy helpers under `lib/` lives under `tests/lib/`.
+- Stage-specific pytest tests live beside the relevant module under `modules/local/*/tests/`.
 
 ## Running tests
 

tests/lib/AGENTS.md

@@ -0,0 +1,18 @@
+# tests/lib/
+
+## Purpose
+
+This directory holds nf-test `nextflow_function` tests for helper code under `lib/`.
+
+Current coverage:
+
+- `SeqeraApi.groovy.test` — unit-ish behavioral coverage for `SeqeraApi` helpers using nf-test function tests and inline Groovy stubbing.
+
+## Conventions
+
+Reference: nf-test Function Testing docs — https://www.nf-test.com/docs/testcases/nextflow_function/
+
+- Prefer `nextflow_function` tests here instead of introducing a separate Spock/Gradle harness.
+- Keep assertions local and explicit.
+- Stub `SeqeraApi.metaClass.'static'.apiGet` inline when isolating pagination or workspace-resolution behavior.
+- Reserve pipeline routing/integration scenarios for the sibling `tests/pipeline_*/` directories.