diff --git a/docs/api-reference.md b/docs/api-reference.md
index f1caca596..cc65eb149 100644
--- a/docs/api-reference.md
+++ b/docs/api-reference.md
@@ -249,6 +249,30 @@ Gateway-side failures use fixed public messages. Diagnose them with protected
 logs and safe metadata such as request ID, provider, model, and status. Do not
 log provider keys, prompts, responses, or raw upstream bodies.
 
+## Error codes
+
+A refusal a caller is expected to act on carries a stable code, both as an
+`Otari-Error-Code` header and as `code` in the body beside the human-readable
+`detail`: `{"detail": "...", "code": "budget_exceeded"}`. Map refusals by the
+code: it keeps its meaning across releases, while the `detail` text may be
+reworded.
+
+| `Otari-Error-Code` | Status | Meaning | Also sent |
+|---|---|---|---|
+| `budget_exceeded` | 403 | A budget refused the request | `Otari-Budget-Scope`: `user` for the billed user's own budget, otherwise the ceiling's scope: `organization`, `workspace`, `workspace_member`, `org_member` or `api_token` |
+| `user_blocked` | 403 | The billed user is blocked | |
+| `user_not_found` | 404 | The billed user does not exist | |
+| `rate_limited` | 429 | A gateway rate limit is full | `Otari-Rate-Limit-Rule` for a `rate_limits` rule; `Retry-After` when waiting helps |
+| `upstream_rate_limited` | 429 | The provider rate limited the gateway | `Retry-After` when the provider sent one |
+| `invalid_model` | 400 | The model selector names no configured provider | |
+| `model_not_allowed` | 403 | The key may not use the model | |
+| `context_length_exceeded` | 400 | The prompt is too long for the model | |
+| `pricing_required` | 402 | `require_pricing` is on and the model has no price | |
+
+A failure after a stream has started arrives as an error event, which carries
+the code as `error.code` on Chat Completions and Responses:
+`{"error": {"message": "...", "type": "server_error", "code": "upstream_rate_limited"}}`.
+
 ## Caller-orchestrated MCP
 
 Two stored-server endpoints let an application own its own MCP tool loop, as an
diff --git a/src/gateway/api/routes/_pipeline.py b/src/gateway/api/routes/_pipeline.py
index f1063416d..848872ed6 100644
--- a/src/gateway/api/routes/_pipeline.py
+++ b/src/gateway/api/routes/_pipeline.py
@@ -49,7 +49,7 @@
 from urllib.parse import ParseResult, urlparse
 
 from any_llm import LLMProvider
