diff --git a/src/api/endpoints.py b/src/api/endpoints.py
index 3c5bf25..13907b4 100644
--- a/src/api/endpoints.py
+++ b/src/api/endpoints.py
@@ -49,6 +49,28 @@ STATUS_CODE_MAP = {
     '2016': '确认双车中的车牌'
 }
 
+
+def _stage_label(normalized_stage_code: str) -> str:
+    """阶段码的中文标签；未知码返回空串（B4 监测点）。"""
+    return STATUS_CODE_MAP.get(normalized_stage_code, '')
+
+
+def _warn_unknown_stage_code(
+    *, session_id: str, client_mode: str,
+    raw_stage_code: str, normalized_stage_code: str,
+) -> None:
+    """B4：LLM/后端输出映射表外的阶段码时告警（灰度监测）。
+
+    未知码当前行为：原样透传 nextStageCode、nextStage 为空串。
+    告警带全定位信息，用于确认提示词输出码集是否收敛。
+    """
+    logger.warning(
+        "Unrecognized stage code "
+        f"sessionId={session_id} clientMode={client_mode} "
+        f"rawStageCode={raw_stage_code} "
+        f"normalizedStageCode={normalized_stage_code}"
+    )
+
 def normalize_stage_code(stage_code: str) -> str:
     """Normalize FastGPT stage codes to external API stage codes."""
     if stage_code in ['3001', '3002', '1002']:
@@ -315,7 +337,14 @@ async def chat(
                                     client_mode,
                                     normalized_stage_code,
                                 )
-                                nextStage = STATUS_CODE_MAP.get(normalized_stage_code, '')
+                                nextStage = _stage_label(normalized_stage_code)
+                                if not nextStage:
+                                    _warn_unknown_stage_code(
+                                        session_id=json_data['sessionId'],
+                                        client_mode=client_mode,
+                                        raw_stage_code=state_code,
+                                        normalized_stage_code=normalized_stage_code,
+                                    )
                                 logger.info(
                                     "FastGPT stream stage_code parsed "
                                     f"sessionId={json_data['sessionId']} "
@@ -470,7 +499,14 @@ async def chat(
             raise ValueError("大模型回复中缺少state")
         normalized_stage_code = normalize_stage_code(raw_stage_code)
         nextStageCode = external_stage_code(client_mode, normalized_stage_code)
-        nextStage = STATUS_CODE_MAP.get(normalized_stage_code, '')
+        nextStage = _stage_label(normalized_stage_code)
+        if not nextStage:
+            _warn_unknown_stage_code(
+                session_id=json_data['sessionId'],
+                client_mode=client_mode,
+                raw_stage_code=raw_stage_code,
+                normalized_stage_code=normalized_stage_code,
+            )
         form_update = result.form_update
         logger.info(
             "Chat non-stream completed "
diff --git a/test/api/test_unknown_stage_code.py b/test/api/test_unknown_stage_code.py
new file mode 100644
index 0000000..480986f
--- /dev/null
+++ b/test/api/test_unknown_stage_code.py
@@ -0,0 +1,107 @@
+"""B4 监测：未识别 stage 码告警（灰度期确认提示词输出码集收敛）。
+
+LLM/后端输出映射表（STATUS_CODE_MAP）外的阶段码时，当前行为是
+原样透传 nextStageCode、nextStage 为空串——静默异常。
+本测试固定：两条路径（非流式/流式）都会打 warning，已知码不打。
+"""
+
+import pytest
+
+import src.api.endpoints as endpoints
+from src.api.endpoints import chat
+from src.backends.chat import ChatInput, ChatResult, TextDelta
+from src.schemas.models import ProcessRequest_chat
+
+
+class FakeBackend:
+    """固定返回指定 state 前缀的 backend。"""
+
+    def __init__(self, content: str):
+        self._content = content
+
+    async def complete(self, chat_input: ChatInput):
+        return ChatResult(content=self._content, status_code=None, form_update={})
+
+    async def stream(self, chat_input: ChatInput):
+        yield TextDelta(self._content)
+
+
+def make_request():
+    return ProcessRequest_chat(
+        sessionId="session-b4",
+        timeStamp="20261022120000",
+        text="测试输入",
+    )
+
+
+class _WarningRecorder:
+    def __init__(self):
+        self.messages: list[str] = []
+
+    def warning(self, message: str) -> None:
+        self.messages.append(message)
+
+    def __getattr__(self, _name):
+        # info/debug/error 等：no-op（本测试只关心 warning）
+        return lambda *a, **k: None
+
+
+@pytest.mark.asyncio
+async def test_unknown_stage_code_warns_non_stream(monkeypatch):
+    recorder = _WarningRecorder()
+    monkeypatch.setattr(endpoints, "logger", recorder)
+    backend = FakeBackend("<state>9999</state>收到了")
+
+    response = await chat(make_request(), stream=False, backend=backend)
+
+    assert response.nextStageCode == "9999"
+    assert response.nextStage == ""
+    assert len(recorder.messages) == 1
+    assert "rawStageCode=9999" in recorder.messages[0]
+    assert "sessionId=session-b4" in recorder.messages[0]
+
+
+@pytest.mark.asyncio
+async def test_unknown_stage_code_warns_stream(monkeypatch):
+    recorder = _WarningRecorder()
+    monkeypatch.setattr(endpoints, "logger", recorder)
+    backend = FakeBackend("<state>9999</state>收到了")
+
+    response = await chat(make_request(), stream=True, backend=backend)
+    chunks = []
+    async for chunk in response.body_iterator:
+        chunks.append(chunk.decode() if isinstance(chunk, bytes) else chunk)
+
+    assert '"nextStageCode": "9999"' in "".join(chunks)
+    assert len(recorder.messages) == 1
+    assert "rawStageCode=9999" in recorder.messages[0]
+
+
+@pytest.mark.asyncio
+async def test_normalized_internal_code_warns_with_normalized_value(monkeypatch):
+    """归一化映射表内的内部码（如 2020->0002 已是已知码）不告警；
+    映射表未覆盖的内部码按归一化后的值告警。"""
+    recorder = _WarningRecorder()
+    monkeypatch.setattr(endpoints, "logger", recorder)
+    known = FakeBackend("<state>2020</state>识别不了")
+    response = await chat(make_request(), stream=False, backend=known)
+    assert response.nextStageCode == "0002"      # 归一化生效
+    assert recorder.messages == []               # 已知码不告警
+
+    unknown = FakeBackend("<state>3009</state>x")
+    response = await chat(make_request(), stream=False, backend=unknown)
+    assert response.nextStageCode == "3009"
+    assert len(recorder.messages) == 1
+    assert "normalizedStageCode=3009" in recorder.messages[0]
+
+
+@pytest.mark.asyncio
+async def test_known_stage_code_no_warning(monkeypatch):
+    recorder = _WarningRecorder()
+    monkeypatch.setattr(endpoints, "logger", recorder)
+    backend = FakeBackend("<state>1002</state>继续")
+
+    response = await chat(make_request(), stream=False, backend=backend)
+
+    assert response.nextStage == "通话中"
+    assert recorder.messages == []