From df13cf31717aabb21da1c6fd79a7433650ab07f7 Mon Sep 17 00:00:00 2001
From: Zhengyuan Su <su.zhengyuan@u.nus.edu>
Date: Tue, 15 Sep 2026 22:48:06 +0700
Subject: [PATCH 1/8] feat: make API executor endpoint-agnostic with env-var
 credential reference
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Extend the API executor with a general, endpoint-agnostic credential
path while keeping the legacy Nebula path intact. spec.api.url wins over
NEBULA_API_BASE_URL; spec.api.auth.credential_env wins over
NEBULA_API_TOKEN. The secret never travels in the workflow spec — only
the env-var reference does.

A call with no credential fails closed unless spec.api.auth.mode: none
opts out explicitly, so a forgotten credential errors instead of
silently going out anonymous.

Co-Authored-By: Claude Code <noreply@anthropic.com>
Signed-off-by: Zhengyuan Su <su.zhengyuan@u.nus.edu>
---
 docs/WORKFLOWS.md                     |  33 +++++
 examples/templates/api_two_stage.yaml |   9 +-
 src/worker/executors/api_executor.py  |  42 ++++--
 tests/worker/test_api_executor.py     | 195 ++++++++++++++++++++++++++
 4 files changed, 269 insertions(+), 10 deletions(-)
 create mode 100644 tests/worker/test_api_executor.py

diff --git a/docs/WORKFLOWS.md b/docs/WORKFLOWS.md
index 10f3c57ef..908e4a054 100644
--- a/docs/WORKFLOWS.md
+++ b/docs/WORKFLOWS.md
@@ -69,6 +69,39 @@ parent output substitution and validation. See
 `src/worker/executors/utils/graph_templates.py` for the templating
 contract.
 
+## API task
+
+`taskType: api` performs a single HTTP request. The endpoint and credential are caller-supplied and endpoint-agnostic, with a legacy Nebula fallback preserved for backward compatibility.
+
+URL precedence: `spec.api.url` names the full request URL and wins; when absent, the executor falls back to `NEBULA_API_BASE_URL` (appending `/v1/chat/completions`).
+
+Credential precedence: an explicit `spec.api.auth.credential_env` (naming a worker env var holding the token) wins; when no `auth` block is present, the executor falls back to `NEBULA_API_TOKEN`. `auth.header` and `auth.scheme` override the default `Authorization` / `Bearer`.
+
+```yaml
+spec:
+  taskType: api
+  api:
+    url: https://api.example.com/v1/chat/completions
+    method: POST
+    auth:
+      credential_env: LLM_API_TOKEN   # env var on the worker holding the token
+      header: Authorization           # default
+      scheme: Bearer                  # default
+    headers:
+      Content-Type: application/json
+    body:
+      model: gpt-4o
+      messages:
+        - role: user
+          content: Hello
+    response:
+      parse_json: true
+```
+
+When `auth.credential_env` is set but the named env var is not present on the worker, the task fails closed with an error naming the missing variable — it never falls back to another credential and never calls unauthenticated.
+
+A call is authenticated from `auth.credential_env`, `NEBULA_API_TOKEN`, or an `Authorization` header supplied directly in `headers`. A call with none of these fails closed unless the caller opts out explicitly with `auth.mode: none`, so a forgotten credential errors instead of silently going out anonymous.
+
 ## data_retrieval: type lumid
 
 `type: lumid` routes the retrieval through lumid-data-app (HTTP). Three
diff --git a/examples/templates/api_two_stage.yaml b/examples/templates/api_two_stage.yaml
index 70a0c7700..fdc5058e8 100644
--- a/examples/templates/api_two_stage.yaml
+++ b/examples/templates/api_two_stage.yaml
@@ -4,8 +4,10 @@
 # Stage 1 calls a chat completion endpoint and returns raw text.
 # Stage 2 sends Stage 1's returned text as the next prompt.
 #
-# NOTE: Configure NEBULA_API_BASE_URL and NEBULA_API_TOKEN on the worker before
-# submitting. APIExecutor injects the Authorization header from NEBULA_API_TOKEN.
+# Stage 1 uses the general path: spec.api.url names the endpoint and
+# spec.api.auth.credential_env names a worker env var holding the token.
+# Stage 2 omits url and auth, so it falls back to the legacy Nebula
+# defaults (NEBULA_API_BASE_URL + NEBULA_API_TOKEN).
 
 apiVersion: flowmesh/v1
 kind: APITask
@@ -19,7 +21,10 @@ spec:
     - name: stage-1
       spec:
         api:
+          url: https://api.openai.com/v1/chat/completions
           method: POST
+          auth:
+            credential_env: LLM_API_TOKEN
           headers:
             Content-Type: application/json
           body:
diff --git a/src/worker/executors/api_executor.py b/src/worker/executors/api_executor.py
index 1a56c7b13..9a6b75620 100644
--- a/src/worker/executors/api_executor.py
+++ b/src/worker/executors/api_executor.py
@@ -19,12 +19,12 @@
 
 
 class APIExecutor(Executor):
-    """Executor that performs a single HTTP request defined by task YAML.
+    """Performs a single HTTP request defined by task YAML.
 
-    Uses a class-level connection pool keyed by (base_url, timeout, verify_tls,
-    follow_redirects) so that repeated calls to the same endpoint (e.g. a trading
-    bot hitting QuantArena every few seconds) reuse the underlying TCP/TLS
-    connection instead of paying the handshake cost on every request.
+    Endpoint-agnostic with a legacy Nebula fallback. ``spec.api.url`` wins over
+    ``NEBULA_API_BASE_URL``; ``spec.api.auth.credential_env`` wins over
+    ``NEBULA_API_TOKEN``. A call with no credential fails closed unless
+    ``spec.api.auth.mode: none`` opts out explicitly.
     """
 
     name = "api"
@@ -110,9 +110,35 @@ def run(self, task: ExecutorTask, out_dir: Path) -> APIResult:
         if not isinstance(headers, dict):
             raise ExecutionError("spec.api.headers must be a mapping")
 
-        token = os.getenv("NEBULA_API_TOKEN")
-        if token and not any(k.lower() == "authorization" for k in headers):
-            headers["Authorization"] = f"Bearer {token}"
+        auth = api_cfg.get("auth")
+        if auth is not None:
+            if not isinstance(auth, dict):
+                raise ExecutionError("spec.api.auth must be a mapping")
+            if auth.get("mode") != "none":
+                credential_env = auth.get("credential_env")
+                if not isinstance(credential_env, str) or not credential_env:
+                    raise ExecutionError(
+                        "spec.api.auth.credential_env must name an env var"
+                    )
+                token = os.getenv(credential_env)
+                if not token:
+                    raise ExecutionError(
+                        f"spec.api.auth.credential_env names {credential_env}, "
+                        "which is not set on the worker"
+                    )
+                header_name = str(auth.get("header", "Authorization"))
+                scheme = str(auth.get("scheme", "Bearer"))
+                if not any(k.lower() == header_name.lower() for k in headers):
+                    headers[header_name] = f"{scheme} {token}"
+        else:
+            token = os.getenv("NEBULA_API_TOKEN")
+            if token and not any(k.lower() == "authorization" for k in headers):
+                headers["Authorization"] = f"Bearer {token}"
+            elif not token and not any(k.lower() == "authorization" for k in headers):
+                raise ExecutionError(
+                    "no credential configured: set spec.api.auth.credential_env, "
+                    "NEBULA_API_TOKEN, an Authorization header, or auth.mode 'none'"
+                )
 
         params = api_cfg.get("params")
         if params is not None and not isinstance(params, dict):
diff --git a/tests/worker/test_api_executor.py b/tests/worker/test_api_executor.py
new file mode 100644
index 000000000..e0ea2a413
--- /dev/null
+++ b/tests/worker/test_api_executor.py
@@ -0,0 +1,195 @@
+"""Tests for the endpoint-agnostic API executor auth and URL handling."""
+
+from pathlib import Path
+from unittest.mock import patch
+
+import httpx
+import pytest
+
+from shared.tasks.worker_message import WorkerTaskMessage
+from worker.executors.api_executor import APIExecutor
+from worker.executors.base_executor import ExecutionError
+
+
+def _task_message(**spec_updates: object) -> WorkerTaskMessage:
+    payload = {
+        "task_id": "task-api",
+        "workflow_id": "wf-1",
+        "owner_id": "owner",
+        "assigned_worker": "worker-1",
+        "dispatched_at": "2026-03-22T00:00:00Z",
+        "task": {
+            "apiVersion": "mloc/v1",
+            "kind": "Task",
+            "metadata": {"name": "wf:api"},
+            "spec": {
+                "taskType": "api",
+                "api": {
+                    "url": "https://api.example.com/v1/chat/completions",
+                    "method": "POST",
+                    "body": {
+                        "model": "gpt-4o",
+                        "messages": [{"role": "user", "content": "hi"}],
+                    },
+                    **spec_updates,
+                },
+            },
+        },
+    }
+    return WorkerTaskMessage.model_validate(payload)
+
+
+class _RecordingTransport(httpx.MockTransport):
+    """MockTransport that records the request it served."""
+
+    def __init__(self) -> None:
+        self.request: httpx.Request | None = None
+        super().__init__(self._handler)
+
+    def _handler(self, request: httpx.Request) -> httpx.Response:
+        self.request = request
+        return httpx.Response(
+            200,
+            json={
+                "choices": [{"message": {"content": "hello"}}],
+                "usage": {"total_tokens": 3},
+            },
+        )
+
+
+def _run(
+    executor: APIExecutor, task: WorkerTaskMessage, transport: _RecordingTransport
+) -> None:
+    with patch.object(
+        APIExecutor, "_get_client", return_value=httpx.Client(transport=transport)
+    ):
+        executor.run(task, Path("/tmp/out"))
+
+
+class TestAuth:
+    def test_injects_bearer_from_credential_env(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setenv("LLM_API_TOKEN", "secret-token")
+        task = _task_message(auth={"credential_env": "LLM_API_TOKEN"})
+        transport = _RecordingTransport()
+        _run(APIExecutor.__new__(APIExecutor), task, transport)
+        assert transport.request is not None
+        assert transport.request.headers["Authorization"] == "Bearer secret-token"
+
+    def test_custom_header_and_scheme(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("MY_KEY", "abc")
+        task = _task_message(
+            auth={"credential_env": "MY_KEY", "header": "X-API-Key", "scheme": "Token"}
+        )
+        transport = _RecordingTransport()
+        _run(APIExecutor.__new__(APIExecutor), task, transport)
+        assert transport.request is not None
+        assert transport.request.headers["X-API-Key"] == "Token abc"
+
+    def test_missing_credential_env_fails_closed(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.delenv("LLM_API_TOKEN", raising=False)
+        task = _task_message(auth={"credential_env": "LLM_API_TOKEN"})
+        with pytest.raises(ExecutionError, match="LLM_API_TOKEN"):
+            _run(APIExecutor.__new__(APIExecutor), task, _RecordingTransport())
+
+    def test_auth_not_mapping_rejected(self) -> None:
+        task = _task_message(auth="Bearer x")
+        with pytest.raises(ExecutionError, match="spec.api.auth must be a mapping"):
+            _run(APIExecutor.__new__(APIExecutor), task, _RecordingTransport())
+
+    def test_credential_env_must_be_nonempty_string(self) -> None:
+        task = _task_message(auth={"credential_env": ""})
+        with pytest.raises(ExecutionError, match="credential_env"):
+            _run(APIExecutor.__new__(APIExecutor), task, _RecordingTransport())
+
+    def test_explicit_authorization_header_wins(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setenv("LLM_API_TOKEN", "secret-token")
+        task = _task_message(
+            auth={"credential_env": "LLM_API_TOKEN"},
+            headers={"Authorization": "Bearer explicit"},
+        )
+        transport = _RecordingTransport()
+        _run(APIExecutor.__new__(APIExecutor), task, transport)
+        assert transport.request is not None
+        assert transport.request.headers["Authorization"] == "Bearer explicit"
+
+
+class TestUrl:
+    def test_url_required_when_no_base_url(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.delenv("NEBULA_API_BASE_URL", raising=False)
+        task = _task_message(url=None)
+        with pytest.raises(ExecutionError, match="spec.api.url or NEBULA_API_BASE_URL"):
+            _run(APIExecutor.__new__(APIExecutor), task, _RecordingTransport())
+
+    def test_url_falls_back_to_nebula_base_url(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setenv("NEBULA_API_BASE_URL", "https://nebula.example.com")
+        monkeypatch.setenv("NEBULA_API_TOKEN", "nebula-token")
+        task = _task_message(url=None)
+        transport = _RecordingTransport()
+        _run(APIExecutor.__new__(APIExecutor), task, transport)
+        assert transport.request is not None
+        assert transport.request.url == "https://nebula.example.com/v1/chat/completions"
+
+    def test_explicit_authorization_header_allows_anonymous_call(self) -> None:
+        task = _task_message(headers={"Authorization": "Bearer explicit"})
+        transport = _RecordingTransport()
+        _run(APIExecutor.__new__(APIExecutor), task, transport)
+        assert transport.request is not None
+        assert transport.request.headers["Authorization"] == "Bearer explicit"
+
+    def test_auth_mode_none_allows_anonymous_call(self) -> None:
+        task = _task_message(auth={"mode": "none"})
+        transport = _RecordingTransport()
+        _run(APIExecutor.__new__(APIExecutor), task, transport)
+        assert transport.request is not None
+        assert "Authorization" not in transport.request.headers
+
+
+class TestLegacyNebulaPath:
+    def test_no_auth_block_uses_nebula_token(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setenv("NEBULA_API_TOKEN", "nebula-token")
+        task = _task_message()
+        transport = _RecordingTransport()
+        _run(APIExecutor.__new__(APIExecutor), task, transport)
+        assert transport.request is not None
+        assert transport.request.headers["Authorization"] == "Bearer nebula-token"
+
+    def test_no_auth_block_and_no_token_fails_closed(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.delenv("NEBULA_API_TOKEN", raising=False)
+        task = _task_message()
+        with pytest.raises(ExecutionError, match="NEBULA_API_TOKEN"):
+            _run(APIExecutor.__new__(APIExecutor), task, _RecordingTransport())
+
+    def test_credential_env_wins_over_nebula_token(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setenv("NEBULA_API_TOKEN", "nebula-token")
+        monkeypatch.setenv("LLM_API_TOKEN", "general-token")
+        task = _task_message(auth={"credential_env": "LLM_API_TOKEN"})
+        transport = _RecordingTransport()
+        _run(APIExecutor.__new__(APIExecutor), task, transport)
+        assert transport.request is not None
+        assert transport.request.headers["Authorization"] == "Bearer general-token"
+
+    def test_spec_authorization_header_wins_over_nebula_token(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setenv("NEBULA_API_TOKEN", "nebula-token")
+        task = _task_message(headers={"Authorization": "Bearer explicit"})
+        transport = _RecordingTransport()
+        _run(APIExecutor.__new__(APIExecutor), task, transport)
+        assert transport.request is not None
+        assert transport.request.headers["Authorization"] == "Bearer explicit"

From 37bc835e47c458f52a5d558471c60144e8fa0465 Mon Sep 17 00:00:00 2001
From: Zhengyuan Su <su.zhengyuan@u.nus.edu>
Date: Wed, 16 Sep 2026 16:13:11 +0700
Subject: [PATCH 2/8] fix: default to lum.id/llm + deepseek and drop the worker
 env-var credential

Signed-off-by: Zhengyuan Su <su.zhengyuan@u.nus.edu>
---
 docs/WORKFLOWS.md                     |  17 ++--
 examples/templates/api_two_stage.yaml |  15 ++--
 src/worker/executors/api_executor.py  |  55 +++++-------
 tests/worker/test_api_executor.py     | 120 +++++++-------------------
 4 files changed, 60 insertions(+), 147 deletions(-)

diff --git a/docs/WORKFLOWS.md b/docs/WORKFLOWS.md
index 908e4a054..e336c6fe4 100644
--- a/docs/WORKFLOWS.md
+++ b/docs/WORKFLOWS.md
@@ -71,26 +71,21 @@ contract.
 
 ## API task
 
-`taskType: api` performs a single HTTP request. The endpoint and credential are caller-supplied and endpoint-agnostic, with a legacy Nebula fallback preserved for backward compatibility.
+`taskType: api` performs a single HTTP request. The endpoint and model default to the lum.id/llm chat-completions endpoint and the deepseek model when the spec omits them.
 
-URL precedence: `spec.api.url` names the full request URL and wins; when absent, the executor falls back to `NEBULA_API_BASE_URL` (appending `/v1/chat/completions`).
+`spec.api.url` names the full request URL; when absent, the executor defaults to `https://lum.id/llm/v1/chat/completions`. When the request body has no `model` key, the executor injects `deepseek-v4-flash`.
 
-Credential precedence: an explicit `spec.api.auth.credential_env` (naming a worker env var holding the token) wins; when no `auth` block is present, the executor falls back to `NEBULA_API_TOKEN`. `auth.header` and `auth.scheme` override the default `Authorization` / `Bearer`.
+The credential is either an `Authorization` header supplied directly in `spec.api.headers`, or the worker's own `NEBULA_API_TOKEN`, which the executor fills in as `Authorization: Bearer <NEBULA_API_TOKEN>`. A call with neither fails closed.
 
 ```yaml
 spec:
   taskType: api
   api:
-    url: https://api.example.com/v1/chat/completions
+    url: https://lum.id/llm/v1/chat/completions
     method: POST
-    auth:
-      credential_env: LLM_API_TOKEN   # env var on the worker holding the token
-      header: Authorization           # default
-      scheme: Bearer                  # default
     headers:
       Content-Type: application/json
     body:
-      model: gpt-4o
       messages:
         - role: user
           content: Hello
@@ -98,9 +93,7 @@ spec:
       parse_json: true
 ```
 
-When `auth.credential_env` is set but the named env var is not present on the worker, the task fails closed with an error naming the missing variable — it never falls back to another credential and never calls unauthenticated.
-
-A call is authenticated from `auth.credential_env`, `NEBULA_API_TOKEN`, or an `Authorization` header supplied directly in `headers`. A call with none of these fails closed unless the caller opts out explicitly with `auth.mode: none`, so a forgotten credential errors instead of silently going out anonymous.
+When no `Authorization` header is supplied and `NEBULA_API_TOKEN` is not set on the worker, the task fails closed with an error — it never calls unauthenticated.
 
 ## data_retrieval: type lumid
 
diff --git a/examples/templates/api_two_stage.yaml b/examples/templates/api_two_stage.yaml
index fdc5058e8..3c204ab50 100644
--- a/examples/templates/api_two_stage.yaml
+++ b/examples/templates/api_two_stage.yaml
@@ -4,10 +4,10 @@
 # Stage 1 calls a chat completion endpoint and returns raw text.
 # Stage 2 sends Stage 1's returned text as the next prompt.
 #
-# Stage 1 uses the general path: spec.api.url names the endpoint and
-# spec.api.auth.credential_env names a worker env var holding the token.
-# Stage 2 omits url and auth, so it falls back to the legacy Nebula
-# defaults (NEBULA_API_BASE_URL + NEBULA_API_TOKEN).
+# Stage 1 names the endpoint explicitly via spec.api.url. Stage 2 omits url
+# and model, so it uses the defaults: https://lum.id/llm/v1/chat/completions
+# with the deepseek-v4-flash model. Both stages authenticate with the
+# worker's own NEBULA_API_TOKEN.
 
 apiVersion: flowmesh/v1
 kind: APITask
@@ -21,14 +21,12 @@ spec:
     - name: stage-1
       spec:
         api:
-          url: https://api.openai.com/v1/chat/completions
+          url: https://lum.id/llm/v1/chat/completions
           method: POST
-          auth:
-            credential_env: LLM_API_TOKEN
           headers:
             Content-Type: application/json
           body:
-            model: gpt-4o
+            model: deepseek-v4-flash
             messages:
               - role: user
                 content: Please explain vector databases in simple terms.
@@ -45,7 +43,6 @@ spec:
           headers:
             Content-Type: application/json
           body:
-            model: gpt-4o
             messages:
               - role: user
                 content: "The previous stage's response is as follows. Please provide a simpler explanation: \n${stage-1.text}"
diff --git a/src/worker/executors/api_executor.py b/src/worker/executors/api_executor.py
index 9a6b75620..436200cf5 100644
--- a/src/worker/executors/api_executor.py
+++ b/src/worker/executors/api_executor.py
@@ -17,14 +17,17 @@
 # Cache key: (base_url, timeout_seconds, verify_tls, follow_redirects)
 _ClientKey = tuple[str, float, bool, bool]
 
+_DEFAULT_BASE_URL = "https://lum.id/llm"
+_DEFAULT_MODEL = "deepseek-v4-flash"
+
 
 class APIExecutor(Executor):
     """Performs a single HTTP request defined by task YAML.
 