-from any_llm.exceptions import AnyLLMError, InvalidRequestError, UnsupportedParameterError
+from any_llm.exceptions import AnyLLMError, ContextLengthExceededError, InvalidRequestError, UnsupportedParameterError
 from any_llm.types.completion import (
     ChatCompletion,
     ChatCompletionChunk,
@@ -103,6 +103,15 @@
 from gateway.core.config import ATTEMPT_ID_HEADER, REQUEST_ID_HEADER, GatewayConfig
 from gateway.core.database import DATABASE_ERRORS, release_session
 from gateway.core.env import otari_env
+from gateway.core.error_codes import (
+    CONTEXT_LENGTH_EXCEEDED,
+    INVALID_MODEL,
+    MODEL_NOT_ALLOWED,
+    PRICING_REQUIRED,
+    UPSTREAM_RATE_LIMITED,
+    error_code_of,
+    error_headers,
+)
 from gateway.core.metered_pricing import calculate_metered_cost, quantize_cost
 from gateway.core.unit_of_work import UnitOfWork
 from gateway.core.usage import (
@@ -636,20 +645,41 @@ def classify_provider_error(exc: BaseException) -> ProviderErrorMapping | None:
 def provider_error_headers(exc: BaseException, status_code: int) -> dict[str, str] | None:
     """Response headers for a classified provider failure, or ``None``.
 
-    Forwards the upstream ``Retry-After`` on a 429, which is the one header a
-    rate-limited caller can act on and the one piece of a provider's rate-limit
-    response that its message body cannot always carry. Restricted to the 429:
-    on the statuses that surface as a fixed-detail 502 the header would describe
-    the gateway's own upstream account, which is not the caller's to read.
-
-    Returns ``None`` rather than an empty dict when there is nothing to send, so
-    ``HTTPException(headers=...)`` stays unset instead of being handed a dict
-    that adds nothing.
+    On a 429, ``Otari-Error-Code: upstream_rate_limited`` (so a caller can tell
+    the provider's limit from the gateway's own) and the upstream ``Retry-After``,
+    the one piece of a provider's rate-limit response that its message body cannot
+    always carry. Restricted to the 429: on the statuses that surface as a
+    fixed-detail 502 the header would describe the gateway's own upstream
+    account, which is not the caller's to read.
     """
+    if status_code == status.HTTP_400_BAD_REQUEST and _is_context_length_error(exc):
+        return error_headers(CONTEXT_LENGTH_EXCEEDED)
     if status_code != status.HTTP_429_TOO_MANY_REQUESTS:
         return None
+    headers = error_headers(UPSTREAM_RATE_LIMITED)
     retry_after = upstream_retry_after(exc)
-    return {"Retry-After": retry_after} if retry_after is not None else None
+    if retry_after is not None:
+        headers["Retry-After"] = retry_after
+    return headers
+
+
+def _is_context_length_error(exc: BaseException) -> bool:
+    """Whether any-llm classified the failure as a prompt too long for the model."""
+    return any(isinstance(current, ContextLengthExceededError) for current in upstream_exception_chain(exc))
+
+
+def refusal_code(exc: BaseException) -> str | None:
+    """The ``Otari-Error-Code`` an exception ending a request stands for, or None.
+
+    Read by the stream error events, which go out after the headers that would
+    otherwise carry it.
+    """
+    if isinstance(exc, HTTPException):
+        return error_code_of(exc.headers)
+    mapping = classify_provider_error(exc)
+    if mapping is None:
+        return None
+    return error_code_of(provider_error_headers(exc, mapping.status_code))
 
 
 def failure_status_code(exc: BaseException) -> int:
@@ -1041,6 +1071,7 @@ def _raise_for_unresolvable_model(model_selector: str, exc: Exception) -> NoRetu
     raise HTTPException(
         status_code=status.HTTP_400_BAD_REQUEST,
         detail=unresolvable_model_detail(model_selector),
+        headers=error_headers(INVALID_MODEL),
     ) from exc
 
 
@@ -1481,6 +1512,7 @@ async def top_up_reservation_for_attempt(ctx: RequestContext, attempt: Attempt)
         raise HTTPException(
             status_code=status.HTTP_402_PAYMENT_REQUIRED,
             detail=no_pricing_error_detail(f"{attempt.instance}:{attempt.model}"),
+            headers=error_headers(PRICING_REQUIRED),
         )
     repriced = estimate_cost(
         pricing,
@@ -1506,7 +1538,9 @@ async def top_up_reservation_for_attempt(ctx: RequestContext, attempt: Attempt)
             attempt.instance,
             attempt.model,
         )
-        raise HTTPException(status_code=exc.status_code, detail=budget_exhausted_mid_failover_detail()) from exc
+        raise HTTPException(
+            status_code=exc.status_code, detail=budget_exhausted_mid_failover_detail(), headers=exc.headers
+        ) from exc
 
 
 def budget_exhausted_mid_failover_detail() -> str:
@@ -1982,7 +2016,7 @@ async def resolve_request_context(
                 started_at=started_at,
                 request_id=request_id,
             )
-            raise adapter.error(403, not_allowed_detail, ErrorKind.PERMISSION)
+            raise adapter.error(403, not_allowed_detail, ErrorKind.PERMISSION, headers=error_headers(MODEL_NOT_ALLOWED))
 
         # Organization-scoped model restriction (otari#643): the org key
         # resolved for this workspace+provider may narrow which models it
@@ -2013,7 +2047,9 @@ async def resolve_request_context(
                     started_at=started_at,
                     request_id=request_id,
                 )
-                raise adapter.error(403, not_allowed_detail, ErrorKind.PERMISSION)
+                raise adapter.error(
+                    403, not_allowed_detail, ErrorKind.PERMISSION, headers=error_headers(MODEL_NOT_ALLOWED)
+                )
 
         if idempotency is not None and session_principal is None:
             try:
@@ -2155,6 +2191,7 @@ async def resolve_request_context(
                 402,
                 no_pricing_detail,
                 ErrorKind.INVALID_REQUEST,
+                headers=error_headers(PRICING_REQUIRED),
             )
 
         # Resolve uploaded attachments only once the request is authorized
diff --git a/src/gateway/api/routes/chat.py b/src/gateway/api/routes/chat.py
index 58d8ae44e..8e06bcc50 100644
--- a/src/gateway/api/routes/chat.py
+++ b/src/gateway/api/routes/chat.py
@@ -41,6 +41,7 @@
     provider_error_headers,
     raise_all_streaming_attempts_failed,
     rate_limit_headers,
+    refusal_code,
     resolve_dispatch_provider,
     resolve_request_context,
     run_platform_non_stream,
@@ -70,7 +71,7 @@
     mcp_tool_loop_stream,
 )
 from gateway.services.tools import CODE_EXECUTION_HEADER, WEB_SEARCH_HEADER, Dialect, ToolUseBudget
-from gateway.streaming import OPENAI_STREAM_FORMAT, StreamFormat
+from gateway.streaming import OPENAI_STREAM_FORMAT, StreamFormat, openai_error_event
 from gateway.types.attempt import Attempt
 from gateway.types.normalization_target import NormalizationTarget
 from gateway.types.session_principal import SessionPrincipal
@@ -208,7 +209,7 @@ def provider_error(self, exc: BaseException) -> HTTPException:
         )
 
     def stream_error_payload(self, exc: BaseException) -> str:
