feat: launcher orchestration support (current-task, ParentRunId tagging, per-launch key retirement)

CHANGELOG.md

@@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Added
+
+- NGS360: a launcher running as a WES run (e.g. an AWS Batch job) finds its own run
+  through `WES_RUN_ID`: `get_current_task()` returns it, and `submit_task()` tags each
+  run it submits with `ParentRunId` (PR #132)
+- NGS360: when `NGS360_AUTH_KEY_ID` names the API key in `NGS360_AUTH_TOKEN`, the key is
+  retired when the process exits, so a key minted for one launch does not outlive it
+  (PR #132)
+
+### Fixed
+
+- `PlatformFactory.detect_platform()` no longer logs credential values in the
+  environment dump it writes when no platform is detected (PR #132)
+
 ## [v0.6.1](https://github.com/NGS360/PAML/releases/tag/v0.6.1) - 2026-09-21
 
 ### Changed

src/cwl_platform/__init__.py

@@ -21,6 +21,26 @@
     'NGS360': NGS360Platform
 }
 
+# Substrings that mark an environment variable as carrying a credential. The
+# environment dump below goes to whatever collects the container's logs, which
+# is a wider audience than whoever provisioned the credential.
+_SECRET_NAME_MARKERS = ('TOKEN', 'KEY', 'SECRET', 'PASSWORD')
+
+
+def _redact(name, value):
+    '''
+    Return an environment variable's value, masked if the name marks it secret
+
+    :param name: Environment variable name
+    :param value: Environment variable value
+    :return: The value, or a mask of it
+    '''
+    if any(marker in name.upper() for marker in _SECRET_NAME_MARKERS):
+        # Length is kept because "set but empty" and "set to a truncated value"
+        # are both real misconfigurations worth telling apart in a log.
+        return f"<redacted, {len(value)} chars>"
+    return value
+
 
 class PlatformFactory():
     ''' PlatformFactory '''
@@ -42,7 +62,7 @@ def detect_platform(self):
         # print out environment variables and raise an error
         logging.info("Environment Variables:")
         for name, value in os.environ.items():
-            logging.info("%s: %s", name, value)
+            logging.info("%s: %s", name, _redact(name, value))
 
         raise ValueError("Unable to detect platform")
 

src/cwl_platform/ngs360_platform.py

@@ -3,8 +3,24 @@
 
 This module implements the Platform abstract base class using the GA4GH
 Workflow Execution Service (WES) API.
+
+Environment contract, set by whoever starts the launcher container:
+
+    WES_API_ENDPOINT    WES API base URL, including the /ga4gh/wes/v1 prefix
+    NGS360_API_ENDPOINT NGS360 API base URL (projects, file upload)
+    NGS360_AUTH_TOKEN   Bearer token (or ngs360_ API key) of the user the
+                        launcher runs as; sent to both WES and the NGS360 API
+    NGS360_AUTH_KEY_ID  Id of that API key, when it was minted for this one
+                        launch. The key is retired when the launcher exits, so
+                        it does not outlive the run; its expiry covers a
+                        launcher that is killed before it can retire it.
+    WES_RUN_ID          This launcher's own WES run id, when the launcher is
+                        itself running as a WES run (e.g. as an AWS Batch job).
+                        Makes get_current_task() work and links every task this
+                        launcher submits back to it.
 """
 
+import atexit
 import os
 import json
 import logging
@@ -86,6 +102,7 @@ def __init__(self, name):
 
         self.ngs360_endpoint = None
         self._ngs360_auth_config = {}
+        self._auth_key_id = None
 
         self.projects = {}  # Map project names to project objects
         self.workflows = {}  # Map workflow names to workflow objects
@@ -98,6 +115,8 @@ def connect(self, **kwargs):
         :param kwargs: Connection parameters
             - api_endpoint: WES API endpoint URL
             - ngs360_auth_token: Authentication token for the WES API
+            - ngs360_auth_key_id: Id of the API key in ngs360_auth_token, to be
+              retired when this process exits
         """
         self.connected = False
 
@@ -137,6 +156,16 @@ def connect(self, **kwargs):
         if not self.ngs360_endpoint:
             raise ValueError("NGS360 API endpoint URL is required")
 
+        # Registered before the /auth/me probe so that a launcher which fails to
+        # connect still retires its key on the way out.
+        auth_key_id = kwargs.get(
+            "ngs360_auth_key_id",
+            os.environ.get("NGS360_AUTH_KEY_ID")
+        )
+        if auth_key_id and not self._auth_key_id:
+            self._auth_key_id = auth_key_id
+            atexit.register(self.retire_auth_key)
+
         # Test connection by getting current user info
         try:
             response = requests.get(
@@ -163,6 +192,32 @@ def _auth_headers(self):
             headers["Authorization"] = f"Bearer {self._ngs360_auth_config['token']}"
         return headers
 
+    def retire_auth_key(self):
+        """
+        Retire the API key this launcher was given for its run
+
+        Runs at process exit when NGS360_AUTH_KEY_ID is set. The key authenticates
+        its own retirement. Failure is logged, not raised: the key's expiry still
+        bounds it, and an exit handler has no caller to report to.
+
+        :return: True if the key was retired, False otherwise
+        """
+        if not self._auth_key_id:
+            return False
+        try:
+            response = requests.delete(
+                f"{self.ngs360_endpoint}/api/v1/auth/api-keys/{self._auth_key_id}",
+                headers=self._auth_headers(),
+                timeout=30
+            )
+            response.raise_for_status()
+        except requests.RequestException as e:
+            self.logger.warning("Failed to retire API key %s: %s", self._auth_key_id, e)
+            return False
+        self.logger.info("Retired API key %s", self._auth_key_id)
+        self._auth_key_id = None
+        return True
+
     def _make_request(self, method, path, **kwargs):
         """
         Make a request to the WES API
@@ -407,9 +462,31 @@ def get_current_task(self):
         """
         Get the current task
 
-        Note: WES API doesn't have a concept of current task, so this returns None
-        """
-        return None
+        A launcher that is itself a WES run -- e.g. one running as an AWS Batch
+        job -- is told its own run id through WES_RUN_ID, the same way the
+        SevenBridges platform uses TASK_ID. Its own run record is where the
+        launcher reads back the inputs it was started with.
+
+        :return: WESTask object for this launcher's own run
+        :raises ValueError: If WES_RUN_ID is not set
+        """
+        run_id = os.environ.get("WES_RUN_ID")
+        if not run_id:
+            raise ValueError("ERROR: Environment variable WES_RUN_ID not set.")
+        self.logger.info("WES_RUN_ID: %s", run_id)
+
+        response = self._make_request("GET", f"runs/{run_id}")
+        request = response.get("request") or {}
+
+        return WESTask(
+            run_id=run_id,
+            name=response.get("name") or "",
+            state=self.STATE_MAP.get(response.get("state", "UNKNOWN"), "Unknown"),
+            # workflow_params are what the launcher was submitted with, which is
+            # what get_task_input reads back.
+            inputs=request.get("workflow_params") or {},
+            outputs=response.get("outputs") or {},
+        )
 
     def get_task_cost(self, task):
         """
@@ -654,16 +731,25 @@ def submit_task(self, name, project, workflow, parameters, execution_settings=No
         else:
             files = None
 
+        tags = {
+            "ProjectId": project["project_id"],
+            "TaskName": name,
+        }
+        # When this launcher is itself a WES run, link what it submits back to it.
+        # WES promotes ParentRunId into an indexed column, which is what lets it
+        # roll up launcher progress without the launcher reporting its own status,
+        # and what lets a restarted launcher find work it already submitted.
+        launcher_run_id = os.environ.get("WES_RUN_ID")
+        if launcher_run_id:
+            tags["ParentRunId"] = launcher_run_id
+
         # Prepare the request data
         data = {
             "workflow_params": json.dumps(parameters),
             "workflow_type": workflow_type,
             "workflow_type_version": workflow_type_version,
             "workflow_url": workflow_url,
-            "tags": json.dumps({
-                "ProjectId": project["project_id"],
-                "TaskName": name,
-            }),
+            "tags": json.dumps(tags),
             "workflow_engine_parameters": json.dumps(
                 workflow_engine_parameters
             ),