-    Endpoint-agnostic with a legacy Nebula fallback. ``spec.api.url`` wins over
-    ``NEBULA_API_BASE_URL``; ``spec.api.auth.credential_env`` wins over
-    ``NEBULA_API_TOKEN``. A call with no credential fails closed unless
-    ``spec.api.auth.mode: none`` opts out explicitly.
+    Defaults to the lum.id/llm chat-completions endpoint and the deepseek model
+    when the spec omits them. The credential is the worker's own
+    ``NEBULA_API_TOKEN`` unless the spec supplies an ``Authorization`` header
+    directly. A call with no credential fails closed.
     """
 
     name = "api"
@@ -100,45 +103,21 @@ def run(self, task: ExecutorTask, out_dir: Path) -> APIResult:
 
         url = api_cfg.get("url")
         if url is None:
-            url = os.getenv("NEBULA_API_BASE_URL")
-            if not url:
-                raise ExecutionError("spec.api.url or NEBULA_API_BASE_URL is required")
-            url = url.rstrip("/") + "/v1/chat/completions"
+            url = _DEFAULT_BASE_URL + "/v1/chat/completions"
 
         method = str(api_cfg.get("method", "POST")).upper()
         headers = api_cfg.get("headers", {})
         if not isinstance(headers, dict):
             raise ExecutionError("spec.api.headers must be a mapping")
 
-        auth = api_cfg.get("auth")
-        if auth is not None:
-            if not isinstance(auth, dict):
-                raise ExecutionError("spec.api.auth must be a mapping")
-            if auth.get("mode") != "none":
-                credential_env = auth.get("credential_env")
-                if not isinstance(credential_env, str) or not credential_env:
-                    raise ExecutionError(
-                        "spec.api.auth.credential_env must name an env var"
-                    )
-                token = os.getenv(credential_env)
-                if not token:
-                    raise ExecutionError(
-                        f"spec.api.auth.credential_env names {credential_env}, "
-                        "which is not set on the worker"
-                    )
-                header_name = str(auth.get("header", "Authorization"))
-                scheme = str(auth.get("scheme", "Bearer"))
-                if not any(k.lower() == header_name.lower() for k in headers):
-                    headers[header_name] = f"{scheme} {token}"
-        else:
+        if not any(k.lower() == "authorization" for k in headers):
             token = os.getenv("NEBULA_API_TOKEN")
-            if token and not any(k.lower() == "authorization" for k in headers):
-                headers["Authorization"] = f"Bearer {token}"
-            elif not token and not any(k.lower() == "authorization" for k in headers):
+            if not token:
                 raise ExecutionError(
-                    "no credential configured: set spec.api.auth.credential_env, "
-                    "NEBULA_API_TOKEN, an Authorization header, or auth.mode 'none'"
+                    "no credential configured: set an Authorization header or "
+                    "NEBULA_API_TOKEN"
                 )
+            headers["Authorization"] = f"Bearer {token}"
 
         params = api_cfg.get("params")
         if params is not None and not isinstance(params, dict):
@@ -163,9 +142,15 @@ def run(self, task: ExecutorTask, out_dir: Path) -> APIResult:
 
         request_kwargs: dict[str, Any] = {}
         if json_payload is not None:
+            if isinstance(json_payload, dict) and "model" not in json_payload:
+                json_payload = {**json_payload, "model": _DEFAULT_MODEL}
             request_kwargs["json"] = json_payload
         elif body is not None:
-            if isinstance(body, (dict, list)):
+            if isinstance(body, dict):
+                if "model" not in body:
+                    body = {**body, "model": _DEFAULT_MODEL}
+                request_kwargs["json"] = body
+            elif isinstance(body, list):
                 request_kwargs["json"] = body
             else:
                 request_kwargs["content"] = body
diff --git a/tests/worker/test_api_executor.py b/tests/worker/test_api_executor.py
index e0ea2a413..12861033a 100644
--- a/tests/worker/test_api_executor.py
+++ b/tests/worker/test_api_executor.py
@@ -1,4 +1,4 @@
-"""Tests for the endpoint-agnostic API executor auth and URL handling."""
+"""Tests for the API executor defaults, credential handling, and model injection."""
 
 from pathlib import Path
 from unittest.mock import patch
@@ -28,7 +28,6 @@ def _task_message(**spec_updates: object) -> WorkerTaskMessage:
                     "url": "https://api.example.com/v1/chat/completions",
                     "method": "POST",
                     "body": {
-                        "model": "gpt-4o",
                         "messages": [{"role": "user", "content": "hi"}],
                     },
                     **spec_updates,
@@ -66,106 +65,60 @@ def _run(
         executor.run(task, Path("/tmp/out"))
 
 
-class TestAuth:
-    def test_injects_bearer_from_credential_env(
-        self, monkeypatch: pytest.MonkeyPatch
-    ) -> None:
-        monkeypatch.setenv("LLM_API_TOKEN", "secret-token")
-        task = _task_message(auth={"credential_env": "LLM_API_TOKEN"})
+class TestDefaults:
+    def test_default_url_is_lum_id_llm(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("NEBULA_API_TOKEN", "token")
+        task = _task_message(url=None)
         transport = _RecordingTransport()
         _run(APIExecutor.__new__(APIExecutor), task, transport)
         assert transport.request is not None
-        assert transport.request.headers["Authorization"] == "Bearer secret-token"
+        assert transport.request.url == "https://lum.id/llm/v1/chat/completions"
 
-    def test_custom_header_and_scheme(self, monkeypatch: pytest.MonkeyPatch) -> None:
-        monkeypatch.setenv("MY_KEY", "abc")
-        task = _task_message(
-            auth={"credential_env": "MY_KEY", "header": "X-API-Key", "scheme": "Token"}
-        )
+    def test_default_model_injected_when_body_has_no_model(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setenv("NEBULA_API_TOKEN", "token")
+        task = _task_message()
         transport = _RecordingTransport()
         _run(APIExecutor.__new__(APIExecutor), task, transport)
         assert transport.request is not None
-        assert transport.request.headers["X-API-Key"] == "Token abc"
+        sent = _sent_json(transport.request)
+        assert sent["model"] == "deepseek-v4-flash"
 
-    def test_missing_credential_env_fails_closed(
+    def test_explicit_model_left_untouched(
         self, monkeypatch: pytest.MonkeyPatch
     ) -> None:
-        monkeypatch.delenv("LLM_API_TOKEN", raising=False)
-        task = _task_message(auth={"credential_env": "LLM_API_TOKEN"})
-        with pytest.raises(ExecutionError, match="LLM_API_TOKEN"):
-            _run(APIExecutor.__new__(APIExecutor), task, _RecordingTransport())
-
-    def test_auth_not_mapping_rejected(self) -> None:
-        task = _task_message(auth="Bearer x")
-        with pytest.raises(ExecutionError, match="spec.api.auth must be a mapping"):
-            _run(APIExecutor.__new__(APIExecutor), task, _RecordingTransport())
-
-    def test_credential_env_must_be_nonempty_string(self) -> None:
-        task = _task_message(auth={"credential_env": ""})
-        with pytest.raises(ExecutionError, match="credential_env"):
-            _run(APIExecutor.__new__(APIExecutor), task, _RecordingTransport())
-
-    def test_explicit_authorization_header_wins(
-        self, monkeypatch: pytest.MonkeyPatch
-    ) -> None:
-        monkeypatch.setenv("LLM_API_TOKEN", "secret-token")
-        task = _task_message(
-            auth={"credential_env": "LLM_API_TOKEN"},
-            headers={"Authorization": "Bearer explicit"},
-        )
+        monkeypatch.setenv("NEBULA_API_TOKEN", "token")
+        task = _task_message(body={"model": "gpt-4o", "messages": []})
         transport = _RecordingTransport()
         _run(APIExecutor.__new__(APIExecutor), task, transport)
         assert transport.request is not None
-        assert transport.request.headers["Authorization"] == "Bearer explicit"
+        sent = _sent_json(transport.request)
+        assert sent["model"] == "gpt-4o"
 
 
-class TestUrl:
-    def test_url_required_when_no_base_url(
+class TestCredential:
+    def test_explicit_authorization_header_left_untouched(
         self, monkeypatch: pytest.MonkeyPatch
     ) -> None:
-        monkeypatch.delenv("NEBULA_API_BASE_URL", raising=False)
-        task = _task_message(url=None)
-        with pytest.raises(ExecutionError, match="spec.api.url or NEBULA_API_BASE_URL"):
-            _run(APIExecutor.__new__(APIExecutor), task, _RecordingTransport())
-
-    def test_url_falls_back_to_nebula_base_url(
-        self, monkeypatch: pytest.MonkeyPatch
-    ) -> None:
-        monkeypatch.setenv("NEBULA_API_BASE_URL", "https://nebula.example.com")
-        monkeypatch.setenv("NEBULA_API_TOKEN", "nebula-token")
-        task = _task_message(url=None)
-        transport = _RecordingTransport()
-        _run(APIExecutor.__new__(APIExecutor), task, transport)
-        assert transport.request is not None
-        assert transport.request.url == "https://nebula.example.com/v1/chat/completions"
-
-    def test_explicit_authorization_header_allows_anonymous_call(self) -> None:
+        monkeypatch.setenv("NEBULA_API_TOKEN", "worker-token")
         task = _task_message(headers={"Authorization": "Bearer explicit"})
         transport = _RecordingTransport()
         _run(APIExecutor.__new__(APIExecutor), task, transport)
         assert transport.request is not None
         assert transport.request.headers["Authorization"] == "Bearer explicit"
 
-    def test_auth_mode_none_allows_anonymous_call(self) -> None:
-        task = _task_message(auth={"mode": "none"})
-        transport = _RecordingTransport()
-        _run(APIExecutor.__new__(APIExecutor), task, transport)
-        assert transport.request is not None
-        assert "Authorization" not in transport.request.headers
-
-
-class TestLegacyNebulaPath:
-    def test_no_auth_block_uses_nebula_token(
+    def test_nebula_token_used_when_no_header(
         self, monkeypatch: pytest.MonkeyPatch
     ) -> None:
-        monkeypatch.setenv("NEBULA_API_TOKEN", "nebula-token")
+        monkeypatch.setenv("NEBULA_API_TOKEN", "worker-token")
         task = _task_message()
         transport = _RecordingTransport()
         _run(APIExecutor.__new__(APIExecutor), task, transport)
         assert transport.request is not None
-        assert transport.request.headers["Authorization"] == "Bearer nebula-token"
+        assert transport.request.headers["Authorization"] == "Bearer worker-token"
 
-    def test_no_auth_block_and_no_token_fails_closed(
+    def test_missing_credential_fails_closed(
         self, monkeypatch: pytest.MonkeyPatch
     ) -> None:
         monkeypatch.delenv("NEBULA_API_TOKEN", raising=False)
@@ -173,23 +126,8 @@ def test_no_auth_block_and_no_token_fails_closed(
         with pytest.raises(ExecutionError, match="NEBULA_API_TOKEN"):
             _run(APIExecutor.__new__(APIExecutor), task, _RecordingTransport())
 
-    def test_credential_env_wins_over_nebula_token(
-        self, monkeypatch: pytest.MonkeyPatch
-    ) -> None:
-        monkeypatch.setenv("NEBULA_API_TOKEN", "nebula-token")
-        monkeypatch.setenv("LLM_API_TOKEN", "general-token")
-        task = _task_message(auth={"credential_env": "LLM_API_TOKEN"})
-        transport = _RecordingTransport()
-        _run(APIExecutor.__new__(APIExecutor), task, transport)
-        assert transport.request is not None
-        assert transport.request.headers["Authorization"] == "Bearer general-token"
 
-    def test_spec_authorization_header_wins_over_nebula_token(
-        self, monkeypatch: pytest.MonkeyPatch
-    ) -> None:
-        monkeypatch.setenv("NEBULA_API_TOKEN", "nebula-token")
-        task = _task_message(headers={"Authorization": "Bearer explicit"})
-        transport = _RecordingTransport()
-        _run(APIExecutor.__new__(APIExecutor), task, transport)
-        assert transport.request is not None
-        assert transport.request.headers["Authorization"] == "Bearer explicit"
+def _sent_json(request: httpx.Request) -> dict:
+    import json
+
+    return json.loads(request.content.decode("utf-8"))

From 586f0ffbf23f3cf710c84c63df6d3bf5c4430687 Mon Sep 17 00:00:00 2001
From: Zhengyuan Su <su.zhengyuan@u.nus.edu>
Date: Wed, 16 Sep 2026 16:43:50 +0700
Subject: [PATCH 3/8] fix: keep Nebula the default route and never send its
 token to a custom url

Signed-off-by: Zhengyuan Su <su.zhengyuan@u.nus.edu>
---
 docs/WORKFLOWS.md                     | 10 ++--
 examples/templates/api_two_stage.yaml | 13 +++--
 src/worker/executors/api_executor.py  | 40 ++++++-------
 tests/worker/test_api_executor.py     | 82 ++++++++++++---------------
 4 files changed, 65 insertions(+), 80 deletions(-)

diff --git a/docs/WORKFLOWS.md b/docs/WORKFLOWS.md
index e336c6fe4..404013e4a 100644
--- a/docs/WORKFLOWS.md
+++ b/docs/WORKFLOWS.md
@@ -71,21 +71,21 @@ contract.
 
 ## API task
 
-`taskType: api` performs a single HTTP request. The endpoint and model default to the lum.id/llm chat-completions endpoint and the deepseek model when the spec omits them.
+`taskType: api` performs a single HTTP request. By default it routes to the Nebula endpoint and authenticates with the worker's `NEBULA_API_TOKEN`.
 
-`spec.api.url` names the full request URL; when absent, the executor defaults to `https://lum.id/llm/v1/chat/completions`. When the request body has no `model` key, the executor injects `deepseek-v4-flash`.
+`spec.api.url` overrides the endpoint; when absent, the executor uses `NEBULA_API_BASE_URL` (appending `/v1/chat/completions`). `spec.api.headers` may supply an `Authorization` header directly.
 
-The credential is either an `Authorization` header supplied directly in `spec.api.headers`, or the worker's own `NEBULA_API_TOKEN`, which the executor fills in as `Authorization: Bearer <NEBULA_API_TOKEN>`. A call with neither fails closed.
+Credential handling: a caller-supplied `Authorization` header is always used as-is and never overwritten. With no header, `NEBULA_API_TOKEN` is injected only when the call is on the Nebula url (no custom `spec.api.url`) — the Nebula token is never sent to a custom endpoint. A Nebula-path call with no token available fails closed.
 
 ```yaml
 spec:
   taskType: api
   api:
-    url: https://lum.id/llm/v1/chat/completions
     method: POST
     headers:
       Content-Type: application/json
     body:
+      model: gpt-4o
       messages:
         - role: user
           content: Hello
@@ -93,8 +93,6 @@ spec:
       parse_json: true
 ```
 
-When no `Authorization` header is supplied and `NEBULA_API_TOKEN` is not set on the worker, the task fails closed with an error — it never calls unauthenticated.
-
 ## data_retrieval: type lumid
 
 `type: lumid` routes the retrieval through lumid-data-app (HTTP). Three
diff --git a/examples/templates/api_two_stage.yaml b/examples/templates/api_two_stage.yaml
index 3c204ab50..a8a83f812 100644
--- a/examples/templates/api_two_stage.yaml
+++ b/examples/templates/api_two_stage.yaml
@@ -4,10 +4,9 @@
 # Stage 1 calls a chat completion endpoint and returns raw text.
 # Stage 2 sends Stage 1's returned text as the next prompt.
 #
-# Stage 1 names the endpoint explicitly via spec.api.url. Stage 2 omits url
-# and model, so it uses the defaults: https://lum.id/llm/v1/chat/completions
-# with the deepseek-v4-flash model. Both stages authenticate with the
-# worker's own NEBULA_API_TOKEN.
+# Stage 1 names a custom endpoint via spec.api.url and supplies its own
+# Authorization header. Stage 2 omits url and header, so it uses the Nebula
+# defaults (NEBULA_API_BASE_URL + NEBULA_API_TOKEN).
 
 apiVersion: flowmesh/v1
 kind: APITask
@@ -21,12 +20,13 @@ spec:
     - name: stage-1
       spec:
         api:
-          url: https://lum.id/llm/v1/chat/completions
+          url: https://api.example.com/v1/chat/completions
           method: POST
           headers:
+            Authorization: Bearer <your-token>
             Content-Type: application/json
           body:
-            model: deepseek-v4-flash
+            model: gpt-4o
             messages:
               - role: user
                 content: Please explain vector databases in simple terms.
@@ -43,6 +43,7 @@ spec:
           headers:
             Content-Type: application/json
           body:
+            model: gpt-4o
             messages:
               - role: user
                 content: "The previous stage's response is as follows. Please provide a simpler explanation: \n${stage-1.text}"
diff --git a/src/worker/executors/api_executor.py b/src/worker/executors/api_executor.py
index 436200cf5..dbe2b630b 100644
--- a/src/worker/executors/api_executor.py
+++ b/src/worker/executors/api_executor.py
@@ -17,17 +17,14 @@
 # Cache key: (base_url, timeout_seconds, verify_tls, follow_redirects)
 _ClientKey = tuple[str, float, bool, bool]
 
-_DEFAULT_BASE_URL = "https://lum.id/llm"
-_DEFAULT_MODEL = "deepseek-v4-flash"
-
 
 class APIExecutor(Executor):
     """Performs a single HTTP request defined by task YAML.
 
