diff --git a/mcp_server/cli.py b/mcp_server/cli.py
index d64deec..ba83f70 100644
--- a/mcp_server/cli.py
+++ b/mcp_server/cli.py
@@ -259,8 +259,7 @@ async def cmd_check_status(args: argparse.Namespace) -> int:
 async def cmd_retry_generation(args: argparse.Namespace) -> int:
     """retry-generation: retry a failed generation."""
     from services.session import set_project_root, resolve_generation_id
-    from services.specflow_backend import call_backend_endpoint
-    from services.tool_helpers import check_status_safe, is_generation_in_progress
+    from services.retry import retry_generation_core
 
     root = resolve_root(args.root_path)
     print(f"Using project root: {root}")
@@ -271,21 +270,12 @@ async def cmd_retry_generation(args: argparse.Namespace) -> int:
         print("No previous generation found. Run `specflow run-generation` to start one.")
         return 0
 
-    status_data = await check_status_safe(generation_id)
-    if is_generation_in_progress(status_data):
-        print(
-            "ERROR: A generation is already running. Wait for it to finish before retrying.",
-            file=sys.stderr,
-        )
+    try:
+        backend_data = await retry_generation_core(generation_id)
+    except Exception as exc:  # noqa: BLE001 - backend validates state; surface its reason
+        print(f"ERROR: Couldn't retry: {exc}", file=sys.stderr)
         return 1
-
-    response_text = await call_backend_endpoint(
-        endpoint=f"/api/v1/generation-sessions/{generation_id}/retry",
-        method="POST",
-        timeout_seconds=30,
-    )
-    data = json.loads(response_text)
-    print(json.dumps(data, indent=2))
+    print(json.dumps(backend_data, indent=2))
     print("\nRetry queued. Generation will resume from the last checkpoint on the same workspaces.")
     return 0
 
diff --git a/mcp_server/server.py b/mcp_server/server.py
index 649d503..1688fe1 100644
--- a/mcp_server/server.py
+++ b/mcp_server/server.py
@@ -57,6 +57,7 @@
     resolve_workspace_count,
 )
 from services.user_response import brief_sentences, chat_json
+from services.retry import retry_generation_core
 
 BACKEND_URL = os.getenv("BACKEND_URL", "http://127.0.0.1:8000")
 logger.info("MCP Server starting with Backend URL: %s", BACKEND_URL)
