diff --git a/tests/tui_gateway/test_session_steer_text_coerce.py b/tests/tui_gateway/test_session_steer_text_coerce.py
new file mode 100644
index 0000000000000..ba4e1e80f7e36
--- /dev/null
+++ b/tests/tui_gateway/test_session_steer_text_coerce.py
@@ -0,0 +1,90 @@
+"""session.steer / session.redirect must tolerate non-string ``text``.
+
+Both RPCs run inline (not in ``_LONG_HANDLERS``). A bare ``.strip()`` on a
+list/int ``text`` raises AttributeError and can tear down the stdin/WS
+reader loop.
+"""
+
+from __future__ import annotations
+
+import threading
+
+from tui_gateway import server
+
+
+def _install_session(sid: str, *, support_steer: bool = True):
+    class _Agent:
+        def steer(self, text):
+            self.last = text
+            return True
+
+        def redirect(self, text):
+            self.last = text
+            return True
+
+    agent = _Agent() if support_steer else object()
+    ready = threading.Event()
+    ready.set()
+    server._sessions[sid] = {
+        "agent": agent,
+        "agent_ready": ready,
+        "session_key": sid,
+        "history_lock": threading.Lock(),
+        "last_active": 0,
+    }
+    return agent
+
+
+def test_session_steer_rejects_empty_after_coerce():
+    sid = "steer-empty"
+    _install_session(sid)
+    try:
+        resp = server.dispatch(
+            {
+                "id": "1",
+                "method": "session.steer",
+                "params": {"session_id": sid, "text": None},
+            }
+        )
+        assert resp["error"]["code"] == 4002
+    finally:
+        server._sessions.pop(sid, None)
+
+
+def test_session_steer_coerces_int_text():
+    sid = "steer-int"
+    agent = _install_session(sid)
+    try:
+        resp = server.dispatch(
+            {
+                "id": "2",
+                "method": "session.steer",
+                "params": {"session_id": sid, "text": 42},
+            }
+        )
+        assert "error" not in resp, resp
+        assert resp["result"]["status"] == "queued"
+        assert resp["result"]["text"] == "42"
+        assert agent.last == "42"
+    finally:
+        server._sessions.pop(sid, None)
+
+
+def test_session_redirect_coerces_list_text_without_crash():
+    """List text must not AttributeError on the inline reader path."""
+    sid = "redir-list"
+    _install_session(sid)
+    try:
+        resp = server.dispatch(
+            {
+                "id": "3",
+                "method": "session.redirect",
+                "params": {"session_id": sid, "text": ["go", "left"]},
+            }
+        )
+        # Coerced to "['go', 'left']" which is non-empty — either queued/rejected
+        # or a later agent error, but never an uncaught AttributeError.
+        assert isinstance(resp, dict)
+        assert "error" in resp or "result" in resp
+    finally:
+        server._sessions.pop(sid, None)
diff --git a/tui_gateway/methods_session.py b/tui_gateway/methods_session.py
index bdbcd1cfafafe..5e870ca9fa86b 100644
--- a/tui_gateway/methods_session.py
+++ b/tui_gateway/methods_session.py
@@ -1082,11 +1082,11 @@ def _(rid, params: dict) -> dict:
     configured auxiliary ``task`` backend. Never mutates session history, so
     prompt caching is untouched.
     """
-    template = (params.get("template") or "").strip() or None
+    template = str(params.get("template") or "").strip() or None
     instructions = params.get("instructions") or ""
     user_input = params.get("input") or ""
     variables = params.get("variables") if isinstance(params.get("variables"), dict) else {}
-    task = (params.get("task") or "title_generation").strip() or "title_generation"
+    task = str(params.get("task") or "title_generation").strip() or "title_generation"
 
     try:
         max_tokens = int(params.get("max_tokens") or 1024)
@@ -1151,7 +1151,7 @@ def _(rid, params: dict) -> dict:
             "session busy — wait for the current turn to finish, then retry the handoff",
         )
 
-    platform_name = (params.get("platform", "") or "").strip().lower()
+    platform_name = str(params.get("platform") or "").strip().lower()
     if not platform_name:
         return _err(rid, 4023, "platform required")
 
@@ -3061,7 +3061,9 @@ def _(rid, params: dict) -> dict:
     it on its next iteration. No interrupt, no new user turn, no role
     alternation violation.
     """
-    text = (params.get("text") or "").strip()
+    # Inline RPC: non-string text (JSON null already falsy; list/int is truthy)
+    # must not AttributeError on .strip() and tear down the reader thread.
+    text = str(params.get("text") or "").strip()
     if not text:
         return _err(rid, 4002, "text is required")
     session, err = _sess_nowait(params, rid)
@@ -3088,7 +3090,7 @@ def _(rid, params: dict) -> dict:
 @method("session.redirect")
 def _(rid, params: dict) -> dict:
     """Redirect the active model turn while preserving valid work/context."""
-    text = (params.get("text") or "").strip()
+    text = str(params.get("text") or "").strip()
     if not text:
         return _err(rid, 4002, "text is required")
     session, err = _sess_nowait(params, rid)