-    Defaults to the lum.id/llm chat-completions endpoint and the deepseek model
-    when the spec omits them. The credential is the worker's own
-    ``NEBULA_API_TOKEN`` unless the spec supplies an ``Authorization`` header
-    directly. A call with no credential fails closed.
+    Defaults to the Nebula endpoint via ``NEBULA_API_BASE_URL`` and authenticates
+    with ``NEBULA_API_TOKEN``. ``spec.api.url`` overrides the endpoint and
+    ``spec.api.headers`` may supply an ``Authorization`` header directly; the
+    Nebula token is never sent to a custom endpoint.
     """
 
     name = "api"
@@ -102,8 +99,12 @@ def run(self, task: ExecutorTask, out_dir: Path) -> APIResult:
             raise ExecutionError("spec.api must be a mapping")
 
         url = api_cfg.get("url")
+        custom_url = url is not None
         if url is None:
-            url = _DEFAULT_BASE_URL + "/v1/chat/completions"
+            url = os.getenv("NEBULA_API_BASE_URL")
+            if not url:
+                raise ExecutionError("spec.api.url or NEBULA_API_BASE_URL is required")
+            url = url.rstrip("/") + "/v1/chat/completions"
 
         method = str(api_cfg.get("method", "POST")).upper()
         headers = api_cfg.get("headers", {})
@@ -111,13 +112,14 @@ def run(self, task: ExecutorTask, out_dir: Path) -> APIResult:
             raise ExecutionError("spec.api.headers must be a mapping")
 
         if not any(k.lower() == "authorization" for k in headers):
-            token = os.getenv("NEBULA_API_TOKEN")
-            if not token:
-                raise ExecutionError(
-                    "no credential configured: set an Authorization header or "
-                    "NEBULA_API_TOKEN"
-                )
-            headers["Authorization"] = f"Bearer {token}"
+            if not custom_url:
+                token = os.getenv("NEBULA_API_TOKEN")
+                if not token:
+                    raise ExecutionError(
+                        "no credential configured: set an Authorization header or "
+                        "NEBULA_API_TOKEN"
+                    )
+                headers["Authorization"] = f"Bearer {token}"
 
         params = api_cfg.get("params")
         if params is not None and not isinstance(params, dict):
@@ -142,15 +144,9 @@ def run(self, task: ExecutorTask, out_dir: Path) -> APIResult:
 
         request_kwargs: dict[str, Any] = {}
         if json_payload is not None:
-            if isinstance(json_payload, dict) and "model" not in json_payload:
-                json_payload = {**json_payload, "model": _DEFAULT_MODEL}
             request_kwargs["json"] = json_payload
         elif body is not None:
-            if isinstance(body, dict):
-                if "model" not in body:
-                    body = {**body, "model": _DEFAULT_MODEL}
-                request_kwargs["json"] = body
-            elif isinstance(body, list):
+            if isinstance(body, (dict, list)):
                 request_kwargs["json"] = body
             else:
                 request_kwargs["content"] = body
diff --git a/tests/worker/test_api_executor.py b/tests/worker/test_api_executor.py
index 12861033a..fe355d816 100644
--- a/tests/worker/test_api_executor.py
+++ b/tests/worker/test_api_executor.py
@@ -1,4 +1,4 @@
-"""Tests for the API executor defaults, credential handling, and model injection."""
+"""Tests for the API executor url override and Nebula credential handling."""
 
 from pathlib import Path
 from unittest.mock import patch
@@ -25,11 +25,8 @@ def _task_message(**spec_updates: object) -> WorkerTaskMessage:
             "spec": {
                 "taskType": "api",
                 "api": {
-                    "url": "https://api.example.com/v1/chat/completions",
                     "method": "POST",
-                    "body": {
-                        "messages": [{"role": "user", "content": "hi"}],
-                    },
+                    "body": {"messages": [{"role": "user", "content": "hi"}]},
                     **spec_updates,
                 },
             },
@@ -65,69 +62,62 @@ def _run(
         executor.run(task, Path("/tmp/out"))
 
 
-class TestDefaults:
-    def test_default_url_is_lum_id_llm(self, monkeypatch: pytest.MonkeyPatch) -> None:
-        monkeypatch.setenv("NEBULA_API_TOKEN", "token")
-        task = _task_message(url=None)
-        transport = _RecordingTransport()
-        _run(APIExecutor.__new__(APIExecutor), task, transport)
-        assert transport.request is not None
-        assert transport.request.url == "https://lum.id/llm/v1/chat/completions"
-
-    def test_default_model_injected_when_body_has_no_model(
+class TestNebulaPath:
+    def test_no_url_no_header_uses_nebula_url_and_token(
         self, monkeypatch: pytest.MonkeyPatch
     ) -> None:
-        monkeypatch.setenv("NEBULA_API_TOKEN", "token")
+        monkeypatch.setenv("NEBULA_API_BASE_URL", "https://nebula.example.com")
+        monkeypatch.setenv("NEBULA_API_TOKEN", "nebula-token")
         task = _task_message()
         transport = _RecordingTransport()
         _run(APIExecutor.__new__(APIExecutor), task, transport)
         assert transport.request is not None
-        sent = _sent_json(transport.request)
-        assert sent["model"] == "deepseek-v4-flash"
+        assert transport.request.url == "https://nebula.example.com/v1/chat/completions"
+        assert transport.request.headers["Authorization"] == "Bearer nebula-token"
 
-    def test_explicit_model_left_untouched(
+    def test_no_url_with_header_preserves_header_and_skips_token(
         self, monkeypatch: pytest.MonkeyPatch
     ) -> None:
-        monkeypatch.setenv("NEBULA_API_TOKEN", "token")
-        task = _task_message(body={"model": "gpt-4o", "messages": []})
+        monkeypatch.setenv("NEBULA_API_BASE_URL", "https://nebula.example.com")
+        monkeypatch.setenv("NEBULA_API_TOKEN", "nebula-token")
+        task = _task_message(headers={"Authorization": "Bearer custom"})
         transport = _RecordingTransport()
         _run(APIExecutor.__new__(APIExecutor), task, transport)
         assert transport.request is not None
-        sent = _sent_json(transport.request)
-        assert sent["model"] == "gpt-4o"
+        assert transport.request.url == "https://nebula.example.com/v1/chat/completions"
+        assert transport.request.headers["Authorization"] == "Bearer custom"
+
+    def test_neither_url_nor_base_url_raises(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.delenv("NEBULA_API_BASE_URL", raising=False)
+        task = _task_message()
+        with pytest.raises(ExecutionError, match="spec.api.url or NEBULA_API_BASE_URL"):
+            _run(APIExecutor.__new__(APIExecutor), task, _RecordingTransport())
 
 
-class TestCredential:
-    def test_explicit_authorization_header_left_untouched(
+class TestCustomUrl:
+    def test_custom_url_no_header_does_not_inject_token(
         self, monkeypatch: pytest.MonkeyPatch
     ) -> None:
-        monkeypatch.setenv("NEBULA_API_TOKEN", "worker-token")
-        task = _task_message(headers={"Authorization": "Bearer explicit"})
+        monkeypatch.setenv("NEBULA_API_TOKEN", "nebula-token")
+        task = _task_message(url="https://custom.example.com/v1/chat/completions")
         transport = _RecordingTransport()
         _run(APIExecutor.__new__(APIExecutor), task, transport)
         assert transport.request is not None
-        assert transport.request.headers["Authorization"] == "Bearer explicit"
+        assert transport.request.url == "https://custom.example.com/v1/chat/completions"
+        assert "Authorization" not in transport.request.headers
 
-    def test_nebula_token_used_when_no_header(
+    def test_custom_url_with_header_preserves_header(
         self, monkeypatch: pytest.MonkeyPatch
     ) -> None:
-        monkeypatch.setenv("NEBULA_API_TOKEN", "worker-token")
-        task = _task_message()
+        monkeypatch.setenv("NEBULA_API_TOKEN", "nebula-token")
+        task = _task_message(
+            url="https://custom.example.com/v1/chat/completions",
+            headers={"Authorization": "Bearer custom"},
+        )
         transport = _RecordingTransport()
         _run(APIExecutor.__new__(APIExecutor), task, transport)
         assert transport.request is not None
-        assert transport.request.headers["Authorization"] == "Bearer worker-token"
-
-    def test_missing_credential_fails_closed(
-        self, monkeypatch: pytest.MonkeyPatch
-    ) -> None:
-        monkeypatch.delenv("NEBULA_API_TOKEN", raising=False)
-        task = _task_message()
-        with pytest.raises(ExecutionError, match="NEBULA_API_TOKEN"):
-            _run(APIExecutor.__new__(APIExecutor), task, _RecordingTransport())
-
-
-def _sent_json(request: httpx.Request) -> dict:
-    import json
-
-    return json.loads(request.content.decode("utf-8"))
+        assert transport.request.url == "https://custom.example.com/v1/chat/completions"
+        assert transport.request.headers["Authorization"] == "Bearer custom"

From 61f118d401f0296f65810b543a97a9255167ee5a Mon Sep 17 00:00:00 2001
From: Zhengyuan Su <su.zhengyuan@u.nus.edu>
Date: Wed, 16 Sep 2026 17:05:35 +0700
Subject: [PATCH 4/8] fix: a custom api url must carry its own credential

Moving the Nebula token injection inside the "no custom url" branch left
headers read before it was assigned, so every Nebula-path call raised
UnboundLocalError. Restore the ordering and gate on a custom_url flag.

A caller-supplied spec.api.url now requires its own Authorization header and
fails closed without one, rather than calling the endpoint anonymously. The
Nebula token is never sent to an endpoint the caller chose.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Signed-off-by: Zhengyuan Su <su.zhengyuan@u.nus.edu>
---
 src/worker/executors/api_executor.py | 30 ++++++++++++++++++----------
 tests/worker/test_api_executor.py    | 14 ++++++++-----
 2 files changed, 28 insertions(+), 16 deletions(-)

diff --git a/src/worker/executors/api_executor.py b/src/worker/executors/api_executor.py
index dbe2b630b..907216d98 100644
--- a/src/worker/executors/api_executor.py
+++ b/src/worker/executors/api_executor.py
@@ -23,8 +23,9 @@ class APIExecutor(Executor):
 
     Defaults to the Nebula endpoint via ``NEBULA_API_BASE_URL`` and authenticates
     with ``NEBULA_API_TOKEN``. ``spec.api.url`` overrides the endpoint and
-    ``spec.api.headers`` may supply an ``Authorization`` header directly; the
-    Nebula token is never sent to a custom endpoint.
+    ``spec.api.headers`` may supply an ``Authorization`` header directly. A
+    custom ``spec.api.url`` requires its own credential: the Nebula token is
+    never sent to an endpoint the caller chose.
     """
 
     name = "api"