-        return self.stream_format.error_payload
+        return openai_error_event(self.stream_format, refusal_code(exc))
 
     def format_chunk(self, chunk: ChatCompletionChunk) -> str:
         return f"data: {chunk.model_dump_json()}\n\n"
diff --git a/src/gateway/api/routes/responses.py b/src/gateway/api/routes/responses.py
index 88dcbdd89..cfa4ce9d3 100644
--- a/src/gateway/api/routes/responses.py
+++ b/src/gateway/api/routes/responses.py
@@ -39,6 +39,7 @@
     prepare_gateway_tools,
     provider_error_headers,
     raise_all_streaming_attempts_failed,
+    refusal_code,
     release_reservation,
     resolve_dispatch_provider,
     resolve_request_context,
@@ -73,7 +74,7 @@
     Dialect,
     ToolUseBudget,
 )
-from gateway.streaming import RESPONSES_STREAM_FORMAT, StreamFormat
+from gateway.streaming import RESPONSES_STREAM_FORMAT, StreamFormat, openai_error_event
 from gateway.types.attempt import Attempt
 from gateway.types.normalization_target import NormalizationTarget
 
@@ -347,7 +348,7 @@ def provider_error(self, exc: BaseException) -> HTTPException:
         )
 
     def stream_error_payload(self, exc: BaseException) -> str:
-        return self.stream_format.error_payload
+        return openai_error_event(self.stream_format, refusal_code(exc))
 
     def format_chunk(self, chunk: ResponseStreamEvent) -> str:
         return f"event: {chunk.type}\ndata: {chunk.model_dump_json(exclude_none=True)}\n\n"
diff --git a/src/gateway/core/error_codes.py b/src/gateway/core/error_codes.py
new file mode 100644
index 000000000..9e0cc069b
--- /dev/null
+++ b/src/gateway/core/error_codes.py
@@ -0,0 +1,41 @@
+"""Stable, machine-readable codes for the refusals a caller is expected to act on.
+
+Sent as the ``Otari-Error-Code`` response header and as ``code`` in the error
+body, beside the human-readable ``detail``, so a client maps a refusal by its
+code rather than by matching text that may be reworded. A streamed error event
+carries it as ``error.code``. A code, once sent, keeps its meaning.
+"""
+
+from collections.abc import Mapping
+
+ERROR_CODE_HEADER = "Otari-Error-Code"
+# Which budget refused: ``user`` for the billed user's own budget, otherwise the
+# scope of the ceiling (one of ``models.budgets.ScopeType``).
+BUDGET_SCOPE_HEADER = "Otari-Budget-Scope"
+# The ``rate_limits`` rule a 429 names.
+RATE_LIMIT_RULE_HEADER = "Otari-Rate-Limit-Rule"
+
+BUDGET_EXCEEDED = "budget_exceeded"
+USER_BLOCKED = "user_blocked"
+USER_NOT_FOUND = "user_not_found"
+RATE_LIMITED = "rate_limited"
+UPSTREAM_RATE_LIMITED = "upstream_rate_limited"
+INVALID_MODEL = "invalid_model"
+MODEL_NOT_ALLOWED = "model_not_allowed"
+CONTEXT_LENGTH_EXCEEDED = "context_length_exceeded"
+PRICING_REQUIRED = "pricing_required"
+
+
+def error_headers(code: str, *, budget_scope: str | None = None, rule: str | None = None) -> dict[str, str]:
+    """``Otari-Error-Code`` plus any of the extra headers that have a value."""
+    headers = {ERROR_CODE_HEADER: code}
+    if budget_scope is not None:
+        headers[BUDGET_SCOPE_HEADER] = budget_scope
+    if rule is not None:
+        headers[RATE_LIMIT_RULE_HEADER] = rule
+    return headers
+
+
+def error_code_of(headers: Mapping[str, str] | None) -> str | None:
+    """The code a refusal's headers carry, or None."""
+    return (headers or {}).get(ERROR_CODE_HEADER)
diff --git a/src/gateway/main.py b/src/gateway/main.py
index 6cf8ac5d7..2c9265ab3 100644
--- a/src/gateway/main.py
+++ b/src/gateway/main.py
@@ -7,11 +7,13 @@
 from urllib.parse import urlsplit
 
 from fastapi import FastAPI, Request, Response, status