@@ -617,58 +618,27 @@ async def retry_generation(
         if sp.is_absolute():
             set_project_root(sp.parent)
 
-    try:
-        generation_id = resolve_generation_id(generation_id)
-        if not generation_id:
-            return chat_json(
-                "No previous generation found. Run `run_generation` to start one.",
-            )
-
-        status_data = await check_status_safe(generation_id)
-        if is_generation_in_progress(status_data):
-            return _rejection_chat_payload({
-                "error": "A generation is already running. Wait for it to finish before retrying.",
-                "code": PrecheckRejectionCode.GENERATION_ALREADY_RUNNING,
-                "generation_id": generation_id,
-            })
-
-        if status_data:
-            st = (status_data.get("status") or "").lower()
-            if st == GenerationStatus.PENDING and (status_data.get("error") or "").strip():
-                return chat_json(
-                    brief_sentences(
-                        f"Last run was rejected before codegen: {status_data.get('error')} "
-                        "Fix files locally and call `run_generation` — not `retry_generation`."
-                    ),
-                    details=status_data,
-                    generation_id=generation_id,
-                )
-            if st == GenerationStatus.PENDING:
-                return chat_json(
-                    "Session is pending but has not failed. Use `run_generation`, not `retry_generation`.",
-                    details=status_data,
-                    generation_id=generation_id,
-                )
-
-        response_text = await call_backend_endpoint(
-            endpoint=f"/api/v1/generation-sessions/{generation_id}/retry",
-            method="POST",
-            timeout_seconds=30,
-        )
-        backend_data = json.loads(response_text)
+    generation_id = resolve_generation_id(generation_id)
+    if not generation_id:
         return chat_json(
-            "Retry queued. Generation will resume from the last checkpoint on the same workspaces.",
-            details=backend_data,
-            generation_id=generation_id,
-            status=backend_data.get("status"),
-            retry_count=backend_data.get("retry_count"),
+            "No previous generation found. Run `run_generation` to start one.",
         )
 
-    except Exception as e:
+    try:
+        backend_data = await retry_generation_core(generation_id)
+    except Exception as e:  # noqa: BLE001 - backend validates state; surface its reason
         return chat_json(
-            "Couldn't retry. The SpecFlow server may be unreachable.",
+            brief_sentences(f"Couldn't retry: {e}"),
             details={"error": str(e)},
+            generation_id=generation_id,
         )
+    return chat_json(
+        "Retry queued. Generation will resume from the last checkpoint on the same workspaces.",
+        details=backend_data,
+        generation_id=generation_id,
+        status=backend_data.get("status"),
+        retry_count=backend_data.get("retry_count"),
+    )
 
 
 def main():
diff --git a/mcp_server/services/retry.py b/mcp_server/services/retry.py
new file mode 100644
index 0000000..294239e
--- /dev/null
+++ b/mcp_server/services/retry.py
@@ -0,0 +1,27 @@
+"""Shared retry call — one definition of the retry endpoint for CLI, MCP, and TUI.
+
+Retry eligibility is validated by the backend state machine (``reset_for_retry``
+accepts FAILED or stuck-PENDING); there is no client-side status pre-check to
+drift from it. Callers POST here and render success (the returned dict) vs the
+backend's error message (``str(exc)``) in their own style — mirroring the cancel
+flow, which likewise defers to the backend.
+"""
+from __future__ import annotations
+
+import json
+
+from services.specflow_backend import call_backend_endpoint
+
+
+async def retry_generation_core(generation_id: str) -> dict:
+    """POST a retry for an already-resolved id; return the parsed backend response.
+
+    Raises on any failure — the backend validates state and returns the reason,
+    so callers surface ``str(exc)`` to the user.
+    """
+    text = await call_backend_endpoint(
+        endpoint=f"/api/v1/generation-sessions/{generation_id}/retry",
+        method="POST",
+        timeout_seconds=30,
+    )
+    return json.loads(text)
diff --git a/mcp_server/tests/test_cli.py b/mcp_server/tests/test_cli.py
index b64e39a..84e635d 100644
--- a/mcp_server/tests/test_cli.py
+++ b/mcp_server/tests/test_cli.py
@@ -285,40 +285,51 @@ async def test_calls_retry_endpoint(self, tmp_project):
 
         args = SimpleNamespace(root_path=str(tmp_project), command="retry-generation", generation_id=None)
 
-        # check_status_safe returns failed → proceed to retry POST
+        # POST succeeds → exit 0. Patched at services.retry.* where the helper binds it.
         with patch(
-            "services.tool_helpers.check_status_safe",
-            new_callable=AsyncMock,
-        ) as mock_status, patch(
-            "services.specflow_backend.call_backend_endpoint",
+            "services.retry.call_backend_endpoint",
             new_callable=AsyncMock,
         ) as mock_ep:
-            mock_status.return_value = {"status": "failed"}
-            mock_ep.return_value = json.dumps({"status": "running", "retry_count": 1})
+            mock_ep.return_value = json.dumps({"status": "pending", "retry_count": 1})
             code = await cmd_retry_generation(args)
 
         assert code == 0
-        # Exactly one backend call: the retry POST
-        assert mock_ep.call_count == 1
-        assert mock_status.call_count == 1
+        assert mock_ep.call_count == 1  # single POST; backend validates state
 
     @pytest.mark.asyncio
-    async def test_blocks_when_already_running(self, tmp_project, capsys):
+    async def test_backend_rejection_returns_one(self, tmp_project):
         from cli import cmd_retry_generation
         from services.session import write_session
         write_session("gen-running", tmp_project)
 
         args = SimpleNamespace(root_path=str(tmp_project), command="retry-generation", generation_id=None)
 
+        # Wrong-state / unreachable both raise from the helper → exit 1.
         with patch(
-            "services.tool_helpers.check_status_safe",
+            "services.retry.call_backend_endpoint",
             new_callable=AsyncMock,
-        ) as mock_cs:
-            mock_cs.return_value = {"status": "running"}
+        ) as mock_ep:
+            mock_ep.side_effect = Exception("Backend returned HTTP 400: Cannot retry in 'running' state")
             code = await cmd_retry_generation(args)
 
         assert code == 1
 
