diff --git a/tests/rl/test_session_server_endpoints.py b/tests/rl/test_session_server_endpoints.py
new file mode 100644
index 000000000..b12609656
--- /dev/null
+++ b/tests/rl/test_session_server_endpoints.py
@@ -0,0 +1,251 @@
+import json
+import unittest
+from unittest.mock import AsyncMock
+
+from aiohttp import web
+from aiohttp.test_utils import TestClient, TestServer
+
+from xtuner.v1.rl.rollout.session_server import (
+    FMT_ANTHROPIC,
+    FMT_OPENAI,
+    SessionServer,
+    _detect_format,
+    _is_generation_endpoint,
+)
+
+
+class TestEndpointClassification(unittest.TestCase):
+    def test_only_generation_endpoints_are_classified(self):
+        cases = [
+            ("POST", "v1/messages"),
+            ("POST", "/proxy/v1/messages?beta=true"),
+            ("POST", "/v1/chat/completions"),
+        ]
+        for method, path in cases:
+            with self.subTest(method=method, path=path):
+                self.assertTrue(_is_generation_endpoint(method, path))
+
+    def test_non_generation_paths_and_wrong_methods_are_not_classified(self):
+        cases = [
+            ("POST", "v1/messages/count_tokens"),
+            ("POST", "prefix/v1/responses"),
+            ("HEAD", "/api/hello"),
+            ("GET", "/v1/models"),
+            ("GET", "/v1/messages"),
+            ("POST", "/terminate"),
+            ("POST", "/update_weights"),
+            ("POST", "/sleep"),
+            ("POST", "/wakeup"),
+        ]
+        for method, path in cases:
+            with self.subTest(method=method, path=path):
+                self.assertFalse(_is_generation_endpoint(method, path))
+
+    def test_count_tokens_uses_anthropic_format(self):
+        self.assertEqual(_detect_format("v1/messages/count_tokens"), FMT_ANTHROPIC)
+        self.assertEqual(_detect_format("v1/messages/batches"), FMT_ANTHROPIC)
+        self.assertEqual(_detect_format("v1/chat/completions"), FMT_OPENAI)
+
+
+class TestSessionServerEndpointHandling(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.upstream_calls = []
+        self.upstream_body = b'{"input_tokens": 42}'
+        self.upstream_content_type = "application/json"
+
+        async def upstream_handler(request):
+            self.upstream_calls.append(
+                {
+                    "method": request.method,
+                    "path": request.path,
+                    "query": request.query_string,
+                    "headers": dict(request.headers),
+                    "body": await request.read(),
+                }
+            )
+            return web.Response(
+                status=200,
+                content_type=self.upstream_content_type,
+                headers={"X-Upstream-Header": "preserved"},
+                body=self.upstream_body,
+            )
+
+        upstream_app = web.Application()
+        upstream_app.router.add_route("*", "/{path:.*}", upstream_handler)
+        self.upstream_server = TestServer(upstream_app)
+        await self.upstream_server.start_server()
+
+        self.session_server = SessionServer.__new__(SessionServer)
+        self.session_server.worker_base_url = str(self.upstream_server.make_url("")).rstrip("/")
+        self.session_server.request_timeout = 10.0
+        self.session_server.read_bufsize = 2**20
+        self.session_server.stop_word = "<eos>"
+
+        self.session_server.on_request = AsyncMock(side_effect=lambda body, _fmt, **_kwargs: body)
+        self.session_server.on_response = AsyncMock(return_value=None)
+
+        proxy_app = web.Application()
+        proxy_app.router.add_route("*", "/{path:.*}", self.session_server._handle_request)
+        self.proxy_client = TestClient(TestServer(proxy_app))
+        await self.proxy_client.start_server()
+
+    async def asyncTearDown(self):
+        await self.proxy_client.close()
+        await self.upstream_server.close()
+
+    async def test_count_tokens_is_transparent_and_untraced(self):
+        payload = {
+            "model": "test-model",
+            "messages": [{"role": "user", "content": "hello"}],
+            "system": "system",
+            "tools": [
+                {"name": "tool", "input_schema": {"type": "object"}},
+                {"type": "web_search_20250305", "name": "web_search"},
+            ],
+            "session_id": "trace-session",
+            "return_token_ids": True,
+            "return_logprob": True,
+        }
+
+        response = await self.proxy_client.post("/v1/messages/count_tokens?beta=true", json=payload)
+
+        self.assertEqual(response.status, 200)
+        self.assertEqual(response.headers["X-Upstream-Header"], "preserved")
+        self.assertEqual(await response.read(), b'{"input_tokens": 42}')
+        self.assertEqual(len(self.upstream_calls), 1)
+        self.assertEqual(self.upstream_calls[0]["headers"]["anthropic-version"], "2023-06-01")
+        forwarded = json.loads(self.upstream_calls[0]["body"])
+        self.assertEqual(forwarded, {key: value for key, value in payload.items() if key != "session_id"})
+        self.assertEqual(self.upstream_calls[0]["query"], "beta=true")
+        self.session_server.on_request.assert_not_awaited()
+        self.session_server.on_response.assert_not_awaited()
+
+    async def test_generation_still_runs_both_hooks(self):
+        self.upstream_body = json.dumps(
+            {
+                "choices": [
+                    {
+                        "message": {"role": "assistant", "content": "ok"},
+                        "output_ids": [1],
+                        "output_token_logprobs": [[-0.1, 1]],
+                    }
+                ]
+            }
+        ).encode()
+        payload = {
+            "model": "test-model",
+            "messages": [{"role": "user", "content": "hello"}],
+            "session_id": "trace-session",
+            "return_token_ids": True,
+            "return_logprob": True,
+        }
+
+        response = await self.proxy_client.post("/v1/chat/completions", json=payload)
+
+        self.assertEqual(response.status, 200)
+        self.session_server.on_request.assert_awaited_once()
+        self.assertTrue(self.session_server.on_request.await_args.kwargs["trace_enabled"])
+        self.session_server.on_response.assert_awaited_once()
+
+    async def test_generation_evaluation_request_runs_only_request_hook(self):
+        self.upstream_body = b'{"choices":[{"message":{"role":"assistant","content":"ok"}}]}'
+        payload = {
+            "model": "test-model",
+            "messages": [{"role": "user", "content": "hello"}],
+            "session_id": "evaluation-session",
+            "return_token_ids": False,
+        }
+
+        response = await self.proxy_client.post("/v1/chat/completions", json=payload)
+
+        self.assertEqual(response.status, 200)
+        self.session_server.on_request.assert_awaited_once()
+        self.assertFalse(self.session_server.on_request.await_args.kwargs["trace_enabled"])
+        self.session_server.on_response.assert_not_awaited()
+
+    async def test_non_generation_endpoints_reach_upstream_without_trace(self):
+        hello = await self.proxy_client.head("/api/hello")
+        unknown = await self.proxy_client.post("/terminate", json={})
+        wrong_method = await self.proxy_client.get("/v1/messages")
+
+        self.assertEqual(hello.status, 200)
+        self.assertEqual(unknown.status, 200)
+        self.assertEqual(wrong_method.status, 200)
+        self.assertEqual([call["path"] for call in self.upstream_calls], ["/api/hello", "/terminate", "/v1/messages"])
+        self.session_server.on_request.assert_not_awaited()
+        self.session_server.on_response.assert_not_awaited()
+
+    async def test_start_registers_handle_request_directly(self):
+        self.session_server.host = "127.0.0.1"
+        self.session_server.port = 0
+        self.session_server._site = None
+        self.session_server._runner = None
+        self.session_server._app = None
+
+        await self.session_server.start()
+        try:
+            route = next(route for route in self.session_server._app.router.routes() if route.method == "*")
+            self.assertIs(route.handler.__func__, SessionServer._handle_request)
+        finally:
+            await self.session_server.stop()
+
+    async def test_non_generation_stream_is_forwarded_without_hooks(self):
+        self.upstream_content_type = "text/event-stream"
+        self.upstream_body = (
+            b"event: ping"
+            + bytes([10])
+            + b'data: {"choices":[{"delta":{"content":"hello<eos>"}}],"output_ids":[9]}'
+            + bytes([10, 10])
+            + b"data: [DONE]"
+            + bytes([10, 10])
+        )
+
+        response = await self.proxy_client.post("/future/stream", json={"stream": True, "session_id": "old-client"})
+
+        self.assertEqual(response.status, 200)
+        self.assertEqual(await response.read(), self.upstream_body)
+        self.assertEqual(json.loads(self.upstream_calls[0]["body"]), {"stream": True})
+        self.session_server.on_request.assert_not_awaited()
+        self.session_server.on_response.assert_not_awaited()
+
+    async def test_non_generation_non_object_body_is_forwarded_unchanged(self):
+        body = b'[{"session_id":"should-stay-in-array"}]'
+
+        response = await self.proxy_client.post("/future", data=body, headers={"Content-Type": "application/json"})
+
+        self.assertEqual(response.status, 200)
+        self.assertEqual(self.upstream_calls[0]["body"], body)
+        self.session_server.on_request.assert_not_awaited()
+        self.session_server.on_response.assert_not_awaited()
+
+    async def test_non_generation_object_without_session_id_preserves_body_bytes(self):
+        body = b' { "stream": false, "value": [1, 2] } '
+
+        response = await self.proxy_client.post("/future", data=body, headers={"Content-Type": "application/json"})
+
+        self.assertEqual(response.status, 200)
+        self.assertEqual(self.upstream_calls[0]["body"], body)
+        self.session_server.on_request.assert_not_awaited()
+        self.session_server.on_response.assert_not_awaited()
+
+    async def test_non_generation_json_does_not_inject_return_fields(self):
+        payload = {"messages": [{"role": "user", "content": "hello"}]}
+
+        response = await self.proxy_client.post("/future", json=payload)
+
+        self.assertEqual(response.status, 200)
+        forwarded = json.loads(self.upstream_calls[0]["body"])
+        self.assertEqual(forwarded, payload)
+        self.assertFalse(any(key.startswith("return_") for key in forwarded))
+        self.session_server.on_request.assert_not_awaited()
+        self.session_server.on_response.assert_not_awaited()
+
+    async def test_responses_keeps_preexisting_rejection(self):
+        responses = await self.proxy_client.post("/v1/responses", json={})
+
+        self.assertEqual(responses.status, 501)
+        self.assertEqual(self.upstream_calls, [])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/rl/test_session_server_trace.py b/tests/rl/test_session_server_trace.py
new file mode 100644
index 000000000..53b44bea4
--- /dev/null
+++ b/tests/rl/test_session_server_trace.py
@@ -0,0 +1,289 @@
+import json
+import unittest
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+from aiohttp import ClientConnectionResetError, web
+from aiohttp.test_utils import TestClient, TestServer
+
+from xtuner.v1.rl.rollout import session_server as session_server_module
+from xtuner.v1.rl.rollout.session_server import SessionServer
+
+
+class TestSessionServerTraceHandling(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.upstream_status = 200
+        self.upstream_body = b'{"choices":[{"message":{"role":"assistant","content":"ok"}}]}'
+        self.upstream_content_type = "application/json"
+        self.upstream_headers = {"X-Upstream-Header": "preserved"}
+        self.upstream_calls = 0
+
+        async def upstream_handler(_request):
+            self.upstream_calls += 1
+            return web.Response(
+                status=self.upstream_status,
+                content_type=self.upstream_content_type,
+                headers=self.upstream_headers,
+                body=self.upstream_body,
+            )
+
+        upstream_app = web.Application()
+        upstream_app.router.add_route("*", "/{path:.*}", upstream_handler)
+        self.upstream_server = TestServer(upstream_app)
+        await self.upstream_server.start_server()
+
+        self.session_server = SessionServer.__new__(SessionServer)
+        self.session_server.worker_base_url = str(self.upstream_server.make_url("")).rstrip("/")
+        self.session_server.request_timeout = 10.0
+        self.session_server.read_bufsize = 2**20
+        self.session_server.stop_word = "<eos>"
+        self.session_server.on_request = AsyncMock(side_effect=lambda body, _fmt, **_kwargs: body)
+        self.session_server.on_response = AsyncMock(return_value=None)
+
+        proxy_app = web.Application()
+        proxy_app.router.add_route("*", "/{path:.*}", self.session_server._handle_request)
+        self.proxy_client = TestClient(TestServer(proxy_app))
+        await self.proxy_client.start_server()
+
+    async def asyncTearDown(self):
+        await self.proxy_client.close()
+        await self.upstream_server.close()
+
+    @staticmethod
+    def _request_payload(stream=False):
+        return {
+            "model": "test-model",
+            "messages": [{"role": "user", "content": "hello"}],
+            "session_id": "trace-session",
+            "return_token_ids": True,
+            "stream": stream,
+        }
+
+    @staticmethod
+    def _openai_stream(*, include_done=True):
+        body = (
+            b'data: {"id":"completion","choices":[{"delta":{"content":"ok"},'
+            b'"output_ids":[1],"output_token_logprobs":[[-0.1,1]],"finish_reason":"stop"}]}\n\n'
+        )
+        if include_done:
+            body += b"data: [DONE]\n\n"
+        return body
+
+    async def test_successful_training_response_runs_response_hook(self):
+        self.upstream_body = json.dumps(
+            {
+                "choices": [
+                    {
+                        "message": {"role": "assistant", "content": "ok"},
+                        "output_ids": [1],
+                        "output_token_logprobs": [[-0.1, 1]],
+                    }
+                ]
+            }
+        ).encode()
+
+        response = await self.proxy_client.post("/v1/chat/completions", json=self._request_payload())
+
+        self.assertEqual(response.status, 200)
+        self.session_server.on_request.assert_awaited_once()
+        self.session_server.on_response.assert_awaited_once()
+
+    async def test_successful_training_stream_sends_done_after_trace(self):
+        self.upstream_content_type = "text/event-stream"
+        self.upstream_body = self._openai_stream()
+
+        response = await self.proxy_client.post("/v1/chat/completions", json=self._request_payload(stream=True))
+
+        self.assertEqual(response.status, 200)
+        body = await response.read()
+        self.assertIn(b'"content": "ok"', body)
+        self.assertIn(b"data: [DONE]\n\n", body)
+        self.session_server.on_response.assert_awaited_once()
+
+    async def test_upstream_error_json_is_returned_unchanged(self):
+        self.upstream_body = b'{"object":"error","message":"upstream failed"}'
+
+        response = await self.proxy_client.post("/v1/chat/completions", json=self._request_payload())
+
+        self.assertEqual(response.status, 200)
+        self.assertEqual(await response.read(), self.upstream_body)
+        self.session_server.on_response.assert_not_awaited()
+
+    async def test_upstream_error_stream_is_not_wrapped_again(self):
+        self.upstream_content_type = "text/event-stream"
+        self.upstream_body = b'data: {"error":{"type":"server_error","message":"upstream failed"}}\n\ndata: [DONE]\n\n'
+
+        response = await self.proxy_client.post("/v1/chat/completions", json=self._request_payload(stream=True))
+        body = await response.read()
+
+        self.assertEqual(response.status, 200)
+        self.assertEqual(body, self.upstream_body)
+        self.assertNotIn(b"SessionServer", body)
+        self.session_server.on_response.assert_not_awaited()
+
+    async def test_non_success_json_is_forwarded_without_cleaning_or_hook(self):
+        self.upstream_status = 400
+        self.upstream_body = json.dumps(
+            {
+                "choices": [{"message": {"role": "assistant", "content": "bad<eos>"}, "output_ids": [1]}],
+                "routed_experts": [[0]],
+            }
+        ).encode()
+
+        response = await self.proxy_client.post("/v1/chat/completions", json=self._request_payload())
+
+        self.assertEqual(response.status, 400)
+        self.assertEqual(await response.read(), self.upstream_body)
+        self.assertEqual(response.headers["X-Upstream-Header"], "preserved")
+        self.session_server.on_response.assert_not_awaited()
+
+    async def test_non_success_stream_is_forwarded_byte_for_byte(self):
+        self.upstream_status = 500
+        self.upstream_content_type = "text/event-stream"
+        self.upstream_body = (
+            b'data: {"choices":[{"delta":{"content":"bad<eos>"},"output_ids":[1]}]}\n\ndata: [DONE]\n\n'
+        )
+
+        response = await self.proxy_client.post("/v1/chat/completions", json=self._request_payload(stream=True))
+
+        self.assertEqual(response.status, 500)
+        self.assertEqual(await response.read(), self.upstream_body)
+        self.session_server.on_response.assert_not_awaited()
+
+    async def test_malformed_or_non_object_success_response_fails_closed(self):
+        for body in (b"not-json", b"[]"):
+            with self.subTest(body=body):
+                self.upstream_status = 200
+                self.upstream_content_type = "application/json"
+                self.upstream_body = body
+                self.session_server.on_response.reset_mock()
+
+                response = await self.proxy_client.post("/v1/chat/completions", json=self._request_payload())
+                error = await response.json()
+
+                self.assertEqual(response.status, 500)
+                self.assertEqual(error["object"], "error")
+                self.session_server.on_response.assert_not_awaited()
+
+    async def test_response_hook_failure_returns_native_error(self):
+        self.session_server.on_response = AsyncMock(side_effect=RuntimeError("missing routed experts"))
+
+        response = await self.proxy_client.post("/v1/chat/completions", json=self._request_payload())
+        error = await response.json()
+
+        self.assertEqual(response.status, 500)
+        self.assertEqual(error["object"], "error")
+        self.assertIn("missing routed experts", error["message"])
+        self.session_server.on_response.assert_awaited_once()
+
+    async def test_response_hook_failure_stream_sends_error_without_done(self):
+        self.upstream_content_type = "text/event-stream"
+        self.upstream_body = self._openai_stream()
+        self.session_server.on_response = AsyncMock(side_effect=RuntimeError("trace write failed"))
+
+        response = await self.proxy_client.post("/v1/chat/completions", json=self._request_payload(stream=True))
+        body = await response.read()
+
+        self.assertEqual(response.status, 200)
+        self.assertIn(b'"object": "error"', body)
+        self.assertNotIn(b"data: [DONE]", body)
+        self.session_server.on_response.assert_awaited_once()
+
+    async def test_incomplete_stream_sends_error_without_done(self):
+        self.upstream_content_type = "text/event-stream"
+        self.upstream_body = self._openai_stream(include_done=False)
+
+        response = await self.proxy_client.post("/v1/chat/completions", json=self._request_payload(stream=True))
+        body = await response.read()
+
+        self.assertEqual(response.status, 200)
+        self.assertIn(b'"object": "error"', body)
+        self.assertNotIn(b"data: [DONE]", body)
+        self.session_server.on_response.assert_not_awaited()
+
+    async def test_disconnect_before_prepare_still_completes_trace(self):
+        self.upstream_content_type = "text/event-stream"
+        self.upstream_body = self._openai_stream()
+
+        class Request:
+            is_proxy = True
+            method = "POST"
+            match_info = {"path": "v1/chat/completions"}
+            query_string = ""
+            headers = {"Content-Type": "application/json"}
+
+            async def read(self):
+                return json.dumps(TestSessionServerTraceHandling._request_payload(stream=True)).encode()
+
+        class FakeStreamResponse:
+            def __init__(self, status, headers):
+                self.status = status
+                self.headers = headers
+                self.prepare_calls = []
+
+            async def prepare(self, request):
+                self.prepare_calls.append(request)
+                raise ClientConnectionResetError("downstream closed")
+
+            async def write_eof(self):
+                pass
+
+        fake_web = SimpleNamespace(
+            StreamResponse=FakeStreamResponse,
+            Response=web.Response,
+            json_response=web.json_response,
+        )
+        with patch.object(session_server_module, "web", fake_web):
+            response = await self.session_server._handle_request(Request())
+
+        self.assertEqual(response.status, 200)
+        self.assertEqual(len(response.prepare_calls), 1)
+        self.session_server.on_response.assert_awaited_once()
+
+    async def test_disconnect_midstream_still_completes_trace(self):
+        self.upstream_content_type = "text/event-stream"
+        self.upstream_body = self._openai_stream()
+
+        class Request:
+            is_proxy = True
+            method = "POST"
+            match_info = {"path": "v1/chat/completions"}
+            query_string = ""
+            headers = {"Content-Type": "application/json"}
+
+            async def read(self):
+                return json.dumps(TestSessionServerTraceHandling._request_payload(stream=True)).encode()
+
+        class FakeStreamResponse:
+            def __init__(self, status, headers):
+                self.status = status
+                self.headers = headers
+                self.prepare_calls = []
+                self.write_calls = []
+
+            async def prepare(self, request):
+                self.prepare_calls.append(request)
+
+            async def write(self, data):
+                self.write_calls.append(data)
+                raise ClientConnectionResetError("downstream closed")
+
+            async def write_eof(self):
+                pass
+
+        fake_web = SimpleNamespace(
+            StreamResponse=FakeStreamResponse,
+            Response=web.Response,
+            json_response=web.json_response,
+        )
+        with patch.object(session_server_module, "web", fake_web):
+            response = await self.session_server._handle_request(Request())
+
+        self.assertEqual(response.status, 200)
+        self.assertEqual(len(response.prepare_calls), 1)
+        self.assertEqual(len(response.write_calls), 1)
+        self.session_server.on_response.assert_awaited_once()
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/xtuner/v1/rl/rollout/session_server.py b/xtuner/v1/rl/rollout/session_server.py
index 9914d734a..09dfda3c4 100644
--- a/xtuner/v1/rl/rollout/session_server.py
+++ b/xtuner/v1/rl/rollout/session_server.py
@@ -23,8 +23,20 @@
 _SESSION_SERVER_ONLY_KEYS = {"session_id"}
 
 
+def _normalize_path(req_path: str) -> str:
+    """Return a leading-slash path without a query string."""
+    return "/" + req_path.split("?", 1)[0].lstrip("/")
+
+
+def _is_generation_endpoint(method: str, req_path: str) -> bool:
+    """Whether this request uses the SessionServer generation hooks."""
+    path = _normalize_path(req_path)
+    return method.upper() == "POST" and (path.endswith("/v1/messages") or path.endswith("/v1/chat/completions"))
+
+
 def _detect_format(req_path: str) -> str:
-    if req_path.endswith("/messages") or "/v1/messages" in req_path:
+    path = _normalize_path(req_path)
+    if path.endswith("/messages") or "/v1/messages" in path:
         return FMT_ANTHROPIC
     return FMT_OPENAI
 
@@ -33,6 +45,10 @@ def _is_error_payload(payload: dict) -> bool:
     return payload.get("error") is not None or payload.get("type") == "error" or payload.get("object") == "error"
 
 
+class _UpstreamResponseError(RuntimeError):
+    """Internal marker for an error explicitly returned by the upstream."""
+
+
 def _error_payload(fmt: str, message: str, status: int = 500, error_type: str = "internal_server_error") -> dict:
     """Error body in the caller's native shape.
 
@@ -533,6 +549,7 @@ async def _handle_request(self, request: web.Request) -> web.Response:
         """Proxy handler: detect format, run hooks, forward, stream back."""
 
         req_path = request.match_info["path"]
+        is_generation_endpoint = _is_generation_endpoint(request.method, req_path)
         fmt = _detect_format(req_path)
 
         # Reject /v1/responses outright — upstream worker doesn't implement it.
@@ -554,40 +571,62 @@ async def _handle_request(self, request: web.Request) -> web.Response:
         trace_enabled = False
         orig_return_logprob = orig_return_token_ids = False
         orig_return_routed_experts = True
-        if request_body:
+        if request_body and not is_generation_endpoint:
+            # Only inspect object-shaped JSON. Lists, scalars and malformed
+            # bodies stay byte-for-byte untouched for the upstream validator.
+            if request_body.lstrip().startswith(b"{"):
+                try:
+                    parsed_request_data = json.loads(request_body)
+                except (json.JSONDecodeError, UnicodeDecodeError):
+                    pass
+                else:
+                    if isinstance(parsed_request_data, dict):
+                        request_data = parsed_request_data
+                        if _SESSION_SERVER_ONLY_KEYS.intersection(request_data):
+                            request_data = {
+                                key: value
+                                for key, value in request_data.items()
+                                if key not in _SESSION_SERVER_ONLY_KEYS
+                            }
+                            request_body = json.dumps(request_data).encode("utf-8")
+        elif request_body:
             try:
-                request_data = json.loads(request_body)
-                # Anthropic server-side built-ins (web_search_*, computer_*, ...)
-                # don't carry ``input_schema`` and the lmdeploy worker can't
-                # route them anyway — drop them before anything downstream
-                # (orig_req_body / on_request / wire forward) sees them.
-                if fmt == FMT_ANTHROPIC and "tools" in request_data:
-                    filtered = _filter_anthropic_user_tools(request_data["tools"])
-                    if filtered:
-                        request_data["tools"] = filtered
-                    else:
-                        request_data.pop("tools", None)
-                orig_req_body = copy.deepcopy(request_data)
-
-                trace_enabled = _request_uses_trace_store(request_data)
-                # Accept either ``return_logprob`` (canonical) or the legacy
-                # ``logprobs`` alias when deciding whether the client wanted
-                # logprobs forwarded back.
-                orig_return_logprob = _bool_request_value(
-                    request_data.get("return_logprob", request_data.get("logprobs")), False
-                )
-                orig_return_token_ids = _bool_request_value(request_data.get("return_token_ids"), False)
-                orig_return_routed_experts = _bool_request_value(request_data.get("return_routed_experts"), True)
-
-                request_data = await self.on_request(request_data, fmt, trace_enabled=trace_enabled)
-                request_body = json.dumps(request_data).encode("utf-8")
-            except json.JSONDecodeError:
+                parsed_request_data = json.loads(request_body)
+                if isinstance(parsed_request_data, dict):
+                    request_data = parsed_request_data
+                    # Anthropic server-side built-ins (web_search_*, computer_, ...)
+                    # don't carry ``input_schema`` and the lmdeploy worker can't
+                    # route them anyway — drop them before anything downstream
+                    # (orig_req_body / on_request / wire forward) sees them.
+                    if fmt == FMT_ANTHROPIC and "tools" in request_data:
+                        filtered = _filter_anthropic_user_tools(request_data["tools"])
+                        if filtered:
+                            request_data["tools"] = filtered
+                        else:
+                            request_data.pop("tools", None)
+                    orig_req_body = copy.deepcopy(request_data)
+
+                    trace_enabled = _request_uses_trace_store(request_data)
+                    # Accept either ``return_logprob`` (canonical) or the legacy
+                    # ``logprobs`` alias when deciding whether the client wanted
+                    # logprobs forwarded back.
+                    orig_return_logprob = _bool_request_value(
+                        request_data.get("return_logprob", request_data.get("logprobs")), False
+                    )
+                    orig_return_token_ids = _bool_request_value(request_data.get("return_token_ids"), False)
+                    orig_return_routed_experts = _bool_request_value(request_data.get("return_routed_experts"), True)
+
+                    request_data = await self.on_request(request_data, fmt, trace_enabled=trace_enabled)
+                    request_body = json.dumps(request_data).encode("utf-8")
+            except (json.JSONDecodeError, UnicodeDecodeError):
                 pass
             except Exception as exc:
                 message = f"SessionServer request hook failed: {type(exc).__name__}: {exc}"
                 get_logger().error(message)
                 return web.json_response(_error_payload(fmt, message), status=500)
 
+        trace_active = is_generation_endpoint and trace_enabled
+
         # Build forwarding headers, dropping original Host / Content-Length.
         forward_headers = dict(request.headers)
         forward_headers.pop("Host", None)
@@ -607,11 +646,15 @@ async def _handle_request(self, request: web.Request) -> web.Response:
         # Build the per-format stream cleaner (strips lmdeploy-injected
         # extension fields that the client didn't ask for, and removes the
         # stop word from any user-visible text).
-        clean_data = self._build_data_cleaner(
-            fmt,
-            orig_return_logprob=orig_return_logprob,
-            orig_return_token_ids=orig_return_token_ids,
-            orig_return_routed_experts=orig_return_routed_experts,
+        clean_data = (
+            self._build_data_cleaner(
+                fmt,
+                orig_return_logprob=orig_return_logprob,
+                orig_return_token_ids=orig_return_token_ids,
+                orig_return_routed_experts=orig_return_routed_experts,
+            )
+            if is_generation_endpoint and request_data is not None
+            else None
         )
 
         timeout = ClientTimeout(total=self.request_timeout, sock_connect=30)
@@ -619,6 +662,9 @@ async def _handle_request(self, request: web.Request) -> web.Response:
             async with client.request(
                 method=request.method, url=target_url, headers=forward_headers, data=request_body
             ) as resp:
+                upstream_status = resp.status
+                is_success_response = HTTPStatus.OK <= upstream_status < HTTPStatus.MULTIPLE_CHOICES
+                trace_response = trace_active and is_success_response
                 if is_stream:
                     response_chunks: list[bytes] = []
                     response = web.StreamResponse(
@@ -629,45 +675,64 @@ async def _handle_request(self, request: web.Request) -> web.Response:
                             if k.lower() not in ("transfer-encoding", "content-length", "content-encoding")
                         },
                     )
-                    await response.prepare(request)
                     # If the downstream client closes the socket mid-stream
                     # (e.g. AsyncAPIClient bails out on a finish_reason=='error'
                     # chunk after the prompt overflowed the session window),
                     # keep draining the upstream so the trace is still recorded
                     # in full but stop attempting to write to the closed socket.
-                    client_alive = True
+                    try:
+                        await response.prepare(request)
+                    except (ConnectionError, ClientConnectionResetError):
+                        client_alive = False
+                    else:
+                        client_alive = True
+                    skip_blank_after_done = False
                     async for line in resp.content:
                         # Only retain chunks when we'll actually need to parse
                         # them for tracing; evaluate-mode requests skip this
                         # so memory does not grow with stream length.
-                        if trace_enabled:
+                        if trace_response:
                             response_chunks.append(line)
 
-                        if request_data is not None and line.startswith(b"data: ") and line.strip() != b"data: [DONE]":
+                        if (
+                            is_success_response
+                            and clean_data is not None
+                            and request_data is not None
+                            and line.startswith(b"data: ")
+                            and line.strip() != b"data: [DONE]"
+                        ):
                             try:
                                 text = line.decode("utf-8")
                                 data = json.loads(text[6:])
-                                if clean_data(data):
+                                if isinstance(data, dict) and not _is_error_payload(data) and clean_data(data):
                                     line = ("data: " + json.dumps(data) + "\n").encode("utf-8")
                             except Exception:
                                 pass
 
                         # Delay [DONE] only while a training trace still needs to be exported.
-                        if client_alive and (not trace_enabled or line.strip() != b"data: [DONE]"):
+                        if trace_response and skip_blank_after_done:
+                            if not line.strip():
+                                continue
+                            skip_blank_after_done = False
+                        if line.strip() == b"data: [DONE]":
+                            if trace_response:
+                                skip_blank_after_done = True
+                                continue
+                        if client_alive:
                             try:
                                 await response.write(line)
                             except (ConnectionError, ClientConnectionResetError):
                                 client_alive = False
 
-                    raw_response = b"".join(response_chunks) if trace_enabled else b""
+                    raw_response = b"".join(response_chunks) if trace_response else b""
                 else:
                     raw_response = await resp.read()
                     final_raw_response = raw_response
 
-                    if request_data is not None:
+                    if is_success_response and clean_data is not None and request_data is not None:
                         try:
                             parsed = json.loads(raw_response)
-                            if clean_data(parsed):
+                            if isinstance(parsed, dict) and not _is_error_payload(parsed) and clean_data(parsed):
                                 final_raw_response = json.dumps(parsed).encode("utf-8")
                         except Exception:
                             pass
@@ -684,27 +749,33 @@ async def _handle_request(self, request: web.Request) -> web.Response:
 
         # Apply abstract on_response processing
         response_data: Optional[dict] = None
-        skip_done = bool(is_stream and not trace_enabled)
+        skip_done = bool(is_stream and not trace_response)
         session_error_msg: Optional[str] = None
-        if request_data and trace_enabled and orig_req_body is not None:
+        if trace_response and request_data and orig_req_body is not None:
             if is_stream:
                 try:
                     response_data = self._parse_stream_response(raw_response, fmt)
                     if response_data is None:
-                        # Upstream emitted no traceable content — suppress the
-                        # synthetic [DONE] line we usually append; the real
-                        # stream content has already been forwarded as-is.
-                        skip_done = True
+                        raise RuntimeError("Upstream SSE stream ended without a complete response.")
+                except _UpstreamResponseError:
+                    # An explicit upstream error is already part of the wire
+                    # response. Do not wrap it in a second SessionServer error.
+                    skip_done = b"data: [DONE]" not in raw_response
                 except Exception as exc:
                     session_error_msg = f"SessionServer stream trace failed: {type(exc).__name__}: {exc}"
                     skip_done = True
             else:
                 try:
-                    response_data = json.loads(raw_response)
-                except json.JSONDecodeError:
-                    pass
-                if isinstance(response_data, dict) and _is_error_payload(response_data):
-                    response_data = None
+                    parsed_response = json.loads(raw_response)
+                    if not isinstance(parsed_response, dict):
+                        raise TypeError(
+                            f"generation response body must be a JSON object; got {type(parsed_response).__name__}"
+                        )
+                except (json.JSONDecodeError, UnicodeDecodeError, TypeError) as exc:
+                    session_error_msg = f"SessionServer response trace failed: {type(exc).__name__}: {exc}"
+                else:
+                    if not _is_error_payload(parsed_response):
+                        response_data = parsed_response
 
             if response_data is not None:
                 try:
@@ -717,18 +788,19 @@ async def _handle_request(self, request: web.Request) -> web.Response:
 
         if is_stream:
             try:
-                if session_error_msg:
-                    error_payload = _error_payload(fmt, session_error_msg)
-                    error_line = "data: " + json.dumps(error_payload, ensure_ascii=False) + "\n\n"
-                    # Anthropic SSE carries a named ``event: error`` line; the recorder keys on the ``data:`` JSON
-                    # either way, but a real Anthropic SDK consumer needs the event name.
-                    if fmt == FMT_ANTHROPIC:
-                        error_line = "event: error\n" + error_line
-                    await response.write(error_line.encode("utf-8"))
-                    skip_done = True
-                if not skip_done:
-                    await response.write(b"data: [DONE]\n\n")
-                await response.write_eof()
+                if client_alive:
+                    if session_error_msg:
+                        error_payload = _error_payload(fmt, session_error_msg)
+                        error_line = "data: " + json.dumps(error_payload, ensure_ascii=False) + "\n\n"
+                        # Anthropic SSE carries a named ``event: error`` line; the recorder keys on the ``data:`` JSON
+                        # either way, but a real Anthropic SDK consumer needs the event name.
+                        if fmt == FMT_ANTHROPIC:
+                            error_line = "event: error\n" + error_line
+                        await response.write(error_line.encode("utf-8"))
+                        skip_done = True
+                    elif trace_response and not skip_done:
+                        await response.write(b"data: [DONE]\n\n")
+                    await response.write_eof()
             except (ConnectionError, ClientConnectionResetError):
                 pass
         elif session_error_msg:
@@ -822,8 +894,8 @@ def clean_openai(data: dict) -> bool:
     def _parse_stream_response(raw: bytes, fmt: str) -> Optional[dict]:
         """Parse an SSE stream and reconstruct the final response object.
 
-        Returns ``None`` when the stream carried no traceable content (so the
-        caller skips the trace hook entirely instead of recording garbage).
+        Returns ``None`` when no complete response can be reconstructed. The
+        caller treats that as a local trace failure for trace-enabled requests.
         """
         text = raw.decode("utf-8", errors="replace")
         events: list[dict] = []
@@ -836,7 +908,9 @@ def _parse_stream_response(raw: bytes, fmt: str) -> Optional[dict]:
             if line.startswith("data: "):
                 event = json.loads(line[6:])
                 if _is_error_payload(event):
-                    raise RuntimeError(f"Upstream SSE stream returned error: {json.dumps(event, ensure_ascii=False)}")
+                    raise _UpstreamResponseError(
+                        f"Upstream SSE stream returned error: {json.dumps(event, ensure_ascii=False)}"
+                    )
                 events.append(event)
 
         if not events:
@@ -864,7 +938,7 @@ def _parse_openai_stream(events: list[dict], *, saw_done: bool) -> Optional[dict
 
             for choice in event.get("choices", []):
                 if choice.get("finish_reason") == "error":
-                    raise RuntimeError(
+                    raise _UpstreamResponseError(
                         f"Upstream SSE choice finished with error: {json.dumps(event, ensure_ascii=False)}"
                     )
                 delta = choice.get("delta", {})