@@ -111,15 +112,22 @@ def run(self, task: ExecutorTask, out_dir: Path) -> APIResult:
         if not isinstance(headers, dict):
             raise ExecutionError("spec.api.headers must be a mapping")
 
-        if not any(k.lower() == "authorization" for k in headers):
-            if not custom_url:
-                token = os.getenv("NEBULA_API_TOKEN")
-                if not token:
-                    raise ExecutionError(
-                        "no credential configured: set an Authorization header or "
-                        "NEBULA_API_TOKEN"
-                    )
-                headers["Authorization"] = f"Bearer {token}"
+        has_credential = any(k.lower() == "authorization" for k in headers)
+        if custom_url:
+            if not has_credential:
+                raise ExecutionError(
+                    "spec.api.url names a custom endpoint but no credential was "
+                    "supplied; set an Authorization header. The Nebula token is "
+                    "never sent to an endpoint the caller chose."
+                )
+        elif not has_credential:
+            token = os.getenv("NEBULA_API_TOKEN")
+            if not token:
+                raise ExecutionError(
+                    "no credential configured: set an Authorization header or "
+                    "NEBULA_API_TOKEN"
+                )
+            headers["Authorization"] = f"Bearer {token}"
 
         params = api_cfg.get("params")
         if params is not None and not isinstance(params, dict):