+    @pytest.mark.asyncio
+    async def test_no_session_returns_zero_without_backend_call(self, tmp_project):
+        from cli import cmd_retry_generation
+
+        # No write_session → resolve_generation_id returns None.
+        args = SimpleNamespace(root_path=str(tmp_project), command="retry-generation", generation_id=None)
+
+        with patch(
+            "services.retry.call_backend_endpoint",
+            new_callable=AsyncMock,
+        ) as mock_ep:
+            code = await cmd_retry_generation(args)
+
+        assert code == 0
+        assert mock_ep.call_count == 0  # helper never called
+
 
 # ---------------------------------------------------------------------------
 # cmd_download_outputs
diff --git a/mcp_server/tests/test_retry_core.py b/mcp_server/tests/test_retry_core.py
new file mode 100644
index 0000000..ed386d3
--- /dev/null
+++ b/mcp_server/tests/test_retry_core.py
@@ -0,0 +1,33 @@
+"""Tests for the shared retry helper (services/retry.py).
+
+The helper POSTs the retry and returns the parsed backend dict, raising on
+failure (the backend validates eligibility). Patch ``call_backend_endpoint`` at
+``services.retry.*`` where the helper binds it.
+"""
+import json
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from services.retry import retry_generation_core
+
+GID = "est-abc123"
+
+
+@pytest.mark.asyncio
+async def test_returns_backend_dict_on_success():
+    post = AsyncMock(return_value=json.dumps({"status": "pending", "retry_count": 1}))
+    with patch("services.retry.call_backend_endpoint", new=post):
+        data = await retry_generation_core(GID)
+    assert data == {"status": "pending", "retry_count": 1}
+    assert post.await_count == 1
+    assert GID in post.await_args.kwargs["endpoint"]
+    assert post.await_args.kwargs["method"] == "POST"
+
+
+@pytest.mark.asyncio
+async def test_raises_on_backend_failure():
+    post = AsyncMock(side_effect=Exception("Backend returned HTTP 400: Cannot retry in 'running' state"))
+    with patch("services.retry.call_backend_endpoint", new=post):
+        with pytest.raises(Exception, match="running"):
+            await retry_generation_core(GID)
diff --git a/mcp_server/tests/test_server_retry.py b/mcp_server/tests/test_server_retry.py
new file mode 100644
index 0000000..aefb408
--- /dev/null
+++ b/mcp_server/tests/test_server_retry.py
@@ -0,0 +1,64 @@
+"""Tests for the MCP ``retry_generation`` tool after deferring to the backend.
+
+The tool resolves the id, POSTs via the shared helper, and renders success vs
+the backend's error message. `ctx` is only consumed by
+`apply_project_root_from_context`, which we stub. The POST is patched at
+`services.retry.*` where the helper binds it.
+"""
+import json
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+import server
+
+GID = "est-abc123"
+
+
+async def _run(*, resolved=GID, post=None):
+    """Invoke retry_generation with IO stubbed; return (payload dict, post mock)."""
+    post_mock = (
+        AsyncMock(side_effect=post) if isinstance(post, Exception)
+        else AsyncMock(return_value=post)
+    )
+    with (
+        patch("server.apply_project_root_from_context", new=AsyncMock()),
+        patch("server.resolve_generation_id", return_value=resolved),
+        patch("services.retry.call_backend_endpoint", new=post_mock),
+    ):
+        payload = await server.retry_generation(generation_id=None, ctx=object())
+    return json.loads(payload), post_mock
+
+
+@pytest.mark.asyncio
+async def test_no_session():
+    with (
+        patch("server.apply_project_root_from_context", new=AsyncMock()),
+        patch("server.resolve_generation_id", return_value=None),
+    ):
+        payload = await server.retry_generation(generation_id=None, ctx=object())
+    data = json.loads(payload)
+    assert "No previous generation found" in data["message"]
+
+
+@pytest.mark.asyncio
+async def test_queued():
+    data, post = await _run(post=json.dumps({"status": "pending", "retry_count": 1}))
+    assert "retry queued" in data["message"].lower()
+    assert data["status"] == "pending"
+    assert data["retry_count"] == 1
+    assert data["details"] == {"status": "pending", "retry_count": 1}
+    assert data["generation_id"] == GID
+    assert post.await_count == 1
+
+
+@pytest.mark.asyncio
+async def test_backend_rejection_or_error_is_surfaced():
+    # Covers wrong-state rejections (HTTP 400) and unreachable backend alike —
+    # both raise from the helper and render the backend's reason.
+    data, post = await _run(
+        post=Exception("Backend returned HTTP 400: Cannot retry in 'running' state")
+    )
+    assert "couldn't retry" in data["message"].lower()
+    assert "running" in data["details"]["error"]
+    assert data["generation_id"] == GID
diff --git a/mcp_server/tests/test_tui_actions.py b/mcp_server/tests/test_tui_actions.py
index bedcb05..495fead 100644
--- a/mcp_server/tests/test_tui_actions.py
+++ b/mcp_server/tests/test_tui_actions.py
@@ -3,9 +3,10 @@
 The wrappers must delegate to the existing cli.cmd_* handlers (single source of
 truth for guards/precheck/backend calls), passing the namespace fields each
 handler reads. We patch the handlers and assert delegation + key arguments.
