fix: send deployment header from serverless presigned-URL IO manager (#25897)

## Summary

Materializing assets in a serverless org with **more than one full
deployment** fails with a 401 from the IO manager:

```
requests.exceptions.HTTPError: 401 Client Error: Unauthorized for url:
  https://<org>.agent.<env>.dagster.cloud/gen_io_storage_url?key=raw_orders&method=PUT
  File ".../dagster_cloud/serverless/io_manager.py", line 138, in _get_presigned_url
    resp.raise_for_status()
```

**Root cause.** `ServerlessPresignedURLIOManager._get_presigned_url`
builds its headers with the raw `get_dagster_cloud_api_headers(token,
DagsterCloudInstanceScope.DEPLOYMENT)` and **never passes a
`deployment_name`**, so the `Dagster-Cloud-Deployment` header is
omitted. Ursula's auth (`_get_deployment`) then falls back to "if the
org has exactly one deployment, use it; otherwise 401":

```
Organization has multiple deployments, but no deployment specified
```

Single-deployment serverless orgs hit the one-deployment branch and
worked; with per-deployment tenancy an org can now have `prod` +
`stage`, so it 401s.

Every other run-pod callback (compute logs, insights, defs-state,
GraphQL) goes through `instance.dagster_cloud_api_headers` →
`get_agent_headers`, which derives `deployment_name` from the instance
config. This IO manager was the only path that bypassed that helper.

## Fix

Thread `init_context.instance.deployment_name` into the manager and send
it in the header — matching how the non-presigned S3 path already reads
`instance.deployment_name`, and how every other caller sets the header.

## Test Plan

- `dagster_cloud_tests/serverless_tests/test_io_manager.py` → 13 passed.
Added an assertion that `dump_to_path` sends `Dagster-Cloud-Deployment`,
and that the factory threads `deployment_name` onto the manager.
- `just ty` → 0 errors.

## Changelog

Fix a 401 from the serverless IO manager when materializing assets in an
org with more than one full deployment.

Internal-RevId: 724b9db96b53cbc581e9460aa2038113b8b7176e

Author: sndagster

python_modules/dagster-cloud/dagster_cloud/serverless/io_manager.py

@@ -108,9 +108,10 @@ class ServerlessPresignedURLIOManager(UPathIOManager):
     obtained from the Dagster+ API on each operation.
     """
 
-    def __init__(self, api_url: str, api_token: str, timeout: int):
+    def __init__(self, api_url: str, api_token: str, timeout: int, deployment_name: str | None):
         self._api_url = api_url
         self._api_token = api_token
+        self._deployment_name = deployment_name
         self._timeout = timeout
         self._session = requests.Session()
         put_retry_adapter = HTTPAdapter(
@@ -131,6 +132,7 @@ def _get_presigned_url(self, key: str, method: str) -> str:
             headers=get_dagster_cloud_api_headers(
                 self._api_token,
                 DagsterCloudInstanceScope.DEPLOYMENT,
+                deployment_name=self._deployment_name,
             ),
             params={"key": key, "method": method},
             timeout=self._timeout,
@@ -182,6 +184,7 @@ def _build_serverless_io_manager(init_context):
             api_url=init_context.instance.dagster_cloud_url,
             api_token=init_context.instance.dagster_cloud_agent_token,
             timeout=init_context.instance.dagster_cloud_api_timeout,
+            deployment_name=init_context.instance.deployment_name,
         )
 
     bucket = os.getenv("DAGSTER_CLOUD_SERVERLESS_STORAGE_S3_BUCKET")

python_modules/dagster-cloud/dagster_cloud_tests/serverless_tests/test_io_manager.py

@@ -58,11 +58,13 @@ def test_factory_returns_presigned_when_env_var_set():
     init_context.instance.dagster_cloud_url = "https://myorg.agent.dagster.cloud"
     init_context.instance.dagster_cloud_agent_token = "test-token"
     init_context.instance.dagster_cloud_api_timeout = 30
+    init_context.instance.deployment_name = "prod"
 
     with environ({"SERVERLESS_IO_MANAGER_USE_PRESIGNED_URL": "True"}):
         manager = _build_serverless_io_manager(init_context)
 
     assert isinstance(manager, ServerlessPresignedURLIOManager)
+    assert manager._deployment_name == "prod"
 
 
 def test_factory_returns_s3_when_env_var_not_set():
@@ -93,12 +95,15 @@ def test_factory_returns_s3_when_env_var_not_set():
 
 API_URL = "https://myorg.agent.dagster.cloud"
 API_TOKEN = "test-token"
+DEPLOYMENT_NAME = "prod"
 PRESIGNED_PUT_URL = "https://s3.amazonaws.com/bucket/key?X-Amz-Signature=put123"
 PRESIGNED_GET_URL = "https://s3.amazonaws.com/bucket/key?X-Amz-Signature=get123"
 
 
 def _make_manager() -> ServerlessPresignedURLIOManager:
-    return ServerlessPresignedURLIOManager(api_url=API_URL, api_token=API_TOKEN, timeout=30)
+    return ServerlessPresignedURLIOManager(
+        api_url=API_URL, api_token=API_TOKEN, timeout=30, deployment_name=DEPLOYMENT_NAME
+    )
 
 
 def _presigned_url_response(url: str):
@@ -145,6 +150,7 @@ def test_dump_to_path_calls_put():
             "key": "storage/run-123/my_step/result",
             "method": "PUT",
         }
+        assert call_kwargs.kwargs["headers"]["Dagster-Cloud-Deployment"] == DEPLOYMENT_NAME
 
         mock_put.assert_called_once()
         (put_url,) = mock_put.call_args.args
@@ -331,7 +337,7 @@ def test_module_imports_without_boto3_installed():
             assert reloaded.ServerlessPresignedURLIOManager is not None
             # Instantiating the presigned-URL manager must not touch boto3.
             manager = reloaded.ServerlessPresignedURLIOManager(
-                api_url=API_URL, api_token=API_TOKEN, timeout=30
+                api_url=API_URL, api_token=API_TOKEN, timeout=30, deployment_name=DEPLOYMENT_NAME
             )
             assert manager is not None
     finally: