diff --git a/README.md b/README.md
index 575e243203..1fa12a450e 100644
--- a/README.md
+++ b/README.md
@@ -2510,8 +2510,9 @@ A keyboard-driven chat surface with a streaming transcript (Markdown-rendered an
 During an active run, `/resume`, `/threads`, and `/switch` ask you to wait before
 switching conversations. An invalid `/resume` reference displays an error without
 closing the TUI or changing the current conversation.
-After an interrupt and conversation switch, late stream actions from the previous
-thread cannot change the new conversation's display or run state.
+After an interrupt, late stream actions from the previous run cannot change the
+next run's display or status, even when both prompts use the same conversation.
+If a run cannot start, the TUI reports an error and returns to idle so you can retry.
 
 At the last composer row, `Down` leaves an unsent draft untouched unless you are
 browsing input history; after recalling history, it moves forward to restore your
diff --git a/backend/docs/TUI.md b/backend/docs/TUI.md
index a8337f6f12..4b6719cd45 100644
--- a/backend/docs/TUI.md
+++ b/backend/docs/TUI.md
@@ -111,8 +111,10 @@ semantics as elsewhere in DeerFlow). `/model` and `/threads` open modal pickers.
 opens the thread picker. During an active run, `/resume`, `/threads`, and `/switch`
 ask you to wait instead of switching away from in-flight output. An invalid
 `/resume` reference displays an error and leaves the current conversation intact.
-After an interrupt and conversation switch, late stream actions from the previous
-thread are discarded when they reach the UI.
+After an interrupt, late stream actions from the previous run are discarded when
+they reach the UI, even if the next prompt uses the same conversation. Its final
+usage, title, and completion event cannot replace those of the new run.
+If a run cannot start, the TUI reports an error and returns to idle so you can retry.
 
 Use `/goal <condition>` to set the active thread goal, `/goal` to show it, and
 `/goal clear` to clear it.
diff --git a/backend/packages/harness/deerflow/tui/AGENTS.md b/backend/packages/harness/deerflow/tui/AGENTS.md
index e721b9fd2c..b8e0e0d9ae 100644
--- a/backend/packages/harness/deerflow/tui/AGENTS.md
+++ b/backend/packages/harness/deerflow/tui/AGENTS.md
@@ -18,10 +18,17 @@ and when applying the picker callback, which may outlive the idle state in which
 it opened. Keep the current thread and transcript unchanged when rejecting a
 switch. Report invalid resume references as error rows without relaxing the
 canonical thread-id validation or terminating the app.
-Worker actions must carry their originating thread id into the UI callback.
-Check it against the displayed thread on delivery, not just before scheduling:
-an interrupt/switch can race with the worker's cancellation check. Drop actions
-for another thread before they reach the reducer or change streaming state.
+Reserve the run identity and busy state when accepting input, before its worker
+starts. Each run owns its cancellation flag; a later send must not reset it.
+If worker creation fails, cancel and clear that reserved run, restore idle state,
+and show a retryable error. Do not cancel a previously completed run's title write.
+Worker callbacks check both run identity and thread id on UI delivery, rejecting
+cancelled, completed, or superseded runs before updating the reducer. Keep the
+completed run's state available for its worker's title persistence check. An
+uncancelled, normally completed run still writes its title to its original
+thread after the UI moves on; keep that write off the UI thread. This does not
+order completed title writes, roll back a write already in progress, or force
+stop synchronous backend work.
 
 `InputHistory.down()` returns `None` when history navigation is inactive; the
 app must then leave the composer untouched, including its cursor and undo state.
diff --git a/backend/packages/harness/deerflow/tui/app.py b/backend/packages/harness/deerflow/tui/app.py
index 325735c0f6..30b3e4a3c8 100644
--- a/backend/packages/harness/deerflow/tui/app.py
+++ b/backend/packages/harness/deerflow/tui/app.py
@@ -9,7 +9,9 @@ thread via ``call_from_thread`` and folded into the reducer.
 from __future__ import annotations
 
 import uuid
+from dataclasses import dataclass, field
 from functools import partial
+from threading import Event
 
 from textual.app import App, ComposeResult
 from textual.binding import Binding
@@ -41,6 +43,15 @@ _HELP_KEYS = "Keys:  Enter send · PgUp/PgDn scroll transcript · Ctrl+C interru
 _HELP_TEXT = f"{format_command_help()}\n{_HELP_KEYS}"
 
 