+(Retry no longer goes through here — it calls the retry core directly; see
+test_tui_app.TestDashboardActionFlows.)
 """
 
-from pathlib import Path
 from unittest.mock import AsyncMock, patch
 
 import pytest
@@ -13,15 +14,6 @@
 from tui import actions
 
 
-@pytest.mark.asyncio
-async def test_do_retry_delegates_with_root():
-    with patch("cli.cmd_retry_generation", new=AsyncMock(return_value=0)) as m:
-        rc = await actions.do_retry(Path("/proj"))
-    assert rc == 0
-    ns = m.await_args.args[0]
-    assert ns.root_path == "/proj"
-
-
 @pytest.mark.asyncio
 async def test_do_clear_set_forces_yes():
     with patch("cli.cmd_clear_workspace", new=AsyncMock(return_value=0)) as m:
diff --git a/mcp_server/tests/test_tui_app.py b/mcp_server/tests/test_tui_app.py
index d0ab110..1859c3d 100644
--- a/mcp_server/tests/test_tui_app.py
+++ b/mcp_server/tests/test_tui_app.py
@@ -1190,15 +1190,15 @@ async def test_cancel_stops_countdown_timer(self):
 
 
 class TestDashboardActionFlows:
-    """Retry/clear worker flows: confirmation gating and clear eligibility.
+    """Retry/clear worker flows: confirmation gating and in-app feedback.
 
-    ``_run_suspended`` (which suspends the app and blocks on ``input()``) is
-    patched out so we exercise only the decision logic; ``actions.do_*`` are
-    patched to assert which CLI handler the flow ultimately reaches.
+    Retry now calls ``retry_generation_core`` directly and renders results via
+    ``notify``/``MessageScreen`` — it must never suspend the app. We patch the
+    core to force each outcome and assert the flow's response.
     """
 
     @pytest.mark.asyncio
-    async def test_retry_runs_action_when_confirmed(self):
+    async def test_retry_calls_core_and_refreshes_without_suspending(self):
         a, b, c = _gate_ready()
         with a, b, c, patch("tui.app.poll_once", new=AsyncMock(return_value=_running_payload())):
             app = tui_app.SpecFlowTUI(root=Path("/tmp/x"), generation_id="gen_x", poll_interval=999)
@@ -1207,19 +1207,47 @@ async def test_retry_runs_action_when_confirmed(self):
                 screen = app.screen
                 with (
                     patch.object(app, "push_screen_wait", new=AsyncMock(return_value=True)),
-                    patch.object(screen, "_run_suspended", new=AsyncMock()) as run_susp,
-                    # do_retry is async; force a sync mock so the flow's
-                    # ``do_retry(root, generation_id)`` yields a sentinel, not a live coroutine.
                     patch(
-                        "tui.app.actions.do_retry", new=MagicMock(return_value="retry-coro")
-                    ) as do_retry,
+                        "tui.app.retry_generation_core",
+                        new=AsyncMock(return_value={"status": "pending", "retry_count": 1}),
+                    ) as core,
+                    patch.object(screen, "refresh_status", new=AsyncMock()) as refresh,
+                    patch.object(screen, "notify") as notify,
+                    patch.object(app, "suspend") as suspend,
                 ):
                     await screen._retry_flow()