+from fastapi.exception_handlers import http_exception_handler
 from fastapi.exceptions import RequestValidationError
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse
 from fastapi.routing import APIRoute
 from fastapi.staticfiles import StaticFiles
+from starlette.exceptions import HTTPException as StarletteHTTPException
 from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
 from typing_extensions import override
 
@@ -22,6 +24,7 @@
 from gateway.context_propagation import TraceContextPropagationMiddleware
 from gateway.core.config import API_KEY_HEADER, API_ROOT, GATEWAY_TOKEN_HEADER, X_API_KEY_HEADER, GatewayConfig
 from gateway.core.database import create_session, dispose_db, init_db
+from gateway.core.error_codes import error_code_of
 from gateway.core.feature import Worker
 from gateway.dashboard import DASHBOARD_PACKAGE_PATH, get_dashboard_build_id, get_dashboard_dir
 from gateway.exceptions import TenancyError
@@ -733,6 +736,24 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     return lifespan
 
 
+async def _http_exception_handler(request: Request, exc: Exception) -> Response:
+    """FastAPI's own HTTPException response, with the refusal's ``Otari-Error-Code`` as ``code`` in the body too.
+
+    A body field survives where a header does not (a proxy that drops it, an SDK
+    that surfaces only the body), so a client can map the refusal from either.
+    """
+    if not isinstance(exc, StarletteHTTPException):
+        raise exc
+    code = error_code_of(exc.headers)
+    if code is None:
+        return await http_exception_handler(request, exc)
+    return JSONResponse(
+        {"detail": exc.detail, "code": code},
+        status_code=exc.status_code,
+        headers=exc.headers,
+    )
+
+
 async def _tenancy_error_handler(_: Request, exc: Exception) -> Response:
     """Render a tenancy domain error as the status it carries.
 
@@ -1060,6 +1081,7 @@ async def root_index() -> str:
     install_rate_limits(app, config)
 
     register_routers(app, config)
+    app.add_exception_handler(StarletteHTTPException, _http_exception_handler)
     app.add_exception_handler(TenancyError, _tenancy_error_handler)
     app.add_exception_handler(ControlPlaneError, _control_plane_error_handler)
     app.add_exception_handler(RequestValidationError, _validation_error_handler)
diff --git a/src/gateway/rate_limit.py b/src/gateway/rate_limit.py
index 42606a99b..bb3043678 100644
--- a/src/gateway/rate_limit.py
+++ b/src/gateway/rate_limit.py
@@ -10,6 +10,7 @@
 
 from fastapi import HTTPException, Request, status
 
+from gateway.core.error_codes import RATE_LIMITED, error_headers
 from gateway.log_config import logger
 from gateway.metrics import REGISTRY, Counter
 from gateway.ports.rate_limit_store_port import RateLimitStorePort, RateLimitWindow
@@ -115,7 +116,7 @@ def _info_or_raise(window: RateLimitWindow, limit: int) -> RateLimitInfo:
         raise HTTPException(
             status_code=status.HTTP_429_TOO_MANY_REQUESTS,
             detail="Rate limit exceeded",
-            headers={"Retry-After": str(math.ceil(window.reset_after))},
+            headers={"Retry-After": str(math.ceil(window.reset_after)), **error_headers(RATE_LIMITED)},
         )
     # Wall-clock time for the externally facing reset header.
     return RateLimitInfo(limit=limit, remaining=limit - window.count, reset=time.time() + window.reset_after)
@@ -293,14 +294,13 @@ def _count(n: int, noun: str) -> str:
     return f"{n:,} {noun}" if n == 1 else f"{n:,} {noun}s"
 
 
-def _refused(detail: str, retry_after: float | None) -> HTTPException:
+def _refused(detail: str, retry_after: float | None, rule: str) -> HTTPException:
     """A 429 with ``detail``, without ``Retry-After`` when no wait would let the request in."""
     RATE_LIMIT_HITS.inc()
-    return HTTPException(
-        status_code=status.HTTP_429_TOO_MANY_REQUESTS,
-        detail=detail,
-        headers={"Retry-After": str(max(math.ceil(retry_after), 1))} if retry_after is not None else None,
-    )
+    headers = error_headers(RATE_LIMITED, rule=rule)
+    if retry_after is not None:
+        headers["Retry-After"] = str(max(math.ceil(retry_after), 1))
+    return HTTPException(status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail=detail, headers=headers)
 
 
 async def _count_rule(
@@ -327,6 +327,7 @@ async def _count_rule(
             raise _refused(
                 f"Rate limit '{rule.name}'{label} exceeded: {_count(rule.rpm, 'request')} per minute",
                 window.reset_after,
+                rule.name,
             )
         hold.entries.append((f"{base}:rpm", window.handle))
     if rule.tpm is not None:
@@ -337,12 +338,13 @@ async def _count_rule(
                 f"Request needs an estimated {_count(cost, 'token')}; "
                 f"rate limit '{rule.name}'{label} allows {rule.tpm:,} per minute"
             )
-            raise _refused(msg, None)
+            raise _refused(msg, None, rule.name)
         window = await store.hit(f"{base}:tpm", rule.tpm, _RULE_WINDOW_SEC, cost=cost)
         if window.handle is None:
             raise _refused(
                 f"Rate limit '{rule.name}'{label} exceeded: {_count(rule.tpm, 'token')} per minute",
                 window.reset_after,
+                rule.name,
             )
         hold.entries.append((f"{base}:tpm", window.handle))
         hold.estimates.append((f"{base}:tpm", window.handle))
@@ -352,6 +354,7 @@ async def _count_rule(
             raise _refused(
                 f"Rate limit '{rule.name}'{label} exceeded: {_count(rule.max_concurrent, 'request')} in flight",
                 _CONCURRENCY_RETRY_AFTER_SEC,
+                rule.name,
             )
         hold.leases.append((f"{base}:concurrent", lease))
 
diff --git a/src/gateway/services/budgets/_reservations.py b/src/gateway/services/budgets/_reservations.py
index 26e98877e..5a66f8b10 100644
--- a/src/gateway/services/budgets/_reservations.py
+++ b/src/gateway/services/budgets/_reservations.py
@@ -14,6 +14,7 @@
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from gateway.core import error_codes
 from gateway.core.metered_pricing import estimate_metered_cost
 from gateway.log_config import logger
 from gateway.metrics import REGISTRY, Counter
@@ -507,11 +508,13 @@ async def reserve_budget(
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
             detail=f"User '{user_id}' not found",
+            headers=error_codes.error_headers(error_codes.USER_NOT_FOUND),
         )
     if user.blocked:
         raise HTTPException(
             status_code=status.HTTP_403_FORBIDDEN,
             detail=f"User '{user_id}' is blocked",
+            headers=error_codes.error_headers(error_codes.USER_BLOCKED),
         )
 
     # Budget-exempt request (e.g. a key flagged exclude_from_budget): the user is
@@ -597,6 +600,7 @@ async def reserve_budget(
             raise HTTPException(
                 status_code=status.HTTP_403_FORBIDDEN,
                 detail=f"{refused.subject} has exceeded {axis} limit",
+                headers=error_codes.error_headers(error_codes.BUDGET_EXCEEDED, budget_scope=refused.scope_type),
             )
 
     if budget is None:
@@ -722,6 +726,7 @@ def guards_for(committed: Any, held_amount: Any, cap: Any) -> list[Any]:
         raise HTTPException(
             status_code=status.HTTP_403_FORBIDDEN,
             detail=f"User '{user_id}' has exceeded {axis} limit",
+            headers=error_codes.error_headers(error_codes.BUDGET_EXCEEDED, budget_scope="user"),
         )
 
     return await _held_handle(
@@ -1005,6 +1010,7 @@ async def increase_reservation(
             raise HTTPException(
                 status_code=status.HTTP_403_FORBIDDEN,
                 detail=f"{refused.subject} has exceeded {axis} limit",
+                headers=error_codes.error_headers(error_codes.BUDGET_EXCEEDED, budget_scope=refused.scope_type),
             )
         # Recorded here, next to the hold it describes, and before the per-user
         # call below can refuse: what the caller's refund releases is what the
diff --git a/src/gateway/streaming.py b/src/gateway/streaming.py
index cec8ff1c8..3b43f3977 100644
--- a/src/gateway/streaming.py
+++ b/src/gateway/streaming.py
@@ -66,6 +66,15 @@ class StreamFormat:
     {"type": "error", "error": {"type": "api_error", "message": "An error occurred during streaming"}}
 )
 
+
+def openai_error_event(fmt: StreamFormat, code: str | None) -> str:
+    """``fmt``'s error event, with ``code`` as the OpenAI error object's ``code`` when there is one."""
+    if code is None:
+        return fmt.error_payload
+    error = {"message": "An error occurred during streaming", "type": "server_error", "code": code}
+    return fmt.error_payload.replace(_OPENAI_ERROR, json.dumps({"error": error}))
+
+
 # An SSE comment line: conformant parsers (including the OpenAI SDKs) drop it, so
 # it keeps the socket warm without ever surfacing as content.
 _SSE_COMMENT_KEEPALIVE = ": keepalive\n\n"
diff --git a/tests/integration/test_hybrid_mode_chat.py b/tests/integration/test_hybrid_mode_chat.py
index 688e80ee8..1d0e3291d 100644
--- a/tests/integration/test_hybrid_mode_chat.py
+++ b/tests/integration/test_hybrid_mode_chat.py
@@ -1341,7 +1341,10 @@ async def fake_acompletion(**kwargs: Any) -> Any:
     )
 
     assert response.status_code == 429
-    assert response.json() == {"detail": "All upstream providers rate-limited this request"}
+    assert response.json() == {
+        "detail": "All upstream providers rate-limited this request",
+        "code": "upstream_rate_limited",
+    }
     assert response.headers["Otari-Attempt-ID"] == "att-b"
     # A 429 advances the plan, so both attempts really ran: the aggregate is
     # reached by exhausting the route, not by one attempt failing outright.
diff --git a/tests/integration/test_rate_limit_rules.py b/tests/integration/test_rate_limit_rules.py
index ad430f1d5..9e59e9cba 100644
--- a/tests/integration/test_rate_limit_rules.py
+++ b/tests/integration/test_rate_limit_rules.py
@@ -93,6 +93,7 @@ async def completion(**kwargs: Any) -> ChatCompletion:
     assert bob_status == 200
     assert refused.json()["detail"] == "Rate limit 'keys' exceeded: 2 requests per minute"
     assert "Retry-After" in refused.headers
+    assert refused.headers["Otari-Error-Code"] == "rate_limited"
 
 
 def test_a_completed_request_is_charged_the_tokens_it_used(tpm_client: TestClient) -> None:
diff --git a/tests/integration/test_service_key_end_users.py b/tests/integration/test_service_key_end_users.py
index 393491968..234943fa3 100644
--- a/tests/integration/test_service_key_end_users.py
+++ b/tests/integration/test_service_key_end_users.py
@@ -279,3 +279,18 @@ def test_an_unknown_end_user_budget_is_refused_on_update_and_leaves_the_key_alon
     fetched = client.get(f"{API_ROOT}/keys/{key_id}", headers=master_key_header)
     assert fetched.json()["key_name"] == "svc"
     assert fetched.json()["end_user_budget_id"] is None
+
+
+def test_an_end_user_refused_by_its_budget_gets_a_stable_code(
+    client: TestClient, master_key_header: dict[str, str]
+) -> None:
+    budget_id = _budget(client, master_key_header, request_limit=1)
+    _, headers = _service_key(client, master_key_header, "svc-code", end_user_budget_id=budget_id)
+
+    assert _chat(client, headers, "carol").status_code == 200
+    refused = _chat(client, headers, "carol")
+
+    assert refused.status_code == 403
+    assert refused.headers["Otari-Error-Code"] == "budget_exceeded"
+    assert refused.headers["Otari-Budget-Scope"] == "user"
+    assert refused.json()["code"] == "budget_exceeded"
diff --git a/tests/unit/test_pipeline_failover_topup.py b/tests/unit/test_pipeline_failover_topup.py
index 5aa4ac24e..49bbffa42 100644
--- a/tests/unit/test_pipeline_failover_topup.py
+++ b/tests/unit/test_pipeline_failover_topup.py
@@ -27,6 +27,7 @@
 
 import gateway.api.routes._pipeline as pipeline
 from gateway.core.config import GatewayConfig
+from gateway.core.error_codes import BUDGET_EXCEEDED, error_headers
 from gateway.services.budgets import ReservationHandle
 from gateway.types.attempt import Attempt
 
@@ -157,6 +158,25 @@ async def refuse(*_args: Any, **_kwargs: Any) -> None:
     assert "failing over" in str(exc_info.value.detail)
 
 
+@pytest.mark.asyncio
+async def test_a_refused_top_up_keeps_the_budget_code(
+    monkeypatch: pytest.MonkeyPatch, increases: list[Decimal]
+) -> None:
+    """The detail is replaced, but the code and scope a client maps the refusal by are not."""
+    refusal_headers = error_headers(BUDGET_EXCEEDED, budget_scope="user")
+
+    async def refuse(*_args: Any, **_kwargs: Any) -> None:
+        raise HTTPException(status_code=403, detail="budget exceeded", headers=refusal_headers)
+
+    monkeypatch.setattr(pipeline, "increase_reservation", refuse)
+
+    with pytest.raises(HTTPException) as exc_info:
+        await pipeline.top_up_reservation_for_attempt(_ctx(estimate=Decimal(5)), _attempt(2, "anthropic", "pricey"))
+
+    assert exc_info.value.detail == pipeline.budget_exhausted_mid_failover_detail()
+    assert exc_info.value.headers == refusal_headers
+
+
 @pytest.mark.asyncio
 async def test_an_unpriced_fallback_is_refused_under_require_pricing(increases: list[Decimal]) -> None:
     """Otherwise a model that 402s when named directly serves for free by being
