diff --git a/skyvern-frontend/src/routes/workflows/components/CredentialSelector.tsx b/skyvern-frontend/src/routes/workflows/components/CredentialSelector.tsx
index 1bc2c4507..6fe24ae73 100644
--- a/skyvern-frontend/src/routes/workflows/components/CredentialSelector.tsx
+++ b/skyvern-frontend/src/routes/workflows/components/CredentialSelector.tsx
@@ -20,9 +20,10 @@ import {
 type Props = {
   value: string;
   onChange: (value: string) => void;
+  placeholder?: string;
 };
 
-function CredentialSelector({ value, onChange }: Props) {
+function CredentialSelector({ value, onChange, placeholder }: Props) {
   const { setIsOpen, setType } = useCredentialModalState();
   const { data: credentials, isFetching } = useCredentialsQuery({
     page_size: 100, // Reasonable limit for dropdown selector
@@ -50,7 +51,7 @@ function CredentialSelector({ value, onChange }: Props) {
         }}
       >
         <SelectTrigger>
-          <SelectValue placeholder="Select a credential" />
+          <SelectValue placeholder={placeholder ?? "Select a credential"} />
         </SelectTrigger>
         <SelectContent>
           {credentials.map((credential) => (
diff --git a/skyvern-frontend/src/routes/workflows/editor/nodes/WorkflowTriggerNode/PayloadParameterFields.tsx b/skyvern-frontend/src/routes/workflows/editor/nodes/WorkflowTriggerNode/PayloadParameterFields.tsx
index b7d711312..d1189aa44 100644
--- a/skyvern-frontend/src/routes/workflows/editor/nodes/WorkflowTriggerNode/PayloadParameterFields.tsx
+++ b/skyvern-frontend/src/routes/workflows/editor/nodes/WorkflowTriggerNode/PayloadParameterFields.tsx
@@ -1,8 +1,12 @@
 import { useCallback, useRef } from "react";
 import { Label } from "@/components/ui/label";
 import { WorkflowBlockInputTextarea } from "@/components/WorkflowBlockInputTextarea";
-import { WorkflowParameter } from "@/routes/workflows/types/workflowTypes";
+import {
+  WorkflowParameter,
+  WorkflowParameterValueType,
+} from "@/routes/workflows/types/workflowTypes";
 import { Skeleton } from "@/components/ui/skeleton";
+import { CredentialSelector } from "../../../components/CredentialSelector";
 
 interface PayloadParameterFieldsProps {
   parameters: Array<WorkflowParameter>;
@@ -89,17 +93,30 @@ function PayloadParameterFields({
           {param.description && (
             <p className="text-[10px] text-slate-500">{param.description}</p>
           )}
-          <WorkflowBlockInputTextarea
-            nodeId={nodeId}
-            onChange={(val) => handleFieldChange(param.key, val)}
-            value={payloadValues[param.key] ?? ""}
-            placeholder={
-              param.default_value != null
-                ? `Default: ${String(param.default_value)}`
-                : `Enter ${param.key}...`
-            }
-            className="nopan text-xs"
-          />
+          {param.workflow_parameter_type ===
+          WorkflowParameterValueType.CredentialId ? (
+            <CredentialSelector
+              value={payloadValues[param.key] ?? ""}
+              onChange={(val) => handleFieldChange(param.key, val)}
+              placeholder={
+                param.default_value != null
+                  ? `Default: ${String(param.default_value)}`
+                  : "Select a credential"
+              }
+            />
+          ) : (
+            <WorkflowBlockInputTextarea
+              nodeId={nodeId}
+              onChange={(val) => handleFieldChange(param.key, val)}
+              value={payloadValues[param.key] ?? ""}
+              placeholder={
+                param.default_value != null
+                  ? `Default: ${String(param.default_value)}`
+                  : `Enter ${param.key}...`
+              }
+              className="nopan text-xs"
+            />
+          )}
         </div>
       ))}
     </div>
diff --git a/skyvern/exceptions.py b/skyvern/exceptions.py
index a8d12f941..328caaf84 100644
--- a/skyvern/exceptions.py
+++ b/skyvern/exceptions.py
@@ -222,10 +222,34 @@ class WorkflowRunParameterPersistenceError(SkyvernException):
         )
 
 
+# Covers the credential dict fields from SKY-8222 (password, username, secret_value, totp).
+# Not exhaustive — this is defense-in-depth; the root cause is fixed in the frontend.
+_SENSITIVE_CREDENTIAL_KEYS = ("password", "username", "secret", "totp", "secret_value")
+
+
+def sanitize_credential_for_error(credential_id: object) -> str:
+    """Prevent credential values from leaking into error messages.
+
+    When a credential dict is accidentally stringified and passed as a credential ID,
+    this ensures the raw values (passwords, usernames, etc.) are never included in
+    user-facing error messages, failure reasons, or logs.
+    """
+    if not isinstance(credential_id, str):
+        return f"<redacted - non-string type: {type(credential_id).__name__}>"
+    lower = credential_id.lower()
+    for key in _SENSITIVE_CREDENTIAL_KEYS:
+        if key in lower:
+            return "<redacted - contains credential data>"
+    if len(credential_id) > 200:
+        return "<redacted - value too long>"
+    return credential_id
+
+
 class InvalidCredentialId(SkyvernHTTPException):
     def __init__(self, credential_id: str) -> None:
         super().__init__(
-            f"Invalid credential ID: {credential_id}. Failed to resolve to a valid credential.",
+            f"Invalid credential ID: {sanitize_credential_for_error(credential_id)}."
+            " Failed to resolve to a valid credential.",
             status_code=status.HTTP_400_BAD_REQUEST,
         )
 
@@ -459,8 +483,10 @@ class CredentialParameterParsingError(SkyvernException):
 
 
 class CredentialParameterNotFoundError(SkyvernException):
-    def __init__(self, credential_parameter_id: str) -> None:
-        super().__init__(f"Could not find credential parameter: {credential_parameter_id}")
+    def __init__(self, credential_parameter_id: str | None) -> None:
+        super().__init__(
+            f"Could not find credential parameter: {sanitize_credential_for_error(credential_parameter_id)}"
+        )
 
 
 class CredentialVaultNotConfiguredError(SkyvernException):
diff --git a/skyvern/forge/sdk/workflow/context_manager.py b/skyvern/forge/sdk/workflow/context_manager.py
index 6698f1139..c43bdca1d 100644
--- a/skyvern/forge/sdk/workflow/context_manager.py
+++ b/skyvern/forge/sdk/workflow/context_manager.py
@@ -15,6 +15,7 @@ from skyvern.exceptions import (
     ImaginarySecretValue,
     SkyvernException,
     WorkflowRunContextNotInitialized,
+    sanitize_credential_for_error,
 )
 from skyvern.forge import app
 from skyvern.forge.sdk.api.aws import AsyncAWSClient
@@ -378,12 +379,15 @@ class WorkflowRunContext:
         """
         # Check if it's in the format vault_id:item_id
         if ":" in credential_id:
-            LOG.info(f"Processing credential in vault_id:item_id format: {credential_id}")
+            LOG.info("Processing credential in vault_id:item_id format")
             vault_id, item_id = credential_id.split(":", 1)
             return vault_id, item_id
 
         # If we can't parse the credential_id, raise an error
-        raise ValueError(f"Invalid credential format: {credential_id}. Expected format: vault_id:item_id")
+        raise ValueError(
+            f"Invalid credential format: {sanitize_credential_for_error(credential_id)}."
+            " Expected format: vault_id:item_id"
+        )
 
     async def _register_credential_parameter_value(
         self,
@@ -447,13 +451,13 @@ class WorkflowRunContext:
                 f"Trying to register workflow parameter as a secret but it is not a string. Parameter key: {parameter.key}"
             )
 
-        LOG.info(f"Fetching credential parameter value for credential: {credential_id}")
+        LOG.info("Fetching credential parameter value", parameter_key=parameter.key)
 
         # Handle regular credentials from the database
         try:
             await self._register_credential_parameter_value(credential_id, parameter, organization)
         except Exception as e:
-            LOG.error(f"Failed to get credential from database: {credential_id}. Error: {e}")
+            LOG.error("Failed to get credential from database", parameter_key=parameter.key, exc_info=True)
             raise e
 
     async def register_credential_parameter_value(
@@ -461,7 +465,7 @@ class WorkflowRunContext:
         parameter: CredentialParameter,
         organization: Organization,
     ) -> None:
-        LOG.info(f"Fetching credential parameter value for credential: {parameter.credential_id}")
+        LOG.info("Fetching credential parameter value", parameter_key=parameter.key)
 
         credential_id = None
         if parameter.credential_id:
@@ -471,7 +475,7 @@ class WorkflowRunContext:
                 credential_id = parameter.credential_id
 
         if credential_id is None:
-            LOG.error(f"Credential ID not found for credential: {parameter.credential_id}")
+            LOG.error("Credential ID not found", parameter_key=parameter.key)
             raise CredentialParameterNotFoundError(parameter.credential_id)
 
         await self._register_credential_parameter_value(credential_id, parameter, organization)
diff --git a/skyvern/forge/sdk/workflow/service.py b/skyvern/forge/sdk/workflow/service.py
index 9d11e851e..17ab75d07 100644
--- a/skyvern/forge/sdk/workflow/service.py
+++ b/skyvern/forge/sdk/workflow/service.py
@@ -673,7 +673,9 @@ class WorkflowService:
                         missing_parameters.append(workflow_parameter.key)
                         continue
                     if workflow_parameter.workflow_parameter_type == WorkflowParameterType.CREDENTIAL_ID:
-                        await self._validate_credential_id(str(request_body_value), organization)
+                        if not isinstance(request_body_value, str):
+                            raise InvalidCredentialId(f"<non-string value of type {type(request_body_value).__name__}>")
+                        await self._validate_credential_id(request_body_value, organization)
                     try:
                         await self.create_workflow_run_parameter(
                             workflow_run_id=workflow_run.workflow_run_id,
@@ -689,7 +691,11 @@ class WorkflowService:
                         ) from parameter_error
                 elif workflow_parameter.default_value is not None:
                     if workflow_parameter.workflow_parameter_type == WorkflowParameterType.CREDENTIAL_ID:
-                        await self._validate_credential_id(str(workflow_parameter.default_value), organization)
+                        if not isinstance(workflow_parameter.default_value, str):
+                            raise InvalidCredentialId(
+                                f"<non-string value of type {type(workflow_parameter.default_value).__name__}>"
+                            )
+                        await self._validate_credential_id(workflow_parameter.default_value, organization)
                     try:
                         await self.create_workflow_run_parameter(
                             workflow_run_id=workflow_run.workflow_run_id,
diff --git a/tests/unit/workflow/test_credential_sanitization.py b/tests/unit/workflow/test_credential_sanitization.py
new file mode 100644
index 000000000..f47031f2c
--- /dev/null
+++ b/tests/unit/workflow/test_credential_sanitization.py
@@ -0,0 +1,106 @@
+from skyvern.exceptions import (
+    CredentialParameterNotFoundError,
+    InvalidCredentialId,
+    sanitize_credential_for_error,
+)
+
+
+class TestSanitizeCredentialForError:
+    def test_normal_credential_id_passes_through(self) -> None:
+        assert sanitize_credential_for_error("cred_abc123") == "cred_abc123"
+
+    def test_uuid_credential_id_passes_through(self) -> None:
+        assert sanitize_credential_for_error("550e8400-e29b-41d4-a716-446655440000") == (
+            "550e8400-e29b-41d4-a716-446655440000"
+        )
+
+    def test_redacts_string_containing_password(self) -> None:
+        value = "{'password': 'secret123', 'username': 'user@example.com'}"
+        result = sanitize_credential_for_error(value)
+        assert "secret123" not in result
+        assert "user@example.com" not in result
+        assert "<redacted" in result
+
+    def test_redacts_string_containing_username(self) -> None:
+        result = sanitize_credential_for_error("username=admin")
+        assert "admin" not in result
+        assert "<redacted" in result
+
+    def test_redacts_string_containing_secret(self) -> None:
+        result = sanitize_credential_for_error("secret_value=my_api_key")
+        assert "my_api_key" not in result
+        assert "<redacted" in result
+
+    def test_redacts_string_containing_totp(self) -> None:
+        result = sanitize_credential_for_error("totp=JBSWY3DPEHPK3PXP")
+        assert "JBSWY3DPEHPK3PXP" not in result
+        assert "<redacted" in result
+
+    def test_redacts_excessively_long_string(self) -> None:
+        value = "a" * 201
+        result = sanitize_credential_for_error(value)
+        assert result == "<redacted - value too long>"
+
+    def test_case_insensitive_detection(self) -> None:
+        result = sanitize_credential_for_error("{'PASSWORD': 'secret'}")
+        assert "<redacted" in result
+
+    def test_empty_string_passes_through(self) -> None:
+        assert sanitize_credential_for_error("") == ""
+
+    def test_none_returns_redacted(self) -> None:
+        assert sanitize_credential_for_error(None) == "<redacted - non-string type: NoneType>"
+
+    def test_dict_returns_redacted(self) -> None:
+        result = sanitize_credential_for_error({"password": "secret", "username": "user"})
+        assert "secret" not in result
+        assert "user" not in result
+        assert "<redacted - non-string type: dict>" == result
+
+
+class TestInvalidCredentialIdSanitization:
+    def test_normal_id_in_message(self) -> None:
+        exc = InvalidCredentialId("cred_abc123")
+        assert "cred_abc123" in str(exc)
+        assert "Invalid credential ID" in str(exc)
+
+    def test_credential_dict_redacted_in_message(self) -> None:
+        credential_dict_str = str({"password": "real_password", "username": "real_user@example.com"})
+        exc = InvalidCredentialId(credential_dict_str)
+        assert "real_password" not in str(exc)
+        assert "real_user@example.com" not in str(exc)
+        assert "<redacted" in str(exc)
+
+    def test_message_attribute_also_sanitized(self) -> None:
+        credential_dict_str = str({"password": "secret", "username": "user"})
+        exc = InvalidCredentialId(credential_dict_str)
+        assert "secret" not in exc.message
+        assert "<redacted" in exc.message
+
+
+class TestInvalidCredentialIdTypeGuard:
+    """Tests for the isinstance guard in service.py that rejects non-string credential values."""
+
+    def test_dict_value_raises_with_safe_message(self) -> None:
+        credential_dict = {"password": "real_password", "username": "real_user@example.com"}
+        exc = InvalidCredentialId(f"<non-string value of type {type(credential_dict).__name__}>")
+        assert "real_password" not in str(exc)
+        assert "real_user@example.com" not in str(exc)
+        assert "non-string value of type dict" in str(exc)
+
+    def test_list_value_raises_with_safe_message(self) -> None:
+        exc = InvalidCredentialId("<non-string value of type list>")
+        assert "non-string value of type list" in str(exc)
+
+
+class TestCredentialParameterNotFoundErrorSanitization:
+    def test_normal_id_in_message(self) -> None:
+        exc = CredentialParameterNotFoundError("cred_abc123")
+        assert "cred_abc123" in str(exc)
+
+    def test_credential_dict_redacted_in_message(self) -> None:
+        credential_dict_str = str({"password": "real_password", "username": "real_user"})
+        exc = CredentialParameterNotFoundError(credential_dict_str)
+        assert "real_password" not in str(exc)
+        assert "real_user" not in str(exc)
+        assert "<redacted" in str(exc)
diff --git a/tests/unit/workflow/test_dag_engine.py b/tests/unit/workflow/test_dag_engine.py
index bcc6af640..da27e1d70 100644
--- a/tests/unit/workflow/test_dag_engine.py
+++ b/tests/unit/workflow/test_dag_engine.py
@@ -84,8 +84,7 @@ class DummyContext:
             return {}
         return self.blocks_metadata.get(label, {})
 
-    def mask_secrets_in_data(self, data: object) -> object:
-        """Mock method - returns data as-is since no secrets in tests."""
+    def mask_secrets_in_data(self, data: object, mask: str = "*****") -> object:
         return data
 
     async def register_output_parameter_value_post_execution(self, parameter: OutputParameter, value: object) -> None:  # noqa: ARG002