+@dataclass(eq=False)
+class _Run:
+    """Delivery and cancellation state owned by one worker run."""
+
+    thread_id: str
+    cancelled: Event = field(default_factory=Event)
+    finished: bool = False
+
+
 _TRANSPARENT_CSS = """
 Screen,
 #header,
@@ -190,7 +201,7 @@ class DeerFlowTUI(App):
         self._skills = 0
         self._spinner_idx = 0
         self._streaming = False
-        self._cancelled = False
+        self._run: _Run | None = None
         self._skills_meta: list[dict] = []
         self._model_override: str | None = None
         self._palette_open = False
@@ -644,16 +655,27 @@ class DeerFlowTUI(App):
             return
         if self._conv_thread_id is None:
             self._conv_thread_id = str(uuid.uuid4())
-        self._cancelled = False
+        run = _Run(self._conv_thread_id)
+        self._run = run
+        self._streaming = True
         self._dispatch(UserSubmitted(text))
-        self.run_worker(
-            partial(self._stream_worker, text, self._conv_thread_id),
-            thread=True,
-            exclusive=True,
-            group="agent",
-        )
+        try:
+            self.run_worker(
+                partial(self._stream_worker, text, run),
+                thread=True,
+                exclusive=True,
+                group="agent",
+            )
+        except Exception:  # noqa: BLE001 - worker creation must release the busy reservation
+            run.cancelled.set()
+            self._run = None
+            self._streaming = False
+            self._dispatch(SystemMessage("Could not start the run. Please try again.", tone="error"))
 
-    def _stream_worker(self, text: str, thread_id: str) -> None:
+    def _stream_worker(self, text: str, run: _Run) -> None:
+        if run.cancelled.is_set():
+            return
+        thread_id = run.thread_id
         kwargs: dict = {}
         if self._model_override:
             kwargs["model_name"] = self._model_override
@@ -667,23 +689,27 @@ class DeerFlowTUI(App):
 
         latest_title: str | None = None
         for action in stream_actions(self.session.client, text, thread_id=thread_id, **kwargs):
-            if self._cancelled:
+            if run.cancelled.is_set():
+                break
+            if not self.call_from_thread(self._on_stream_action, run, action):
                 break
             if isinstance(action, ThreadTitle):
                 latest_title = action.title
-            self.call_from_thread(self._on_stream_action, thread_id, action)
 
         # Only persist a title for a run that completed normally — an interrupted
         # run may only have emitted the title middleware's first, truncated guess.
-        if writer is not None and latest_title and not self._cancelled:
+        if writer is not None and latest_title and run.finished and not run.cancelled.is_set():
             writer.set_title(thread_id, latest_title)
 
-    def _on_stream_action(self, thread_id: str, action) -> None:
-        # Interrupt/switch may happen after the worker's cancellation check.
-        # Validate the destination when Textual delivers the action to the UI.
-        if thread_id != self._conv_thread_id:
-            return
+    def _on_stream_action(self, run: _Run, action) -> bool:
+        # Cancellation and a new send can occur after the worker's check,
+        # including when both runs belong to the same conversation.
+        if self._run is not run or run.thread_id != self._conv_thread_id or run.cancelled.is_set() or run.finished:
+            return False
+        if isinstance(action, RunEnded):
+            run.finished = True
         self._on_action(action)
+        return True
 
     def _on_action(self, action) -> None:
         self.state = reduce(self.state, action)
@@ -720,7 +746,8 @@ class DeerFlowTUI(App):
             self._interrupt_run()
 
     def _interrupt_run(self) -> None:
-        self._cancelled = True
+        if self._run is not None:
+            self._run.cancelled.set()
         self.workers.cancel_group(self, "agent")
         self._streaming = False
         self.state = reduce(self.state, RunEnded())
diff --git a/backend/tests/test_tui_run_isolation.py b/backend/tests/test_tui_run_isolation.py
new file mode 100644
index 0000000000..ce9206d9d0
--- /dev/null
+++ b/backend/tests/test_tui_run_isolation.py
@@ -0,0 +1,346 @@
+"""Per-run delivery through the real Textual worker, pilot and view reducer."""
+
+import asyncio
+import threading
+
+import pytest
+
+from deerflow.client import StreamEvent
+from deerflow.tui.app import DeerFlowTUI
+from deerflow.tui.cli import LaunchPlan
+from deerflow.tui.session import Session
+from deerflow.tui.view_state import AssistantDelta, RunEnded, RunStarted, ThreadTitle
+
+
+class _Client:
+    def __init__(self):
+        self.release_old = threading.Event()
+        self.release_new = threading.Event()
+        self.new_started = threading.Event()
+        self.old_terminal_emitted = threading.Event()
+        self.calls = []
+
+    def list_models(self):
+        return {"models": []}
+
+    def list_skills(self, **kwargs):
+        return {"skills": []}
+
+    def list_threads(self, **kwargs):
+        return {"thread_list": [{"thread_id": "thread-a", "title": "A"}, {"thread_id": "thread-b", "title": "B"}]}
+
+    def stream(self, message, *, thread_id=None, **kwargs):
+        self.calls.append((message, thread_id))
+        old = message == "old question"
+        prefix = "old" if old else "new"
+        if not old:
+            yield StreamEvent(type="values", data={"title": "New title"})
+        yield StreamEvent(type="messages-tuple", data={"type": "ai", "content": f"{prefix}-before ", "id": f"{prefix}-answer"})
+        if not old:
+            self.new_started.set()
+        release = self.release_old if old else self.release_new
+        if not release.wait(10):
+            raise TimeoutError("Controlled stream was not released")
+        yield StreamEvent(type="messages-tuple", data={"type": "ai", "content": f"{prefix}-tail", "id": f"{prefix}-answer"})
+        if old:
+            yield StreamEvent(type="values", data={"title": "Old title"})
+            self.old_terminal_emitted.set()
+        yield StreamEvent(type="end", data={"usage": {"total_tokens": 111 if old else 222}})
+
+
+class _Writer:
+    def __init__(self):
+        self.titles = []
+
+    def ensure_created(self, *args, **kwargs):
+        pass
+
+    def set_title(self, thread_id, title):
+        self.titles.append((thread_id, title, threading.get_ident()))
+
+
+async def _settle(pilot, predicate):
+    for _ in range(200):
+        await pilot.pause()
+        if predicate():
+            return
+        await asyncio.sleep(0.01)
+    raise AssertionError("Expected controlled UI event did not arrive")
+
+
+async def _late_delivery(monkeypatch, gate, switch_thread=False, restart=True):
+    client, writer = _Client(), _Writer()
+    app = DeerFlowTUI(Session(client=client, writer=writer), LaunchPlan(mode="tui", thread_id="thread-a"))
+    queued, deliver, old_done, new_done = (threading.Event() for _ in range(4))
+    old_worker_id = None
+    original_worker, original_callback = app._stream_worker, app.call_from_thread
+
+    def observed_worker(text, *args):
+        nonlocal old_worker_id
+        if text == "old question":
+            old_worker_id = threading.get_ident()
+        try:
+            return original_worker(text, *args)
+        finally:
+            (old_done if text == "old question" else new_done).set()
+
+    def gated_callback(callback, *args):
+        action = args[-1]
+        is_old = threading.get_ident() == old_worker_id
+        pause = (gate == "delta" and isinstance(action, AssistantDelta) and action.text == "old-tail") or (gate == "title" and isinstance(action, ThreadTitle)) or (gate == "end" and isinstance(action, RunEnded))
+        if is_old and pause:
+            queued.set()
+            if not deliver.wait(10):
+                raise TimeoutError("Controlled UI delivery was not released")
+        return original_callback(callback, *args)
+
+    monkeypatch.setattr(app, "_stream_worker", observed_worker)
+    monkeypatch.setattr(app, "call_from_thread", gated_callback)
+    async with app.run_test() as pilot:
+        try:
+            await pilot.pause()
+            app.query_one("#composer").value = "old question"
+            await pilot.press("enter")
+            await _settle(pilot, lambda: app._streaming and any(row.kind == "assistant" for row in app.state.rows))
+            client.release_old.set()
+            await _settle(pilot, queued.is_set)
+            if gate == "end":
+                # The old source has emitted its terminal event; only UI delivery
+                # is pending, so this case needs no overlapping backend runs.
+                assert client.old_terminal_emitted.is_set()
+            await pilot.press("ctrl+c")
+            if restart:
+                if switch_thread:
+                    app.query_one("#composer").value = "/resume thread-b"
+                    await pilot.press("enter")
+                app.query_one("#composer").value = "new question"
+                await pilot.press("enter")
+                await _settle(pilot, lambda: client.new_started.is_set() and app._streaming)
+            deliver.set()
+            await _settle(pilot, old_done.is_set)
+            snapshot = {
+                "streaming": app._streaming,
+                "state_streaming": app.state.streaming,
+                "usage": app.state.usage,
+                "title": app.state.title,
+                "assistant_rows": [(row.id, row.text) for row in app.state.rows if row.kind == "assistant"],
+                "written_titles": list(writer.titles),
+            }
+            if restart:
+                assert not client.release_new.is_set()
+                client.release_new.set()
+                await _settle(pilot, new_done.is_set)
+                assert not app._streaming
+                assert app.state.usage == {"total_tokens": 222}
+                assert app.state.title == "New title"
+                assert any(row.kind == "assistant" and row.text == "new-before new-tail" for row in app.state.rows)
+            return snapshot, writer.titles
+        finally:
+            client.release_old.set()
+            client.release_new.set()
+            deliver.set()
+            await _settle(pilot, old_done.is_set)
+            if restart and client.new_started.is_set():
+                await _settle(pilot, new_done.is_set)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("gate", ["delta", "title", "end"])
+@pytest.mark.parametrize("switch_thread", [False, True])
+async def test_interrupted_run_does_not_change_new_run(monkeypatch, gate, switch_thread):
+    state, _ = await _late_delivery(monkeypatch, gate, switch_thread)
+    assert state["streaming"]
+    assert state["state_streaming"]
+    assert state["usage"] is None
+    assert state["title"] == "New title"
+    if gate == "delta":
+        assert not any("old-tail" in text for _, text in state["assistant_rows"])
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("switch_thread", [False, True])
+async def test_cancelled_worker_cannot_write_its_title_after_new_run(monkeypatch, switch_thread):
+    state, titles = await _late_delivery(monkeypatch, "end", switch_thread)
+    assert state["written_titles"] == []
+    assert [(thread_id, title) for thread_id, title, _ in titles] == [("thread-b" if switch_thread else "thread-a", "New title")]
+    assert titles[0][2] != threading.get_ident()
+
+
+@pytest.mark.asyncio
+async def test_interrupt_without_restart_discards_pending_delta(monkeypatch):
+    state, titles = await _late_delivery(monkeypatch, "delta", restart=False)
+    assert not state["streaming"]
+    assert state["assistant_rows"] == [("old-answer", "old-before ")]
+    assert titles == []
+
+
+@pytest.mark.asyncio
+async def test_send_reserves_busy_state_before_worker_starts(monkeypatch):
+    client = _Client()
+    app = DeerFlowTUI(Session(client=client), LaunchPlan(mode="tui", thread_id="thread-a"))
+    queued, release, done = (threading.Event() for _ in range(3))
+    original_worker = app._stream_worker
+
+    def delayed_worker(*args):
+        queued.set()
+        try:
+            if not release.wait(10):
+                raise TimeoutError("Worker start was not released")
+            return original_worker(*args)
+        finally:
+            done.set()
+
+    monkeypatch.setattr(app, "_stream_worker", delayed_worker)
+    async with app.run_test() as pilot:
+        try:
+            app.query_one("#composer").value = "old question"
+            await pilot.press("enter")
+            await _settle(pilot, queued.is_set)
+            assert app._streaming
+            app.query_one("#composer").value = "new question"
+            await pilot.press("enter")
+            assert any(row.kind == "system" and "Still working" in row.text for row in app.state.rows)
+            await pilot.press("ctrl+c")
+            release.set()
+            client.release_old.set()
+            await _settle(pilot, done.is_set)
+            assert client.calls == []
+        finally:
+            release.set()
+            client.release_old.set()
+            client.release_new.set()
+            await _settle(pilot, done.is_set)
+
+
+@pytest.mark.asyncio
+async def test_worker_start_failure_restores_idle_state_and_allows_retry(monkeypatch):
+    client, writer = _Client(), _Writer()
+    client.release_new.set()
+    app = DeerFlowTUI(Session(client=client, writer=writer), LaunchPlan(mode="tui", thread_id="thread-a"))
+    failed_runs = []
+    original_start = app.run_worker
+
+    def fail_first_agent_worker(*args, **kwargs):
+        if kwargs.get("group") == "agent" and not failed_runs:
+            failed_runs.append(app._run)
+            raise RuntimeError("private worker startup detail")
+        return original_start(*args, **kwargs)
+
+    monkeypatch.setattr(app, "run_worker", fail_first_agent_worker)
+    async with app.run_test() as pilot:
+        app.query_one("#composer").value = "old question"
+        await pilot.press("enter")
+        assert not app._streaming
+        assert not app.state.streaming
+        assert app._run is None
+        assert failed_runs[0].cancelled.is_set()
+        assert any(row.kind == "system" and row.tone == "error" for row in app.state.rows)
+        assert "private worker startup detail" not in str(app.state.rows)
+        assert client.calls == []
+
+        idle_state = app.state
+        for action in [RunStarted(), AssistantDelta(id="failed-answer", text="stale"), ThreadTitle("Failed title"), RunEnded(usage={"total_tokens": 999})]:
+            assert not app._on_stream_action(failed_runs[0], action)
+            assert app.state == idle_state
+            assert not app._streaming
+        app._stream_worker("old question", failed_runs[0])
+        assert client.calls == []
+
+        app.query_one("#composer").value = "new question"
+        await pilot.press("enter")
+        await _settle(pilot, lambda: writer.titles and not app._streaming)
+        assert client.calls == [("new question", "thread-a")]
+        assert app.state.usage == {"total_tokens": 222}
+        assert any(row.kind == "assistant" and row.text == "new-before new-tail" for row in app.state.rows)
+        assert not any(row.kind == "system" and "Still working" in row.text for row in app.state.rows)
+
+
+@pytest.mark.asyncio
+async def test_completed_run_rejects_a_duplicate_terminal_callback(monkeypatch):
+    client, writer = _Client(), _Writer()
+    client.release_old.set()
+    app = DeerFlowTUI(Session(client=client, writer=writer), LaunchPlan(mode="tui", thread_id="thread-a"))
+    delivered = []
+    original = app.call_from_thread
+
+    def remember(callback, *args):
+        result = original(callback, *args)
+        if isinstance(args[-1], RunEnded):
+            delivered.append((callback, args))
+        return result
+
+    monkeypatch.setattr(app, "call_from_thread", remember)
+    async with app.run_test() as pilot:
+        app.query_one("#composer").value = "old question"
+        await pilot.press("enter")
+        await _settle(pilot, lambda: delivered and writer.titles)
+        assert app.state.usage == {"total_tokens": 111}
+        callback, args = delivered[0]
+        completed = app.state
+        for action in [RunStarted(), AssistantDelta(id="old-answer", text="stale"), ThreadTitle("Stale title"), RunEnded(usage={"total_tokens": 999})]:
+            callback(*args[:-1], action)
+            assert app.state == completed
+            assert not app._streaming
+        assert [(thread_id, title) for thread_id, title, _ in writer.titles] == [("thread-a", "Old title")]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("next_action", ["switch", "send", "failed-send"])
+async def test_completed_run_keeps_title_write_after_ui_moves_on(monkeypatch, next_action):
+    client, writer = _Client(), _Writer()
+    client.release_old.set()
+    app = DeerFlowTUI(Session(client=client, writer=writer), LaunchPlan(mode="tui", thread_id="thread-a"))
+    terminal_delivered, release, old_done, new_done = (threading.Event() for _ in range(4))
+    original_worker, original_callback = app._stream_worker, app.call_from_thread
+    old_worker_id = None
+
+    def observed_worker(text, *args):
+        nonlocal old_worker_id
+        if text == "old question":
+            old_worker_id = threading.get_ident()
+        try:
+            return original_worker(text, *args)
+        finally:
+            (old_done if text == "old question" else new_done).set()
+
+    def pause_after_terminal(callback, *args):
+        result = original_callback(callback, *args)
+        if isinstance(args[-1], RunEnded) and threading.get_ident() == old_worker_id:
+            terminal_delivered.set()
+            if not release.wait(10):
+                raise TimeoutError("Completed worker was not released")
+        return result
+
+    monkeypatch.setattr(app, "_stream_worker", observed_worker)
+    monkeypatch.setattr(app, "call_from_thread", pause_after_terminal)
+    async with app.run_test() as pilot:
+        try:
+            app.query_one("#composer").value = "old question"
+            await pilot.press("enter")
+            await _settle(pilot, terminal_delivered.is_set)
+            assert not app._streaming
+            if next_action == "failed-send":
+
+                def fail_to_start(*args, **kwargs):
+                    raise RuntimeError("Cannot start the next worker")
+
+                monkeypatch.setattr(app, "run_worker", fail_to_start)
+            app.query_one("#composer").value = "/resume thread-b" if next_action == "switch" else "new question"
+            await pilot.press("enter")
+            if next_action == "failed-send":
+                assert not app._streaming
+                assert not app.state.streaming
+                assert app._run is None
+                assert app.state.usage == {"total_tokens": 111}
+            if next_action == "send":
+                await _settle(pilot, client.new_started.is_set)
+            release.set()
+            await _settle(pilot, old_done.is_set)
+            assert [(thread_id, title) for thread_id, title, _ in writer.titles] == [("thread-a", "Old title")]
+            assert app._streaming == (next_action == "send")
+        finally:
+            release.set()
+            client.release_new.set()
+            await _settle(pilot, old_done.is_set)
+            if client.new_started.is_set():
+                await _settle(pilot, new_done.is_set)