diff --git a/tests/worker/test_api_executor.py b/tests/worker/test_api_executor.py
index fe355d816..4dafbcfc9 100644
--- a/tests/worker/test_api_executor.py
+++ b/tests/worker/test_api_executor.py
@@ -97,16 +97,20 @@ def test_neither_url_nor_base_url_raises(
 
 
 class TestCustomUrl:
-    def test_custom_url_no_header_does_not_inject_token(
+    def test_custom_url_without_credential_raises(
         self, monkeypatch: pytest.MonkeyPatch
     ) -> None:
+        """A custom endpoint must carry its own credential.
+
+        The Nebula token is available here, so the failure proves it is withheld
+        rather than merely absent: a caller-chosen endpoint never receives it.
+        """
         monkeypatch.setenv("NEBULA_API_TOKEN", "nebula-token")
         task = _task_message(url="https://custom.example.com/v1/chat/completions")
         transport = _RecordingTransport()
-        _run(APIExecutor.__new__(APIExecutor), task, transport)
-        assert transport.request is not None
-        assert transport.request.url == "https://custom.example.com/v1/chat/completions"
-        assert "Authorization" not in transport.request.headers
+        with pytest.raises(ExecutionError, match="custom endpoint"):
+            _run(APIExecutor.__new__(APIExecutor), task, transport)
+        assert transport.request is None
 
     def test_custom_url_with_header_preserves_header(
         self, monkeypatch: pytest.MonkeyPatch

From eb09f859eae6e5a0d8ab22bb68811c6322fb5cac Mon Sep 17 00:00:00 2001
From: Zhengyuan Su <su.zhengyuan@u.nus.edu>
Date: Wed, 16 Sep 2026 17:55:44 +0700
Subject: [PATCH 5/8] fix(security): never persist an API credential in
 plaintext

Signed-off-by: Zhengyuan Su <su.zhengyuan@u.nus.edu>
---
 src/server/dispatcher/base.py |  28 ++++++
 src/server/task/models.py     |  19 +++-
 src/server/task/redact.py     |  84 ++++++++++++++++++
 tests/server/test_redact.py   | 163 ++++++++++++++++++++++++++++++++++
 4 files changed, 293 insertions(+), 1 deletion(-)
 create mode 100644 src/server/task/redact.py
 create mode 100644 tests/server/test_redact.py

diff --git a/src/server/dispatcher/base.py b/src/server/dispatcher/base.py
index b6db7f21f..f650d2968 100644
--- a/src/server/dispatcher/base.py
+++ b/src/server/dispatcher/base.py
@@ -36,6 +36,7 @@
 from ..services.metrics import MetricsRecorder
 from ..task.metadata import extract_model_dataset_names
 from ..task.models import TaskRecord, TaskStatus
+from ..task.redact import REDACTED
 from ..task.runtime import TaskRuntime
 from ..utils.time import now_iso
 from .worker_selector import DEFAULT_WORKER_SELECTION, select_worker
@@ -393,6 +394,23 @@ def dispatch_once(self, task_id: str) -> bool:
             )
             return True
 
+        # A redacted credential means the task was rehydrated from a dump that
+        # could not retain the secret; it cannot authenticate, so fail it rather
+        # than dispatch with a placeholder bearer token.
+        if self._has_redacted_credential(rendered_task.spec):
+            self._runtime.release_merge(task_id)
+            self.fail_task(
+                task_id,
+                "credential_not_retained",
+                payload={
+                    "error": (
+                        "the API credential was not retained across the server "
+                        "restart; resubmit the workflow with the credential"
+                    )
+                },
+            )
+            return True
+
         # Conditional execution: skip dispatch if condition not met
         if self._evaluate_condition_skip(task_id, rendered_task, record):
             return True
@@ -1055,6 +1073,16 @@ def _collect_upstream_results(
             results[name] = envelope.result
         return results
 
+    def _has_redacted_credential(self, spec: TaskSpecStrict) -> bool:
+        """Whether an api spec carries a redacted credential placeholder."""
+        api = getattr(spec, "api", None)
+        if not isinstance(api, dict):
+            return False
+        headers = api.get("headers")
+        if not isinstance(headers, dict):
+            return False
+        return any(value == REDACTED for value in headers.values())
+
     def _resolve_upstream_task_ids(
         self, record: TaskRecord, spec: TaskSpecStrict
     ) -> dict[str, str] | None:
diff --git a/src/server/task/models.py b/src/server/task/models.py
index 0c64a20d8..722f835af 100644
--- a/src/server/task/models.py
+++ b/src/server/task/models.py
@@ -1,12 +1,19 @@
 import time
 from typing import Any
 
-from pydantic import BaseModel, Field, computed_field
+from pydantic import (
+    BaseModel,
+    Field,
+    SerializerFunctionWrapHandler,
+    computed_field,
+    model_serializer,
+)
 
 from shared.tasks import TaskEnvelopeTemplate
 from shared.tasks.worker_message import HardwareUsage
 
 from ..utils.time import now_iso
+from .redact import redact_api, redact_raw_yaml
 
 TRAINING_TASK_TYPES = {
     "sft",
@@ -161,6 +168,16 @@ def last_failed_worker(self) -> str | None:
         """The most recent worker to have failed this task."""
         return self.failed_workers[-1] if self.failed_workers else None
 
+    @model_serializer(mode="wrap")
+    def _serialize(self, handler: SerializerFunctionWrapHandler) -> dict[str, Any]:
+        data = handler(self)
+        data["raw_yaml"] = redact_raw_yaml(self.raw_yaml)
+        spec = self.task.spec
+        api = getattr(spec, "api", None)
+        if isinstance(api, dict):
+            data["task"]["spec"]["api"] = redact_api(api)
+        return data
+
 
 class TaskInfo(TaskRecord):
     depends_on: list[str] = Field(description="Dependency task IDs.")
diff --git a/src/server/task/redact.py b/src/server/task/redact.py
new file mode 100644
index 000000000..d2da3c924
--- /dev/null
+++ b/src/server/task/redact.py
@@ -0,0 +1,84 @@
+"""Redaction of API credentials before a task record is serialized.
+
+The in-memory ``TaskRecord`` keeps the real credential so dispatch works; the
+serializer applies this module so that every dump to Redis is redacted.
+
+``raw_yaml`` is re-emitted via ``yaml.safe_dump``, which does not preserve
+comments, key order or original formatting. That is an accepted cost: the field
+is a stored record and is never re-parsed, so losing formatting is fine. If the
+YAML cannot be parsed, the whole field is redacted rather than storing text
+that might contain a key.
+"""
+
+from typing import Any
+
+import yaml
+
+REDACTED = "[REDACTED]"
+
+# Whole-key matches; a bare substring test would over-redact (e.g. "monkey").
+_SENSITIVE_KEYS = frozenset(
+    {
+        "authorization",
+        "token",
+        "api-key",
+        "api_key",
+        "apikey",
+        "secret",
+        "access_token",
+        "bearer",
+        "x-api-key",
+    }
+)
+_SENSITIVE_SUFFIXES = ("_key", "-key", "_token", "-token")
+
+
+def _is_sensitive_key(name: str) -> bool:
+    lowered = name.lower()
+    if lowered in _SENSITIVE_KEYS:
+        return True
+    return any(lowered.endswith(suffix) for suffix in _SENSITIVE_SUFFIXES)
+
+
+def _redact_value(value: Any) -> Any:
+    """Recursively redact credential values by key name at any depth."""
+    if isinstance(value, dict):
+        return {
+            key: (REDACTED if _is_sensitive_key(str(key)) else _redact_value(val))
+            for key, val in value.items()
+        }
+    if isinstance(value, list):
+        return [_redact_value(item) for item in value]
+    return value
+
+
+def redact_api(api: dict[str, Any] | None) -> dict[str, Any] | None:
+    """Return a copy of an api spec with credential values replaced.
+
+    Only the five credential-bearing locations are touched; the rest of the
+    spec is returned unchanged. The original mapping is never mutated.
+    """
+    if not isinstance(api, dict):
+        return api
+    redacted = dict(api)
+    for field in ("headers", "params", "body", "json", "data"):
+        value = redacted.get(field)
+        if isinstance(value, (dict, list)):
+            redacted[field] = _redact_value(value)
+    return redacted
+
+
+def redact_raw_yaml(raw_yaml: str) -> str:
+    """Redact credential values from the original workflow YAML text.
+
+    The YAML is parsed and redacted with the same recursive key rule used for
+    the parsed spec, then re-emitted. If parsing fails, the whole field is
+    replaced with a marker rather than storing un-analysed text.
+    """
+    try:
+        tree = yaml.safe_load(raw_yaml)
+    except yaml.YAMLError:
+        return REDACTED
+    if tree is None:
+        return raw_yaml
+    return yaml.safe_dump(_redact_value(tree), sort_keys=False)
diff --git a/tests/server/test_redact.py b/tests/server/test_redact.py
new file mode 100644
index 000000000..83fbba0b1
--- /dev/null
+++ b/tests/server/test_redact.py
@@ -0,0 +1,163 @@
+"""Tests for API credential redaction at serialization time."""
+
+import json
+
+from server.task.models import TaskRecord
+from server.task.redact import REDACTED, _is_sensitive_key, redact_api, redact_raw_yaml
+from shared.tasks import TaskEnvelopeTemplate
+from shared.tasks.specs import ApiSpecTemplate
+
+
+def _api_task(api: dict) -> TaskEnvelopeTemplate:
+    return TaskEnvelopeTemplate.model_validate(
+        {
+            "apiVersion": "mloc/v1",
+            "kind": "Task",
+            "metadata": {"name": "t"},
+            "spec": {"taskType": "api", "api": api},
+        }
+    )
+
+
+def _record(api: dict, raw_yaml: str = "") -> TaskRecord:
+    return TaskRecord(
+        task_id="tsk-1",
+        workflow_id="wfl-1",
+        owner_id="owner",
+        raw_yaml=raw_yaml,
+        task=_api_task(api),
+    )
+
+
+class TestSensitiveKey:
+    def test_sensitive_keys_match(self) -> None:
+        for key in (
+            "Authorization",
+            "authorization",
+            "token",
+            "api-key",
+            "api_key",
+            "apikey",
+            "secret",
+            "access_token",
+            "bearer",
+            "x-api-key",
+            "my_token",
+            "my_key",
+        ):
+            assert _is_sensitive_key(key), key
+
+    def test_innocent_keys_do_not_match(self) -> None:
+        for key in ("monkey", "turkey", "keyword", "keys", "model", "messages"):
+            assert not _is_sensitive_key(key), key
+
+
+class TestRedactApi:
+    def test_headers_redacted(self) -> None:
+        out = redact_api({"headers": {"Authorization": "Bearer SECRET"}})
+        assert out is not None
+        assert out["headers"]["Authorization"] == REDACTED
+
+    def test_nested_in_dict_redacted(self) -> None:
+        out = redact_api({"json": {"auth": {"token": "SECRET-NESTED"}}})
+        assert out is not None
+        assert out["json"]["auth"]["token"] == REDACTED
+
+    def test_nested_in_list_redacted(self) -> None:
+        out = redact_api({"json": [{"token": "SECRET"}]})
+        assert out is not None
+        assert out["json"][0]["token"] == REDACTED
+
+    def test_all_five_locations_redacted(self) -> None:
+        api = {
+            "headers": {"Authorization": "Bearer H"},
+            "params": {"api_key": "P"},
+            "body": {"secret": "B"},
+            "json": {"token": "J"},
+            "data": {"access_token": "D"},
+        }
+        out = redact_api(api)
+        assert out is not None
+        for field in ("headers", "params", "body", "json", "data"):
+            assert list(out[field].values()) == [REDACTED], field
+
+    def test_innocent_values_preserved(self) -> None:
+        out = redact_api({"json": {"model": "gpt", "monkey": "x"}})
+        assert out is not None
+        assert out["json"]["model"] == "gpt"
+        assert out["json"]["monkey"] == "x"
+
+    def test_original_not_mutated(self) -> None:
+        api = {"headers": {"Authorization": "Bearer SECRET"}}
+        redact_api(api)
+        assert api["headers"]["Authorization"] == "Bearer SECRET"
+
+
+class TestRedactRawYaml:
+    def test_plain_scalar_redacted(self) -> None:
+        out = redact_raw_yaml("Authorization: Bearer SECRET\n")
+        assert "SECRET" not in out
+        assert REDACTED in out
+
+    def test_block_scalar_redacted(self) -> None:
+        out = redact_raw_yaml("Authorization: |\n  Bearer SECRET\n")
+        assert "SECRET" not in out
+        assert REDACTED in out
+
+    def test_folded_scalar_redacted(self) -> None:
+        out = redact_raw_yaml("Authorization: >-\n  Bearer SECRET\n")
+        assert "SECRET" not in out
+        assert REDACTED in out
+
+    def test_nested_yaml_redacted(self) -> None:
+        out = redact_raw_yaml("api:\n  json:\n    token: SECRET\n")
+        assert "SECRET" not in out
+
+    def test_innocent_yaml_preserved(self) -> None:
+        out = redact_raw_yaml("model: gpt-4o\nmonkey: x\n")
+        assert "gpt-4o" in out
+        assert "monkey: x" in out
+
+    def test_unparseable_yaml_fails_closed(self) -> None:
+        assert redact_raw_yaml("a: [unclosed") == REDACTED
+
+
+class TestTaskRecordSerializer:
+    def test_dump_redacts_all_five_locations(self) -> None:
+        api = {
+            "headers": {"Authorization": "Bearer H"},
+            "params": {"api_key": "P"},
+            "body": {"secret": "B"},
+            "json": {"token": "J"},
+            "data": {"access_token": "D"},
+        }
+        rec = _record(api)
+        dumped = rec.model_dump()
+        dumped_api = dumped["task"]["spec"]["api"]
+        for field in ("headers", "params", "body", "json", "data"):
+            assert list(dumped_api[field].values()) == [REDACTED], field
+
+    def test_dump_json_redacts(self) -> None:
+        rec = _record({"headers": {"Authorization": "Bearer SECRET"}})
+        dumped = json.loads(rec.model_dump_json())
+        assert dumped["task"]["spec"]["api"]["headers"]["Authorization"] == REDACTED
+
+    def test_raw_yaml_redacted_in_dump(self) -> None:
+        rec = _record(
+            {"headers": {"Authorization": "Bearer SECRET"}},
+            raw_yaml="api:\n  headers:\n    Authorization: Bearer SECRET\n",
+        )
+        dumped = rec.model_dump()
+        assert "SECRET" not in dumped["raw_yaml"]
+
+    def test_in_memory_keeps_real_credential(self) -> None:
+        rec = _record({"headers": {"Authorization": "Bearer SECRET"}})
+        assert isinstance(rec.task.spec, ApiSpecTemplate)
+        assert rec.task.spec.api is not None
+        assert rec.task.spec.api["headers"]["Authorization"] == "Bearer SECRET"
+
+    def test_no_credential_unchanged(self) -> None:
+        api = {"url": "http://x", "json": {"model": "gpt"}}
+        rec = _record(api)
+        dumped = rec.model_dump()
+        assert dumped["task"]["spec"]["api"] == api

From b05b6d1973c3777ace29b6204c7693d2a3b842f3 Mon Sep 17 00:00:00 2001
From: Zhengyuan Su <su.zhengyuan@u.nus.edu>
Date: Wed, 16 Sep 2026 18:15:39 +0700
Subject: [PATCH 6/8] refactor: narrow the redacted-credential check to
 ApiSpecStrict

Signed-off-by: Zhengyuan Su <su.zhengyuan@u.nus.edu>
---
 src/server/dispatcher/base.py | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/src/server/dispatcher/base.py b/src/server/dispatcher/base.py
index f650d2968..797f8dd81 100644
--- a/src/server/dispatcher/base.py
+++ b/src/server/dispatcher/base.py
@@ -25,6 +25,7 @@
 )
 from shared.tasks.placeholders import PLACEHOLDER_PATTERN
 from shared.tasks.specs import (
+    ApiSpecStrict,
     ConditionSpec,
     SSHSpecStrict,
     SSHSpecTemplate,
@@ -394,9 +395,6 @@ def dispatch_once(self, task_id: str) -> bool:
             )
             return True
 
-        # A redacted credential means the task was rehydrated from a dump that
-        # could not retain the secret; it cannot authenticate, so fail it rather
-        # than dispatch with a placeholder bearer token.
         if self._has_redacted_credential(rendered_task.spec):
             self._runtime.release_merge(task_id)
             self.fail_task(
@@ -1075,8 +1073,10 @@ def _collect_upstream_results(
 
     def _has_redacted_credential(self, spec: TaskSpecStrict) -> bool:
         """Whether an api spec carries a redacted credential placeholder."""
-        api = getattr(spec, "api", None)
-        if not isinstance(api, dict):
+        if not isinstance(spec, ApiSpecStrict):
+            return False
+        api = spec.api
+        if api is None:
             return False
         headers = api.get("headers")
         if not isinstance(headers, dict):

From 7ac065cf4dab62121e1859fc5c05f708524e9f99 Mon Sep 17 00:00:00 2001
From: Zhengyuan Su <su.zhengyuan@u.nus.edu>
Date: Fri, 18 Sep 2026 09:32:57 +0700
Subject: [PATCH 7/8] refactor(security): share the redaction helpers and let
 an unauthenticated endpoint serve

Addresses the PR review comments.

redact.py moves to src/shared/utils so the worker executor can reuse its key
detection instead of keeping a second notion of what counts as a credential.
_is_sensitive_key becomes public for that reuse, and credential detection is no
longer Authorization-only: X-API-Key and the other sensitive header names now
satisfy it.

The redacted-credential check in the dispatcher covered only headers while
redaction itself covered five locations, so a credential redacted in params,
body, json or data was not detected. contains_redacted walks all of them.

TaskRecord.raw_yaml becomes source: runtime.register stores the payload
verbatim regardless of format, and redact_raw_yaml re-emits it via
yaml.safe_dump, so the field stops being raw once n8n JSON is converted and
comments are stripped. The serializer memoises both redactions rather than
recomputing them on every dump.

A custom spec.api.url without its own credential now calls the endpoint
anonymously instead of failing closed, so an unauthorized serving endpoint
stays usable. The Nebula token is still never sent to a caller-chosen
endpoint: the token injection lives inside the no-custom-url branch, and the
custom-url tests keep NEBULA_API_TOKEN set so an absent Authorization header
proves the token is withheld rather than merely unavailable.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Signed-off-by: Zhengyuan Su <su.zhengyuan@u.nus.edu>
---
 sdk/src/flowmesh/models/tasks.py              |   2 +-
 src/server/dispatcher/base.py                 |  17 ++-
 src/server/task/models.py                     |  32 ++++--
 src/server/task/runtime.py                    |   4 +-
 src/{server/task => shared/utils}/redact.py   |  26 +++--
 src/worker/executors/api_executor.py          |  42 +++-----
 tests/sdk/test_models.py                      |   2 +-
 tests/sdk/test_resource_helpers.py            |   2 +-
 .../dispatcher/test_redacted_credential.py    | 101 ++++++++++++++++++
 tests/server/task/test_ssh_result_mounting.py |  26 ++---
 tests/server/test_hooks_wiring.py             |   2 +-
 tests/server/test_redact.py                   |  35 ++++--
 tests/worker/test_api_executor.py             |  51 +++++++--
 13 files changed, 264 insertions(+), 78 deletions(-)
 rename src/{server/task => shared/utils}/redact.py (71%)
 create mode 100644 tests/server/dispatcher/test_redacted_credential.py

diff --git a/sdk/src/flowmesh/models/tasks.py b/sdk/src/flowmesh/models/tasks.py
index 870361335..13649490e 100644
--- a/sdk/src/flowmesh/models/tasks.py
+++ b/sdk/src/flowmesh/models/tasks.py
@@ -27,7 +27,7 @@ class TaskInfo(BaseModel):
     owner_id: str
     org_id: str
     supplier_id: str
-    raw_yaml: str
+    source: str
     task: dict[str, Any]
     status: TaskStatus
     task_type: str | None = None
diff --git a/src/server/dispatcher/base.py b/src/server/dispatcher/base.py
index 797f8dd81..95d52b951 100644
--- a/src/server/dispatcher/base.py
+++ b/src/server/dispatcher/base.py
@@ -31,13 +31,13 @@
     SSHSpecTemplate,
 )
 from shared.tasks.worker_message import WorkerStatus, WorkerTaskMessage
+from shared.utils.redact import contains_redacted
 
 from ..clients.redis import REDIS_CONN_ERRORS
 from ..registries.worker import Worker, WorkerRegistry
 from ..services.metrics import MetricsRecorder
 from ..task.metadata import extract_model_dataset_names
 from ..task.models import TaskRecord, TaskStatus
-from ..task.redact import REDACTED
 from ..task.runtime import TaskRuntime
 from ..utils.time import now_iso
 from .worker_selector import DEFAULT_WORKER_SELECTION, select_worker
@@ -1078,10 +1078,17 @@ def _has_redacted_credential(self, spec: TaskSpecStrict) -> bool:
         api = spec.api
         if api is None:
             return False
-        headers = api.get("headers")
-        if not isinstance(headers, dict):
-            return False
-        return any(value == REDACTED for value in headers.values())
+        for field in ("headers", "params", "body", "json", "data"):
+            value = api.get(field)
+            if isinstance(value, dict):
+                candidates: list[Any] = list(value.values())
+            elif isinstance(value, list):
+                candidates = value
+            else:
+                continue
+            if any(contains_redacted(item) for item in candidates):
+                return True
+        return False
 
     def _resolve_upstream_task_ids(
         self, record: TaskRecord, spec: TaskSpecStrict
diff --git a/src/server/task/models.py b/src/server/task/models.py
index 722f835af..4445af8f4 100644
--- a/src/server/task/models.py
+++ b/src/server/task/models.py
@@ -4,16 +4,18 @@
 from pydantic import (
     BaseModel,
     Field,
+    PrivateAttr,
     SerializerFunctionWrapHandler,
     computed_field,
     model_serializer,
 )
 
 from shared.tasks import TaskEnvelopeTemplate
+from shared.tasks.specs import ApiSpecStrict, ApiSpecTemplate
 from shared.tasks.worker_message import HardwareUsage
+from shared.utils.redact import redact_api, redact_raw_yaml
 
 from ..utils.time import now_iso
-from .redact import redact_api, redact_raw_yaml
 
 TRAINING_TASK_TYPES = {
     "sft",
@@ -81,7 +83,7 @@ class TaskRecord(BaseModel):
     owner_id: str = Field(description="Owner principal identifier.")
     org_id: str = Field(default="", description="Owner organization identifier.")
     supplier_id: str = Field(default="", description="Supplier identifier.")
-    raw_yaml: str = Field(description="Original workflow YAML.")
+    source: str = Field(description="Original workflow source (YAML or JSON).")
     task: TaskEnvelopeTemplate = Field(description="Task template.")
     status: str = Field(default=TaskStatus.PENDING, description="Task status.")
     task_type: str | None = Field(default=None, description="Task type.")
@@ -168,14 +170,30 @@ def last_failed_worker(self) -> str | None:
         """The most recent worker to have failed this task."""
         return self.failed_workers[-1] if self.failed_workers else None
 
+    _redacted_source: str | None = PrivateAttr(default=None)
+    _redacted_api: dict[str, Any] | None = PrivateAttr(default=None)
+
+    def _redact_source(self) -> str:
+        if self._redacted_source is None:
+            self._redacted_source = redact_raw_yaml(self.source)
+        return self._redacted_source
+
+    def _redact_api(self) -> dict[str, Any] | None:
+        if self._redacted_api is None:
+            spec = self.task.spec
+            if isinstance(spec, (ApiSpecStrict, ApiSpecTemplate)):
+                self._redacted_api = redact_api(spec.api)
+            else:
+                self._redacted_api = None
+        return self._redacted_api
+
     @model_serializer(mode="wrap")
     def _serialize(self, handler: SerializerFunctionWrapHandler) -> dict[str, Any]:
         data = handler(self)
-        data["raw_yaml"] = redact_raw_yaml(self.raw_yaml)
-        spec = self.task.spec
-        api = getattr(spec, "api", None)
-        if isinstance(api, dict):
-            data["task"]["spec"]["api"] = redact_api(api)
+        data["source"] = self._redact_source()
+        redacted_api = self._redact_api()
+        if redacted_api is not None:
+            data["task"]["spec"]["api"] = redacted_api
         return data
 
 
diff --git a/src/server/task/runtime.py b/src/server/task/runtime.py
index 1547c8cc5..b10e7730a 100644
--- a/src/server/task/runtime.py
+++ b/src/server/task/runtime.py
@@ -109,7 +109,7 @@ async def register(
     ) -> tuple[str, list[TaskParsingResult]]:
         parsed_workflow = parse_workflow(payload, format)
         specs = parsed_workflow.tasks
-        yaml_text = payload
+        source_text = payload
         results: list[TaskParsingResult] = []
         workflow_id = new_workflow_id()
         task_records: list[TaskRecord] = []
@@ -155,7 +155,7 @@ async def register(
                     task_id=task_id,
                     workflow_id=workflow_id,
                     owner_id=owner_id,
-                    raw_yaml=yaml_text,
+                    source=source_text,
                     task=task,
                     local_name=entry.local_name,
                     graph_node_name=entry.graph_node_name,
diff --git a/src/server/task/redact.py b/src/shared/utils/redact.py
similarity index 71%
rename from src/server/task/redact.py
rename to src/shared/utils/redact.py
index d2da3c924..2c0412a1f 100644
--- a/src/server/task/redact.py
+++ b/src/shared/utils/redact.py
@@ -3,11 +3,11 @@
 The in-memory ``TaskRecord`` keeps the real credential so dispatch works; the
 serializer applies this module so that every dump to Redis is redacted.
 
-``raw_yaml`` is re-emitted via ``yaml.safe_dump``, which does not preserve
-comments, key order or original formatting. That is an accepted cost: the field
-is a stored record and is never re-parsed, so losing formatting is fine. If the
-YAML cannot be parsed, the whole field is redacted rather than storing text
-that might contain a key.
+The ``source`` field is re-emitted via ``yaml.safe_dump``, which does not
+preserve comments, key order or original formatting. That is an accepted cost:
+the field is a stored record and is never re-parsed, so losing formatting is
+fine. If the YAML cannot be parsed, the whole field is redacted rather than
+storing text that might contain a key.
 """
 
 from typing import Any
@@ -33,7 +33,8 @@
 _SENSITIVE_SUFFIXES = ("_key", "-key", "_token", "-token")
 
 
-def _is_sensitive_key(name: str) -> bool:
+def is_sensitive_key(name: str) -> bool:
+    """Whether a header/param name carries a credential value."""
     lowered = name.lower()
     if lowered in _SENSITIVE_KEYS:
         return True
@@ -44,7 +45,7 @@ def _redact_value(value: Any) -> Any:
     """Recursively redact credential values by key name at any depth."""
     if isinstance(value, dict):
         return {
-            key: (REDACTED if _is_sensitive_key(str(key)) else _redact_value(val))
+            key: (REDACTED if is_sensitive_key(str(key)) else _redact_value(val))
             for key, val in value.items()
         }
     if isinstance(value, list):
@@ -52,6 +53,17 @@ def _redact_value(value: Any) -> Any:
     return value
 
 
+def contains_redacted(value: Any) -> bool:
+    """Whether a value contains a redacted placeholder at any depth."""
+    if value == REDACTED:
+        return True
+    if isinstance(value, dict):
+        return any(contains_redacted(item) for item in value.values())
+    if isinstance(value, list):
+        return any(contains_redacted(item) for item in value)
+    return False
+
+
 def redact_api(api: dict[str, Any] | None) -> dict[str, Any] | None:
     """Return a copy of an api spec with credential values replaced.
 
diff --git a/src/worker/executors/api_executor.py b/src/worker/executors/api_executor.py
index 907216d98..8052d10b5 100644
--- a/src/worker/executors/api_executor.py
+++ b/src/worker/executors/api_executor.py
@@ -9,6 +9,7 @@
 from shared.schemas.result import APIResult
 from shared.tasks.specs import ApiSpecStrict
 from shared.tasks.task_type import TaskType
+from shared.utils.redact import is_sensitive_key
 
 from .base_executor import ExecutionError, Executor, ExecutorTask
 
@@ -23,9 +24,9 @@ class APIExecutor(Executor):
 
     Defaults to the Nebula endpoint via ``NEBULA_API_BASE_URL`` and authenticates
     with ``NEBULA_API_TOKEN``. ``spec.api.url`` overrides the endpoint and
-    ``spec.api.headers`` may supply an ``Authorization`` header directly. A
-    custom ``spec.api.url`` requires its own credential: the Nebula token is
-    never sent to an endpoint the caller chose.
+    ``spec.api.headers`` may supply a credential header (``Authorization``,
+    ``X-API-Key``, etc.) directly. A custom ``spec.api.url`` requires its own
+    credential: the Nebula token is never sent to an endpoint the caller chose.
     """
 
     name = "api"
@@ -100,34 +101,25 @@ def run(self, task: ExecutorTask, out_dir: Path) -> APIResult:
             raise ExecutionError("spec.api must be a mapping")
 
         url = api_cfg.get("url")
-        custom_url = url is not None
+        method = str(api_cfg.get("method", "POST")).upper()
+        headers = api_cfg.get("headers", {})
+        if not isinstance(headers, dict):
+            raise ExecutionError("spec.api.headers must be a mapping")
+
         if url is None:
             url = os.getenv("NEBULA_API_BASE_URL")
             if not url:
                 raise ExecutionError("spec.api.url or NEBULA_API_BASE_URL is required")
             url = url.rstrip("/") + "/v1/chat/completions"
 
-        method = str(api_cfg.get("method", "POST")).upper()
-        headers = api_cfg.get("headers", {})
-        if not isinstance(headers, dict):
-            raise ExecutionError("spec.api.headers must be a mapping")
-
-        has_credential = any(k.lower() == "authorization" for k in headers)
-        if custom_url:
-            if not has_credential:
-                raise ExecutionError(
-                    "spec.api.url names a custom endpoint but no credential was "
-                    "supplied; set an Authorization header. The Nebula token is "
-                    "never sent to an endpoint the caller chose."
-                )
-        elif not has_credential:
-            token = os.getenv("NEBULA_API_TOKEN")
-            if not token:
-                raise ExecutionError(
-                    "no credential configured: set an Authorization header or "
-                    "NEBULA_API_TOKEN"
-                )
-            headers["Authorization"] = f"Bearer {token}"
+            if not any(is_sensitive_key(k) for k in headers):
+                token = os.getenv("NEBULA_API_TOKEN")
+                if not token:
+                    raise ExecutionError(
+                        "no credential configured: set an Authorization header or "
+                        "NEBULA_API_TOKEN"
+                    )
+                headers["Authorization"] = f"Bearer {token}"
 
         params = api_cfg.get("params")
         if params is not None and not isinstance(params, dict):
diff --git a/tests/sdk/test_models.py b/tests/sdk/test_models.py
index 982aa3c1f..5daf04aa0 100644
--- a/tests/sdk/test_models.py
+++ b/tests/sdk/test_models.py
@@ -133,7 +133,7 @@
     task_id="t-abc",
     workflow_id="wf-abc",
     owner_id="usr-1",
-    raw_yaml="apiVersion: flowmesh/v1\nkind: Task",
+    source="apiVersion: flowmesh/v1\nkind: Task",
     task=_TASK_ENVELOPE,
     status="DONE",
     task_type="echo",
diff --git a/tests/sdk/test_resource_helpers.py b/tests/sdk/test_resource_helpers.py
index c3c2f2242..e2723416f 100644
--- a/tests/sdk/test_resource_helpers.py
+++ b/tests/sdk/test_resource_helpers.py
@@ -107,7 +107,7 @@ def _task_info(status: TaskStatus, latest_update: dict | None = None) -> TaskInf
         owner_id="u-1",
         org_id="o-1",
         supplier_id="s-1",
-        raw_yaml="kind: Task",
+        source="kind: Task",
         task={},
         status=status,
         submitted_at="2025-01-01T00:00:00Z",
diff --git a/tests/server/dispatcher/test_redacted_credential.py b/tests/server/dispatcher/test_redacted_credential.py
new file mode 100644
index 000000000..6616fd1d3
--- /dev/null
+++ b/tests/server/dispatcher/test_redacted_credential.py
@@ -0,0 +1,101 @@
+"""Tests for the dispatcher's redacted-credential detection.
+
+The dispatcher refuses to dispatch an api task whose credential was not
+retained across a server restart (it was redacted to ``[REDACTED]`` at
+persist time). Detection must cover every location that redaction touches,
+not just headers.
+"""
+
+from server.dispatcher import Dispatcher
+from shared.tasks import TaskEnvelopeStrict
+from shared.tasks.specs import ApiSpecStrict
+from shared.utils.redact import REDACTED, contains_redacted
+
+from .helpers import make_capturing_dispatcher
+
+
+def _strict_api_task(api: dict) -> TaskEnvelopeStrict:
+    return TaskEnvelopeStrict.model_validate(
+        {
+            "apiVersion": "mloc/v1",
+            "kind": "Task",
+            "metadata": {"name": "t"},
+            "spec": {"taskType": "api", "api": api},
+        }
+    )
+
+
+def _strict_spec(api: dict) -> ApiSpecStrict:
+    spec = _strict_api_task(api).spec
+    assert isinstance(spec, ApiSpecStrict)
+    return spec
+
+
+def _dispatcher() -> Dispatcher:
+    return make_capturing_dispatcher()
+
+
+class TestContainsRedacted:
+    def test_plain_placeholder(self) -> None:
+        assert contains_redacted(REDACTED)
+
+    def test_nested_in_dict(self) -> None:
+        assert contains_redacted({"auth": {"token": REDACTED}})
+
+    def test_nested_in_list(self) -> None:
+        assert contains_redacted([{"token": REDACTED}])
+
+    def test_innocent_value(self) -> None:
+        assert not contains_redacted("Bearer SECRET")
+
+    def test_innocent_nested(self) -> None:
+        assert not contains_redacted({"model": "gpt", "monkey": "x"})
+
+
+class TestHasRedactedCredential:
+    def test_headers_redacted_detected(self) -> None:
+        spec = _strict_spec({"headers": {"Authorization": REDACTED}})
+        assert _dispatcher()._has_redacted_credential(spec)
+
+    def test_params_redacted_detected(self) -> None:
+        spec = _strict_spec({"params": {"api_key": REDACTED}})
+        assert _dispatcher()._has_redacted_credential(spec)
+
+    def test_body_redacted_detected(self) -> None:
+        spec = _strict_spec({"body": {"secret": REDACTED}})
+        assert _dispatcher()._has_redacted_credential(spec)
+
+    def test_json_redacted_detected(self) -> None:
+        spec = _strict_spec({"json": {"token": REDACTED}})
+        assert _dispatcher()._has_redacted_credential(spec)
+
+    def test_data_redacted_detected(self) -> None:
+        spec = _strict_spec({"data": {"access_token": REDACTED}})
+        assert _dispatcher()._has_redacted_credential(spec)
+
+    def test_nested_redacted_detected(self) -> None:
+        spec = _strict_spec({"json": {"auth": {"token": REDACTED}}})
+        assert _dispatcher()._has_redacted_credential(spec)
+
+    def test_redacted_in_list_detected(self) -> None:
+        spec = _strict_spec({"json": [{"token": REDACTED}]})
+        assert _dispatcher()._has_redacted_credential(spec)
+
+    def test_no_credential_not_detected(self) -> None:
+        spec = _strict_spec({"json": {"model": "gpt"}})
+        assert not _dispatcher()._has_redacted_credential(spec)
+
+    def test_real_credential_not_detected(self) -> None:
+        spec = _strict_spec({"headers": {"Authorization": "Bearer SECRET"}})
+        assert not _dispatcher()._has_redacted_credential(spec)
+
+    def test_non_api_spec_not_detected(self) -> None:
+        spec = TaskEnvelopeStrict.model_validate(
+            {
+                "apiVersion": "flowmesh/v1",
+                "kind": "Task",
+                "metadata": {"name": "t"},
+                "spec": {"taskType": "echo", "data": {"token": REDACTED}},
+            }
+        ).spec
+        assert not _dispatcher()._has_redacted_credential(spec)
diff --git a/tests/server/task/test_ssh_result_mounting.py b/tests/server/task/test_ssh_result_mounting.py
index 333d9dafe..1953be090 100644
--- a/tests/server/task/test_ssh_result_mounting.py
+++ b/tests/server/task/test_ssh_result_mounting.py
@@ -76,7 +76,7 @@ def test_dispatcher_resolves_ssh_input_stage_names_from_local_stage_names() -> N
         task_id="task-pre",
         workflow_id="wf-1",
         owner_id="owner",
-        raw_yaml="raw",
+        source="raw",
         task=_task_template(TaskType.ECHO),
         status=TaskStatus.DONE,
         task_type="echo",
@@ -91,7 +91,7 @@ def test_dispatcher_resolves_ssh_input_stage_names_from_local_stage_names() -> N
         task_id="task-ssh",
         workflow_id="wf-1",
         owner_id="owner",
-        raw_yaml="raw",
+        source="raw",
         task=current_task,
         status=TaskStatus.PENDING,
         task_type="ssh",
@@ -122,7 +122,7 @@ def test_dispatcher_requeues_when_ssh_input_stage_not_done() -> None:
         task_id="task-pre",
         workflow_id="wf-1",
         owner_id="owner",
-        raw_yaml="raw",
+        source="raw",
         task=_task_template(TaskType.ECHO),
         status=TaskStatus.PENDING,
         task_type="echo",
@@ -132,7 +132,7 @@ def test_dispatcher_requeues_when_ssh_input_stage_not_done() -> None:
         task_id="task-ssh",
         workflow_id="wf-1",
         owner_id="owner",
-        raw_yaml="raw",
+        source="raw",
         task=_task_template(TaskType.SSH, inputs=[{"stage": "preprocess"}]),
         status=TaskStatus.PENDING,
         task_type="ssh",
@@ -161,7 +161,7 @@ def test_build_stage_context_includes_only_transitive_dependencies() -> None:
         task_id="task-pre",
         workflow_id="wf-1",
         owner_id="owner",
-        raw_yaml="raw",
+        source="raw",
         task=_task_template(TaskType.ECHO),
         status=TaskStatus.DONE,
         task_type="echo",
@@ -171,7 +171,7 @@ def test_build_stage_context_includes_only_transitive_dependencies() -> None:
         task_id="task-mid",
         workflow_id="wf-1",
         owner_id="owner",
-        raw_yaml="raw",
+        source="raw",
         task=_task_template(TaskType.ECHO),
         status=TaskStatus.DONE,
         task_type="echo",
@@ -181,7 +181,7 @@ def test_build_stage_context_includes_only_transitive_dependencies() -> None:
         task_id="task-unrelated",
         workflow_id="wf-1",
         owner_id="owner",
-        raw_yaml="raw",
+        source="raw",
         task=_task_template(TaskType.ECHO),
         status=TaskStatus.DONE,
         task_type="echo",
@@ -191,7 +191,7 @@ def test_build_stage_context_includes_only_transitive_dependencies() -> None:
         task_id="task-final",
         workflow_id="wf-1",
         owner_id="owner",
-        raw_yaml="raw",
+        source="raw",
         task=_task_template(
             TaskType.ECHO,
             data={"message": "${preprocess.responses.0.output}"},
@@ -253,7 +253,7 @@ def test_collect_upstream_results_excludes_unrelated_completed_stages(
         task_id="task-pre",
         workflow_id="wf-1",
         owner_id="owner",
-        raw_yaml="raw",
+        source="raw",
         task=_task_template(TaskType.ECHO),
         status=TaskStatus.DONE,
         task_type="echo",
@@ -263,7 +263,7 @@ def test_collect_upstream_results_excludes_unrelated_completed_stages(
         task_id="task-other",
         workflow_id="wf-1",
         owner_id="owner",
-        raw_yaml="raw",
+        source="raw",
         task=_task_template(TaskType.ECHO),
         status=TaskStatus.DONE,
         task_type="echo",
@@ -273,7 +273,7 @@ def test_collect_upstream_results_excludes_unrelated_completed_stages(
         task_id="task-final",
         workflow_id="wf-1",
         owner_id="owner",
-        raw_yaml="raw",
+        source="raw",
         task=_task_template(
             TaskType.ECHO,
             data={"message": "${preprocess.responses.0.output}"},
@@ -351,7 +351,7 @@ def test_stage_reference_uses_payload_root_for_local_and_http_results(
         task_id="task-local",
         workflow_id="wf-1",
         owner_id="owner",
-        raw_yaml="raw",
+        source="raw",
         task=_task_template(TaskType.ECHO),
         status=TaskStatus.DONE,
         task_type="echo",
@@ -361,7 +361,7 @@ def test_stage_reference_uses_payload_root_for_local_and_http_results(
         task_id="task-http",
         workflow_id="wf-1",
         owner_id="owner",
-        raw_yaml="raw",
+        source="raw",
         task=_task_template(TaskType.ECHO),
         status=TaskStatus.DONE,
         task_type="echo",
diff --git a/tests/server/test_hooks_wiring.py b/tests/server/test_hooks_wiring.py
index bd3843649..b443651a4 100644
--- a/tests/server/test_hooks_wiring.py
+++ b/tests/server/test_hooks_wiring.py
@@ -544,7 +544,7 @@ def _make_runtime_with_record(task_id: str) -> tuple[TaskRuntime, TaskRecord]:
         task_id=task_id,
         workflow_id="wfl-1",
         owner_id="admin",
-        raw_yaml="",
+        source="",
         task=env,
     )
     runtime._tasks[task_id] = record  # type: ignore[attr-defined]
diff --git a/tests/server/test_redact.py b/tests/server/test_redact.py
index 83fbba0b1..1b0fbb4ae 100644
--- a/tests/server/test_redact.py
+++ b/tests/server/test_redact.py
@@ -1,11 +1,12 @@
 """Tests for API credential redaction at serialization time."""
 
 import json
+from unittest import mock
 
 from server.task.models import TaskRecord
-from server.task.redact import REDACTED, _is_sensitive_key, redact_api, redact_raw_yaml
 from shared.tasks import TaskEnvelopeTemplate
 from shared.tasks.specs import ApiSpecTemplate
+from shared.utils.redact import REDACTED, is_sensitive_key, redact_api, redact_raw_yaml
 
 
 def _api_task(api: dict) -> TaskEnvelopeTemplate:
@@ -19,12 +20,12 @@ def _api_task(api: dict) -> TaskEnvelopeTemplate:
     )
 
 
-def _record(api: dict, raw_yaml: str = "") -> TaskRecord:
+def _record(api: dict, source: str = "") -> TaskRecord:
     return TaskRecord(
         task_id="tsk-1",
         workflow_id="wfl-1",
         owner_id="owner",
-        raw_yaml=raw_yaml,
+        source=source,
         task=_api_task(api),
     )
 
@@ -45,11 +46,11 @@ def test_sensitive_keys_match(self) -> None:
             "my_token",
             "my_key",
         ):
-            assert _is_sensitive_key(key), key
+            assert is_sensitive_key(key), key
 
     def test_innocent_keys_do_not_match(self) -> None:
         for key in ("monkey", "turkey", "keyword", "keys", "model", "messages"):
-            assert not _is_sensitive_key(key), key
+            assert not is_sensitive_key(key), key
 
 
 class TestRedactApi:
@@ -142,13 +143,13 @@ def test_dump_json_redacts(self) -> None:
         dumped = json.loads(rec.model_dump_json())
         assert dumped["task"]["spec"]["api"]["headers"]["Authorization"] == REDACTED
 
-    def test_raw_yaml_redacted_in_dump(self) -> None:
+    def test_source_redacted_in_dump(self) -> None:
         rec = _record(
             {"headers": {"Authorization": "Bearer SECRET"}},
-            raw_yaml="api:\n  headers:\n    Authorization: Bearer SECRET\n",
+            source="api:\n  headers:\n    Authorization: Bearer SECRET\n",
         )
         dumped = rec.model_dump()
-        assert "SECRET" not in dumped["raw_yaml"]
+        assert "SECRET" not in dumped["source"]
 
     def test_in_memory_keeps_real_credential(self) -> None:
         rec = _record({"headers": {"Authorization": "Bearer SECRET"}})
@@ -161,3 +162,21 @@ def test_no_credential_unchanged(self) -> None:
         rec = _record(api)
         dumped = rec.model_dump()
         assert dumped["task"]["spec"]["api"] == api
+
+    def test_redaction_cached_across_dumps(self) -> None:
+        rec = _record(
+            {"headers": {"Authorization": "Bearer SECRET"}},
+            source="api:\n  headers:\n    Authorization: Bearer SECRET\n",
+        )
+        with mock.patch(
+            "server.task.models.redact_raw_yaml",
+            wraps=redact_raw_yaml,
+        ) as spy:
+            for _ in range(3):
+                dumped = rec.model_dump()
+                assert dumped["source"] != "SECRET"
+                assert (
+                    dumped["task"]["spec"]["api"]["headers"]["Authorization"]
+                    == REDACTED
+                )
+        assert spy.call_count == 1
diff --git a/tests/worker/test_api_executor.py b/tests/worker/test_api_executor.py
index 4dafbcfc9..906422a56 100644
--- a/tests/worker/test_api_executor.py
+++ b/tests/worker/test_api_executor.py
@@ -97,20 +97,23 @@ def test_neither_url_nor_base_url_raises(
 
 
 class TestCustomUrl:
-    def test_custom_url_without_credential_raises(
+    def test_custom_url_without_credential_sends_no_nebula_token(
         self, monkeypatch: pytest.MonkeyPatch
     ) -> None:
-        """A custom endpoint must carry its own credential.
+        """A custom endpoint may be unauthenticated, but never gets the Nebula token.
 
-        The Nebula token is available here, so the failure proves it is withheld
-        rather than merely absent: a caller-chosen endpoint never receives it.
+        An unauthorized serving endpoint must stay usable, so a missing
+        credential is not an error. The Nebula token IS set in the environment
+        here, so the absent Authorization header proves it is withheld rather
+        than merely unavailable: a caller-chosen endpoint never receives it.
         """
         monkeypatch.setenv("NEBULA_API_TOKEN", "nebula-token")
         task = _task_message(url="https://custom.example.com/v1/chat/completions")
         transport = _RecordingTransport()
-        with pytest.raises(ExecutionError, match="custom endpoint"):
-            _run(APIExecutor.__new__(APIExecutor), task, transport)
-        assert transport.request is None
+        _run(APIExecutor.__new__(APIExecutor), task, transport)
+        assert transport.request is not None
+        assert transport.request.url == "https://custom.example.com/v1/chat/completions"
+        assert "Authorization" not in transport.request.headers
 
     def test_custom_url_with_header_preserves_header(
         self, monkeypatch: pytest.MonkeyPatch
@@ -125,3 +128,37 @@ def test_custom_url_with_header_preserves_header(
         assert transport.request is not None
         assert transport.request.url == "https://custom.example.com/v1/chat/completions"
         assert transport.request.headers["Authorization"] == "Bearer custom"
+
+    def test_custom_url_with_x_api_key_accepted(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """A custom endpoint may authenticate with a non-Authorization header."""
+        monkeypatch.setenv("NEBULA_API_TOKEN", "nebula-token")
+        task = _task_message(
+            url="https://custom.example.com/v1/chat/completions",
+            headers={"X-API-Key": "custom-key"},
+        )
+        transport = _RecordingTransport()
+        _run(APIExecutor.__new__(APIExecutor), task, transport)
+        assert transport.request is not None
+        assert transport.request.url == "https://custom.example.com/v1/chat/completions"
+        assert transport.request.headers["X-API-Key"] == "custom-key"
+
+    def test_custom_url_with_only_innocent_header_stays_unauthenticated(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """A non-credential header leaves the request unauthenticated, not rejected.
+
+        Content-Type is not a credential, so nothing here authenticates the
+        request -- and the Nebula token still must not be substituted in.
+        """
+        monkeypatch.setenv("NEBULA_API_TOKEN", "nebula-token")
+        task = _task_message(
+            url="https://custom.example.com/v1/chat/completions",
+            headers={"Content-Type": "application/json"},
+        )
+        transport = _RecordingTransport()
+        _run(APIExecutor.__new__(APIExecutor), task, transport)
+        assert transport.request is not None
+        assert transport.request.headers["Content-Type"] == "application/json"
+        assert "Authorization" not in transport.request.headers

From ab0a91c05035b64421d1112d9902beb41f385d45 Mon Sep 17 00:00:00 2001
From: Zhengyuan Su <su.zhengyuan@u.nus.edu>
Date: Fri, 18 Sep 2026 18:48:35 +0700
Subject: [PATCH 8/8] Address PR comments.

Signed-off-by: Zhengyuan Su <su.zhengyuan@u.nus.edu>
---
 src/server/task/models.py   |  9 ++++++++-
 src/shared/utils/redact.py  | 22 ++++++++++++----------
 tests/server/test_redact.py | 36 ++++++++++++++++++++++++++++++++++++
 3 files changed, 56 insertions(+), 11 deletions(-)

diff --git a/src/server/task/models.py b/src/server/task/models.py
index 4445af8f4..4524df171 100644
--- a/src/server/task/models.py
+++ b/src/server/task/models.py
@@ -2,7 +2,9 @@
 from typing import Any
 
 from pydantic import (
+    AliasChoices,
     BaseModel,
+    ConfigDict,
     Field,
     PrivateAttr,
     SerializerFunctionWrapHandler,
@@ -78,12 +80,17 @@ def from_payload(cls, payload: dict[str, Any], status: str) -> "TaskUsage | None
 
 
 class TaskRecord(BaseModel):
+    model_config = ConfigDict(validate_by_alias=True)
+
     task_id: str = Field(description="Task identifier.")
     workflow_id: str = Field(description="Workflow identifier.")
     owner_id: str = Field(description="Owner principal identifier.")
     org_id: str = Field(default="", description="Owner organization identifier.")
     supplier_id: str = Field(default="", description="Supplier identifier.")
-    source: str = Field(description="Original workflow source (YAML or JSON).")
+    source: str = Field(
+        validation_alias=AliasChoices("source", "raw_yaml"),
+        description="Original workflow source (YAML or JSON).",
+    )
     task: TaskEnvelopeTemplate = Field(description="Task template.")
     status: str = Field(default=TaskStatus.PENDING, description="Task status.")
     task_type: str | None = Field(default=None, description="Task type.")
diff --git a/src/shared/utils/redact.py b/src/shared/utils/redact.py
index 2c0412a1f..c81ce0a74 100644
--- a/src/shared/utils/redact.py
+++ b/src/shared/utils/redact.py
@@ -1,13 +1,15 @@
-"""Redaction of API credentials before a task record is serialized.
-
-The in-memory ``TaskRecord`` keeps the real credential so dispatch works; the
-serializer applies this module so that every dump to Redis is redacted.
-
-The ``source`` field is re-emitted via ``yaml.safe_dump``, which does not
-preserve comments, key order or original formatting. That is an accepted cost:
-the field is a stored record and is never re-parsed, so losing formatting is
-fine. If the YAML cannot be parsed, the whole field is redacted rather than
-storing text that might contain a key.
+"""Redacts credential-shaped fields from an object before it is serialized.
+
+Applies a key-based rule: any header, parameter, or nested field whose name
+looks like a credential (see ``is_sensitive_key``) has its value replaced with
+a fixed marker. Callers keep an unredacted copy for in-process use and apply
+this module only at the serialization boundary.
+
+Raw YAML text is redacted by parsing and re-emitting it via
+``yaml.safe_dump``, which does not preserve comments, key order or original
+formatting. That is an accepted cost for a value that is stored and never
+re-parsed. If the YAML cannot be parsed, the whole field is redacted rather
+than storing text that might contain a key.
 """
 
 from typing import Any
diff --git a/tests/server/test_redact.py b/tests/server/test_redact.py
index 1b0fbb4ae..5f89fa385 100644
--- a/tests/server/test_redact.py
+++ b/tests/server/test_redact.py
@@ -180,3 +180,39 @@ def test_redaction_cached_across_dumps(self) -> None:
                     == REDACTED
                 )
         assert spy.call_count == 1
+
+
+class TestSourceFieldAlias:
+    def test_old_key_populates_source(self) -> None:
+        rec = TaskRecord.model_validate(
+            {
+                "task_id": "tsk-1",
+                "workflow_id": "wfl-1",
+                "owner_id": "owner",
+                "raw_yaml": "model: gpt-4o\n",
+                "task": _api_task({"url": "http://x"}),
+            }
+        )
+        assert rec.source == "model: gpt-4o\n"
+
+    def test_new_key_populates_source(self) -> None:
+        rec = TaskRecord.model_validate(
+            {
+                "task_id": "tsk-1",
+                "workflow_id": "wfl-1",
+                "owner_id": "owner",
+                "source": "model: gpt-4o\n",
+                "task": _api_task({"url": "http://x"}),
+            }
+        )
+        assert rec.source == "model: gpt-4o\n"
+
+    def test_constructed_with_source_kwarg(self) -> None:
+        rec = TaskRecord(
+            task_id="tsk-1",
+            workflow_id="wfl-1",
+            owner_id="owner",
+            source="model: gpt-4o\n",
+            task=_api_task({"url": "http://x"}),
+        )
+        assert rec.source == "model: gpt-4o\n"