diff --git a/tests/unit/test_provider_error_classification.py b/tests/unit/test_provider_error_classification.py
index eb5bc16b6..3a21493a3 100644
--- a/tests/unit/test_provider_error_classification.py
+++ b/tests/unit/test_provider_error_classification.py
@@ -13,11 +13,12 @@
 """
 
 import asyncio
+import json
 
 import httpx
 import pytest
 from anthropic import APITimeoutError as AnthropicAPITimeoutError
-from any_llm.exceptions import InvalidRequestError, UnsupportedParameterError
+from any_llm.exceptions import ContextLengthExceededError, InvalidRequestError, UnsupportedParameterError
 from openai import APITimeoutError as OpenAIAPITimeoutError
 
 from gateway.api.routes._pipeline import (
@@ -31,11 +32,13 @@
     classify_provider_error,
     failure_status_code,
     provider_error_headers,
+    refusal_code,
 )
 from gateway.api.routes._platform import _provider_failure_http_exc, upstream_retry_after
 from gateway.api.routes._schema_derive import SENSITIVE_PARAM_FIELDS
 from gateway.services.mcp_loop import MaxToolIterationsExceeded
 from gateway.services.upstream_redaction import MAX_EXPOSED_DETAIL_CHARS, redact_upstream_message
+from gateway.streaming import OPENAI_STREAM_FORMAT, openai_error_event
 
 _RAW = "raw provider detail SECRET token=abc123"
 
@@ -251,19 +254,28 @@ def _rate_limited_with(retry_after: str) -> Exception:
 
 
 def test_retry_after_is_forwarded_on_a_429() -> None:
-    assert provider_error_headers(_rate_limited_with("34"), 429) == {"Retry-After": "34"}
+    assert provider_error_headers(_rate_limited_with("34"), 429) == {
+        "Otari-Error-Code": "upstream_rate_limited",
+        "Retry-After": "34",
+    }
 
 
 def test_retry_after_rounds_a_fraction_up() -> None:
     """A client honoring the header must not retry before the window the
     provider named, so 0.4s becomes 1s rather than 0s."""
-    assert provider_error_headers(_rate_limited_with("0.4"), 429) == {"Retry-After": "1"}
+    assert provider_error_headers(_rate_limited_with("0.4"), 429) == {
+        "Otari-Error-Code": "upstream_rate_limited",
+        "Retry-After": "1",
+    }
 
 
 def test_retry_after_is_clamped() -> None:
     """A provider does not get to tell this gateway's callers to sleep for a
     year."""
-    assert provider_error_headers(_rate_limited_with("99999999"), 429) == {"Retry-After": "86400"}
+    assert provider_error_headers(_rate_limited_with("99999999"), 429) == {
+        "Otari-Error-Code": "upstream_rate_limited",
+        "Retry-After": "86400",
+    }
 
 
 @pytest.mark.parametrize(
@@ -286,7 +298,7 @@ def test_retry_after_that_is_not_a_number_is_dropped(raw: str) -> None:
     """The value is re-serialized from a parsed number, never relayed as
     received: a header value is not a body, and CRLF in one is not a formatting
     problem."""
-    assert provider_error_headers(_rate_limited_with(raw), 429) is None
+    assert provider_error_headers(_rate_limited_with(raw), 429) == {"Otari-Error-Code": "upstream_rate_limited"}
 
 
 def test_retry_after_is_not_forwarded_on_a_gateway_fault() -> None:
@@ -297,8 +309,8 @@ def test_retry_after_is_not_forwarded_on_a_gateway_fault() -> None:
     assert provider_error_headers(exc, 502) is None
 
 
-def test_retry_after_absent_sends_no_header() -> None:
-    assert provider_error_headers(_StatusError(429), 429) is None
+def test_retry_after_absent_sends_only_the_error_code() -> None:
+    assert provider_error_headers(_StatusError(429), 429) == {"Otari-Error-Code": "upstream_rate_limited"}
 
 
 def test_retry_after_read_through_the_exception_chain() -> None:
@@ -310,7 +322,7 @@ def test_retry_after_read_through_the_exception_chain() -> None:
 def test_platform_terminal_exc_forwards_retry_after() -> None:
     exc = _provider_failure_http_exc(_rate_limited_with("34"), fallback_detail="LLM provider error")
     assert exc.status_code == 429
-    assert exc.headers == {"Retry-After": "34"}
+    assert exc.headers == {"Otari-Error-Code": "upstream_rate_limited", "Retry-After": "34"}
 
 
 @pytest.mark.parametrize("exc", [_StatusError(500), _StatusError(503), Exception(_RAW), ValueError(_RAW)])
@@ -807,3 +819,22 @@ def test_invalid_request_error_with_status_is_classified_by_status() -> None:
     mapping = classify_provider_error(exc)
     assert mapping is not None
     assert mapping.status_code == 400
+
+
+def test_a_prompt_too_long_for_the_model_has_its_own_code() -> None:
+    exc = ContextLengthExceededError("prompt is too long", status_code=400)
+
+    mapping = classify_provider_error(exc)
+
+    assert mapping is not None
+    assert mapping.status_code == 400
+    assert provider_error_headers(exc, 400) == {"Otari-Error-Code": "context_length_exceeded"}
+    assert refusal_code(exc) == "context_length_exceeded"
+
+
+def test_a_stream_error_event_carries_the_code_that_ended_it() -> None:
+    event = openai_error_event(OPENAI_STREAM_FORMAT, refusal_code(_rate_limited_with("3")))
+
+    payload = json.loads(event.removeprefix("data: "))
+    assert payload["error"]["code"] == "upstream_rate_limited"
+    assert openai_error_event(OPENAI_STREAM_FORMAT, None) == OPENAI_STREAM_FORMAT.error_payload
diff --git a/tests/unit/test_rate_limit_rules.py b/tests/unit/test_rate_limit_rules.py
index d7939e3d0..c4fd81763 100644
--- a/tests/unit/test_rate_limit_rules.py
+++ b/tests/unit/test_rate_limit_rules.py
@@ -126,7 +126,8 @@ async def test_a_request_larger_than_the_limit_is_not_told_to_retry() -> None:
     with pytest.raises(HTTPException) as exc_info:
         await _admit(rules, tokens=1001)
 
-    assert exc_info.value.headers is None
+    assert exc_info.value.headers is not None
+    assert "Retry-After" not in exc_info.value.headers
     assert exc_info.value.detail == "Request needs an estimated 1,001 tokens; rate limit 'tpm' allows 1,000 per minute"
 
 
@@ -150,7 +151,11 @@ async def test_a_full_concurrency_limit_refuses_until_a_slot_is_given_back() ->
     grant = await _admit(rules)
     with pytest.raises(HTTPException) as exc_info:
         await _admit(rules)
-    assert exc_info.value.headers == {"Retry-After": "1"}
+    assert exc_info.value.headers == {
+        "Retry-After": "1",
+        "Otari-Error-Code": "rate_limited",
+        "Otari-Rate-Limit-Rule": "inflight",
+    }
     assert exc_info.value.detail == "Rate limit 'inflight' exceeded: 1 request in flight"
 
     await grant.release()