-                do_retry.assert_called_once_with(app.root, "gen_x")
-                run_susp.assert_awaited_once_with("retry-coro")
+                core.assert_awaited_once_with("gen_x")
+                refresh.assert_awaited_once()
+                suspend.assert_not_called()  # never drops to the terminal
+                assert any("queued" in str(c.args[0]).lower() for c in notify.call_args_list)
 
     @pytest.mark.asyncio
-    async def test_retry_skips_action_when_cancelled(self):
+    async def test_retry_backend_error_shows_message_screen(self):
+        a, b, c = _gate_ready()
+        with a, b, c, patch("tui.app.poll_once", new=AsyncMock(return_value=_running_payload())):
+            app = tui_app.SpecFlowTUI(root=Path("/tmp/x"), generation_id="gen_x", poll_interval=999)
+            async with app.run_test() as pilot:
+                await pilot.pause()
+                screen = app.screen
+                psw = AsyncMock(return_value=True)
+                with (
+                    patch.object(app, "push_screen_wait", new=psw),
+                    patch(
+                        "tui.app.retry_generation_core",
+                        new=AsyncMock(side_effect=Exception("Backend returned HTTP 400: Cannot retry")),
+                    ),
+                    patch.object(screen, "refresh_status", new=AsyncMock()) as refresh,
+                    patch.object(app, "suspend") as suspend,
+                ):
+                    await screen._retry_flow()
+                suspend.assert_not_called()
+                # push_screen_wait called twice: ConfirmScreen, then the error MessageScreen.
+                assert psw.await_count == 2
+                assert isinstance(psw.await_args_list[1].args[0], tui_app.MessageScreen)
+                refresh.assert_not_awaited()  # error returns before the refresh
+
+    @pytest.mark.asyncio
+    async def test_retry_skips_core_when_cancelled(self):
         a, b, c = _gate_ready()
         with a, b, c, patch("tui.app.poll_once", new=AsyncMock(return_value=_running_payload())):
             app = tui_app.SpecFlowTUI(root=Path("/tmp/x"), generation_id="gen_x", poll_interval=999)
@@ -1228,12 +1256,33 @@ async def test_retry_skips_action_when_cancelled(self):
                 screen = app.screen
                 with (
                     patch.object(app, "push_screen_wait", new=AsyncMock(return_value=False)),
-                    patch.object(screen, "_run_suspended", new=AsyncMock()) as run_susp,
-                    patch("tui.app.actions.do_retry") as do_retry,
+                    patch("tui.app.retry_generation_core", new=AsyncMock()) as core,
+                    patch.object(screen, "refresh_status", new=AsyncMock()) as refresh,
                 ):
                     await screen._retry_flow()
-                do_retry.assert_not_called()
-                run_susp.assert_not_awaited()
+                core.assert_not_awaited()
+                refresh.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_poll_timer_self_heals_after_terminal(self):
+        a, b, c = _gate_ready()
+        with a, b, c, patch("tui.app.poll_once", new=AsyncMock(return_value=_running_payload())):
+            app = tui_app.SpecFlowTUI(root=Path("/tmp/x"), generation_id="gen_x", poll_interval=999)
+            async with app.run_test() as pilot:
+                await pilot.pause()
+                screen = app.screen
+                # Terminal payload stops the poll timer.
+                with patch("tui.app.poll_once", new=AsyncMock(return_value={"status": "failed"})):
+                    await screen.refresh_status()
+                assert screen._poll_timer is None
+                # A subsequent non-terminal payload re-arms exactly one timer...
+                with patch("tui.app.poll_once", new=AsyncMock(return_value={"status": "pending"})):
+                    await screen.refresh_status()
+                    assert screen._poll_timer is not None
+                    armed = screen._poll_timer
+                    # ...and refreshing again while non-terminal does not create a second.
+                    await screen.refresh_status()
+                    assert screen._poll_timer is armed
 
     @pytest.mark.asyncio
     async def test_clear_runs_for_eligible_set_when_confirmed(self):
