diff --git a/qingjian_cs/app/client.py b/qingjian_cs/app/client.py
index b9de03c..1afd5ae 100644
--- a/qingjian_cs/app/client.py
+++ b/qingjian_cs/app/client.py
@@ -20,6 +20,7 @@ from .logger import setup_logger
 from .observability import activity_event, build_trace_id
 from .orchestrator import Orchestrator
 from .rules import extract_image_urls, prefilter_message
+from .runtime_switch import is_listen_only
 
 
 class QingjianClient:
@@ -214,6 +215,14 @@ class QingjianClient:
         return False
 
     async def _handle_decision(self, data: dict, merged_msg: str, *, auto_quote: bool = False) -> None:
+        if is_listen_only():
+            activity_event(
+                self.logger,
+                "ai_reply_skipped",
+                customer_id=data.get("from_id", "-"),
+                reason="listen_only_mode",
+            )
+            return
         key = self._customer_key(data)
         trace_id = build_trace_id(data.get("acc_id", ""), data.get("from_id", ""), merged_msg)
         t0 = time.perf_counter()
@@ -402,6 +411,15 @@ class QingjianClient:
         key = self._customer_key(patched)
         self._append_dialogue(key, "user", patched["msg"])
 
+        if is_listen_only():
+            activity_event(
+                self.logger,
+                "ai_reply_skipped",
+                customer_id=patched.get("from_id", "-"),
+                reason="listen_only_mode",
+            )
+            return
+
         # 硬编码：每个客户首条消息先快速回复“在的”
         if key not in self.first_msg_replied:
             await self.send_reply(patched, "在的")
diff --git a/qingjian_cs/app/http_api.py b/qingjian_cs/app/http_api.py
index e80b3e4..ea58cb9 100644
--- a/qingjian_cs/app/http_api.py
+++ b/qingjian_cs/app/http_api.py
@@ -3,6 +3,7 @@ from __future__ import annotations
 from flask import Flask, jsonify, request
 
 from .logger import setup_logger
+from .runtime_switch import is_listen_only, set_listen_only
 from .task_manager import TaskManager
 
 
@@ -15,6 +16,20 @@ def create_http_app(task_manager: TaskManager | None = None) -> Flask:
     def health():
         return jsonify({'ok': True})
 
+    @app.get('/api/runtime/listen_only')
+    def get_listen_only():
+        return jsonify({'ok': True, 'listen_only': is_listen_only()})
+
+    @app.post('/api/runtime/listen_only')
+    def set_listen_only_mode():
+        body = request.get_json(silent=True) or {}
+        if "enabled" not in body:
+            return jsonify({'ok': False, 'error': 'enabled required'}), 400
+        enabled = bool(body.get("enabled"))
+        current = set_listen_only(enabled)
+        logger.info('[运行时] listen_only=%s', current)
+        return jsonify({'ok': True, 'listen_only': current})
+
     @app.post('/api/task/receive')
     def receive_task():
         payload = request.get_json(silent=True) or {}
diff --git a/qingjian_cs/app/runtime_switch.py b/qingjian_cs/app/runtime_switch.py
new file mode 100644
index 0000000..ab66a43
--- /dev/null
+++ b/qingjian_cs/app/runtime_switch.py
@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+import threading
+
+
+_lock = threading.Lock()
+_listen_only_mode = False
+
+
+def set_listen_only(enabled: bool) -> bool:
+    global _listen_only_mode
+    with _lock:
+        _listen_only_mode = bool(enabled)
+        return _listen_only_mode
+
+
+def is_listen_only() -> bool:
+    with _lock:
+        return _listen_only_mode
+