diff --git a/mcp_server/tui/actions.py b/mcp_server/tui/actions.py
index 556a32c..a858685 100644
--- a/mcp_server/tui/actions.py
+++ b/mcp_server/tui/actions.py
@@ -1,19 +1,20 @@
 """In-app actions — thin wrappers over the existing CLI command handlers.
 
 These deliberately call the same ``cmd_*`` coroutines the standalone
-subcommands use (``cli.cmd_retry_generation`` etc.), so every guard, capacity
+subcommands use (``cli.cmd_clear_workspace`` etc.), so every guard, capacity
 message, precheck, and backend call stays in exactly one place. The TUI never
 re-implements those flows; it suspends its screen (see ``app.py``) and runs the
 real handler, which prints its familiar output to the terminal.
 
 Each wrapper builds the ``argparse.Namespace`` the handler expects and returns
 its integer exit code. ``--yes`` is forced on for the workspace clear because
-the TUI gathers confirmation through its own dialog before calling.
+the TUI gathers confirmation through its own dialog before calling. (Retry does
+not go through here — it calls ``services.retry.retry_generation_core`` directly
+so it can render feedback in-app instead of suspending.)
 """
 
 from __future__ import annotations
 
-from pathlib import Path
 from types import SimpleNamespace
 
 import cli
@@ -44,13 +45,6 @@ def _ns(**overrides: object) -> SimpleNamespace:
     return SimpleNamespace(**base)
 
 
-async def do_retry(root: Path, generation_id: str | None = None) -> int:
-    """Retry the current generation (reuses ``cmd_retry_generation`` guards)."""
-    return await cli.cmd_retry_generation(
-        _ns(root_path=str(root), generation_id=generation_id)
-    )
-
-
 async def do_clear_set(set_number: int) -> int:
     """Clear all members of a workspace set (confirmation handled by the TUI)."""
     return await cli.cmd_clear_workspace(_ns(set=set_number, yes=True))
diff --git a/mcp_server/tui/app.py b/mcp_server/tui/app.py
index 83040a9..bf808f6 100644
--- a/mcp_server/tui/app.py
+++ b/mcp_server/tui/app.py
@@ -63,6 +63,7 @@
 from cli import resolve_backend_config
 from services import local_env, validate_models
 from services.llm_tiers import LLM_TIER_KEYS
+from services.retry import retry_generation_core
 from services.session import resolve_generation_id, set_project_root
 from services.specflow_backend import call_backend_endpoint, call_backend_endpoint_bytes
 from tui import actions, activity, mcp_clients, onboarding, render
@@ -536,11 +537,15 @@ async def refresh_status(self) -> None:
             build_dashboard(payload, self.app.root, self._generation_id, self._selected_ws_id)
         )
         self.refresh_bindings()
-        # A finished run will not change again — stop polling it.
+        # A finished run will not change again — stop polling it. Conversely, a
+        # run that left a terminal state (e.g. failed → retry → pending) must
+        # resume polling, so re-arm the timer if it was previously stopped.
         if payload and (payload.get("status") or "").lower() in TERMINAL_STATUSES:
             if self._poll_timer is not None:
                 self._poll_timer.stop()
                 self._poll_timer = None
+        elif self._poll_timer is None:
+            self._poll_timer = self.set_interval(self.app.poll_interval, self.refresh_status)
 
     def check_action(self, action: str, parameters: tuple[object, ...]) -> bool | None:
         if action == "retry":
@@ -631,8 +636,16 @@ async def _retry_flow(self) -> None:
                 countdown=0,
             )
         )
-        if ok:
-            await self._run_suspended(actions.do_retry(self.app.root, self._generation_id))
+        if not ok:
+            return
+        self.notify("Retrying generation…", severity="information")
+        try:
+            await retry_generation_core(self._generation_id)
+        except Exception as exc:  # noqa: BLE001 - surface the backend's reason to the user
+            await self.app.push_screen_wait(MessageScreen("Retry failed", str(exc)))
+            return
+        self.notify("Retry queued. Resuming from the last checkpoint.", severity="information")
+        await self.refresh_status()
 
     def action_cancel(self) -> None:
         self.run_worker(self._cancel_flow(), exclusive=True)