diff --git a/python/packages/core/agent_framework/_harness/_background_agents.py b/python/packages/core/agent_framework/_harness/_background_agents.py
index 631a8bd7705..fb93292aec3 100644
--- a/python/packages/core/agent_framework/_harness/_background_agents.py
+++ b/python/packages/core/agent_framework/_harness/_background_agents.py
@@ -55,7 +55,12 @@ class BackgroundTaskStatus(str, Enum):
 
 @experimental(feature_id=ExperimentalFeature.HARNESS)
 class BackgroundTaskInfo(SerializationMixin):
-    """Metadata for a single background task."""
+    """Metadata for a single background task.
+
+    ``generation`` identifies which execution of the task this record describes. It starts at 1
+    and increases every time the task is continued, so a restored older snapshot of the session
+    is never reconciled against a later execution that reused the same task ID.
+    """
 
     DEFAULT_EXCLUDE: ClassVar[set[str]] = set()
 
@@ -65,7 +70,8 @@ class BackgroundTaskInfo(SerializationMixin):
     status: BackgroundTaskStatus
     result_text: str | None
     error_text: str | None
-    __slots__ = ("agent_name", "description", "error_text", "id", "result_text", "status")
+    generation: int
+    __slots__ = ("agent_name", "description", "error_text", "generation", "id", "result_text", "status")
 
     def __init__(
         self,
@@ -75,6 +81,7 @@ def __init__(
         status: BackgroundTaskStatus = BackgroundTaskStatus.RUNNING,
         result_text: str | None = None,
         error_text: str | None = None,
+        generation: int = 1,
     ) -> None:
         """Initialize a background task info entry."""
         self.id = id
@@ -83,6 +90,7 @@ def __init__(
         self.status = status
         self.result_text = result_text
         self.error_text = error_text
+        self.generation = generation
 
     def to_dict(self, *, exclude: set[str] | None = None, exclude_none: bool = True) -> dict[str, Any]:
         """Serialize for session state persistence."""
@@ -92,6 +100,7 @@ def to_dict(self, *, exclude: set[str] | None = None, exclude_none: bool = True)
             "agent_name": self.agent_name,
             "description": self.description,
             "status": self.status.value,
+            "generation": self.generation,
         }
         if not exclude_none or self.result_text is not None:
             data["result_text"] = self.result_text
@@ -109,24 +118,51 @@ def from_dict(cls, data: MutableMapping[str, Any], **kwargs: Any) -> BackgroundT
             status=BackgroundTaskStatus(data["status"]),
             result_text=data.get("result_text"),
             error_text=data.get("error_text"),
+            generation=data.get("generation", 1),
         )
 
 
 @dataclass
 class _RuntimeState:
-    """Non-serializable per-session runtime state for background tasks."""
+    """Non-serializable per-session runtime state for background tasks.
+
+    ``in_flight_tasks`` holds the latest execution started for each task ID and
+    ``task_generations`` records which generation that execution belongs to. Lookups made on
+    behalf of a persisted :class:`BackgroundTaskInfo` go through :meth:`get_execution` so that a
+    record only ever sees the execution it was created for.
+    """
 
     in_flight_tasks: dict[int, asyncio.Task[AgentResponse[Any]]] = field(default_factory=lambda: {})
+    task_generations: dict[int, int] = field(default_factory=lambda: {})
     background_sessions: dict[int, AgentSession] = field(default_factory=lambda: {})
     closed: bool = False
 
-    def track_task(self, task_id: int, task: asyncio.Task[AgentResponse[Any]]) -> None:
+    def track_task(self, task_id: int, task: asyncio.Task[AgentResponse[Any]], *, generation: int = 1) -> None:
         """Track a background task if this runtime is still open."""
         if self.closed:
             task.cancel()
             raise RuntimeError("Session runtime is closed; cannot start background task.")
 
         self.in_flight_tasks[task_id] = task
+        self.task_generations[task_id] = generation
+
+    def current_generation(self, task_id: int) -> int | None:
+        """Return the generation of the tracked execution for a task ID, if any."""
+        if task_id not in self.in_flight_tasks:
+            return None
+        return self.task_generations.get(task_id, 1)
+
+    def get_execution(self, task_id: int, generation: int) -> asyncio.Task[AgentResponse[Any]] | None:
+        """Return the tracked execution for a task ID only if it belongs to the given generation."""
+        if self.current_generation(task_id) != generation:
+            return None
+        return self.in_flight_tasks.get(task_id)
+
+    def forget_task(self, task_id: int) -> None:
+        """Drop all runtime references for a task ID."""
+        self.in_flight_tasks.pop(task_id, None)
+        self.task_generations.pop(task_id, None)
+        self.background_sessions.pop(task_id, None)
 
 
 # ---------------------------------------------------------------------------
@@ -215,9 +251,27 @@ def _save_tasks(state: dict[str, Any], tasks: list[BackgroundTaskInfo]) -> None:
 def _finalize_task(
     task_info: BackgroundTaskInfo,
     completed_task: asyncio.Task[AgentResponse[Any]],
-    runtime: _RuntimeState,
 ) -> None:
-    """Extract results from a completed asyncio task and update task info."""
+    """Extract results from a completed asyncio task and update task info.
+
+    The finished :class:`asyncio.Task` is intentionally **not** removed from
+    ``runtime.in_flight_tasks`` here.  Session state is persisted by the host
+    only after the agent run completes; if the run fails, is stopped early, or
+    the host restores an older session snapshot, the persisted record will still
+    say ``RUNNING`` while this in-memory task no longer exists.  On the next
+    call to :func:`_refresh_task_state` a missing entry would be misidentified
+    as ``LOST``, discarding a result that is already available.
+
+    Because the retained entry is keyed by task ID, callers must only pass the
+    execution matching ``task_info.generation`` (see
+    :meth:`_RuntimeState.get_execution`); a later continuation of the same task
+    ID must never be applied to an older record.
+
+    Tasks are removed from ``in_flight_tasks`` by two callers:
+
+    - :func:`background_agents_clear_completed_task` — explicit caller-driven cleanup.
+    - :meth:`BackgroundAgentsProvider.release_session` — bulk cleanup on session teardown.
+    """
     if completed_task.cancelled():
         task_info.status = BackgroundTaskStatus.FAILED
         task_info.error_text = "Task was canceled."
@@ -229,9 +283,6 @@ def _finalize_task(
         else:
             task_info.status = BackgroundTaskStatus.COMPLETED
             task_info.result_text = completed_task.result().text
-    # A stale waiter may observe a task ID already reused by a continuation.
-    if runtime.in_flight_tasks.get(task_info.id) is completed_task:
-        runtime.in_flight_tasks.pop(task_info.id, None)
 
 
 def _refresh_task_state(
@@ -245,14 +296,16 @@ def _refresh_task_state(
         if task_info.status != BackgroundTaskStatus.RUNNING:
             continue
 
-        in_flight = runtime.in_flight_tasks.get(task_info.id)
+        # A missing execution, or one from a different generation (the task was continued from
+        # another snapshot of this session), means this record's execution is no longer tracked.
+        in_flight = runtime.get_execution(task_info.id, task_info.generation)
         if in_flight is None:
             task_info.status = BackgroundTaskStatus.LOST
             changed = True
             continue
 
         if in_flight.done():
-            _finalize_task(task_info, in_flight, runtime)
+            _finalize_task(task_info, in_flight)
             changed = True
 
     if changed:
@@ -399,6 +452,7 @@ async def release_session(
                     await asyncio.gather(*completed, return_exceptions=True)
         finally:
             runtime.in_flight_tasks.clear()
+            runtime.task_generations.clear()
             runtime.background_sessions.clear()
 
             if self._runtime.get(session_id) is runtime:
@@ -528,12 +582,16 @@ async def background_agents_wait_for_first_completion(task_ids: list[int]) -> st
             if not task_ids:
                 return "Error: No task IDs provided."
 
-            # Collect in-flight tasks matching the requested IDs.
+            # Collect the executions that the requested running records refer to.
             waitable: list[tuple[int, asyncio.Task[AgentResponse[Any]]]] = []
+            tasks = _get_tasks(provider_state)
+            running = {t.id: t.generation for t in tasks if t.status == BackgroundTaskStatus.RUNNING}
             for tid in task_ids:
-                in_flight = runtime.in_flight_tasks.get(tid)
-                if in_flight is not None:
-                    waitable.append((tid, in_flight))
+                generation = running.get(tid)
+                if generation is not None:
+                    in_flight = runtime.get_execution(tid, generation)
+                    if in_flight is not None:
+                        waitable.append((tid, in_flight))
 
             if not waitable:
                 # Refresh state to catch any that completed.
@@ -572,9 +630,10 @@ async def background_agents_wait_for_first_completion(task_ids: list[int]) -> st
             tasks = _get_tasks(provider_state)
             task_info = next((t for t in tasks if t.id == completed_id), None)
             if task_info is not None and completed_task is not None:
-                # Report the waited-on execution without overwriting a newer one.
-                is_current_task = runtime.in_flight_tasks.get(task_info.id) is completed_task
-                _finalize_task(task_info, completed_task, runtime)
+                # Report the waited-on execution without overwriting a newer one: persist only if the
+                # record's generation still refers to the execution this waiter observed.
+                is_current_task = runtime.get_execution(task_info.id, task_info.generation) is completed_task
+                _finalize_task(task_info, completed_task)
                 if is_current_task:
                     _save_tasks(provider_state, tasks)
                     _save_provider_state(session, provider_state, source_id=source_id)
@@ -642,13 +701,26 @@ def background_agents_continue_task(task_id: int, text: str) -> str:
             if sub_session is None:
                 return f"Error: Session for task {task_id} is no longer available."
 
+            # This record is not running, so a pending tracked execution must be a newer one started
+            # from another snapshot of this session. Running both against one child session is unsafe.
+            current = runtime.in_flight_tasks.get(task_id)
+            if current is not None and not current.done():
+                return (
+                    f"Error: Task {task_id} has a newer execution that is still running. "
+                    "Wait for it to complete before continuing."
+                )
+
             bg_agent = self._agents[key]
+            # Always move past any generation this runtime has seen so the new execution can't be
+            # mistaken for one referenced by another snapshot.
+            generation = max(task_info.generation, runtime.current_generation(task_id) or 0) + 1
 
             async_task = asyncio.create_task(_run_agent(bg_agent.run(text, session=sub_session)))
             try:
-                runtime.track_task(task_id, async_task)
+                runtime.track_task(task_id, async_task, generation=generation)
             except RuntimeError as exc:
                 return f"Error: {exc}"
+            task_info.generation = generation
             task_info.status = BackgroundTaskStatus.RUNNING
             task_info.result_text = None
             task_info.error_text = None
@@ -678,9 +750,12 @@ def background_agents_clear_completed_task(task_id: int) -> str:
             tasks = [t for t in tasks if t.id != task_id]
             _save_tasks(provider_state, tasks)
 
-            # Clean up runtime references.
-            runtime.in_flight_tasks.pop(task_id, None)
-            runtime.background_sessions.pop(task_id, None)
+            # Release runtime references unless a newer execution of this task ID exists (started from
+            # another snapshot of this session); that execution must stay tracked so it can still be
+            # retrieved, cancelled, or drained by release_session.
+            current_generation = runtime.current_generation(task_id)
+            if current_generation is None or current_generation <= task_info.generation:
+                runtime.forget_task(task_id)
 
             _save_provider_state(session, provider_state, source_id=source_id)
             return f"Task {task_id} cleared."
diff --git a/python/packages/core/tests/core/test_harness_background_agents.py b/python/packages/core/tests/core/test_harness_background_agents.py
index 7e14c074d6f..1904b5100f8 100644
--- a/python/packages/core/tests/core/test_harness_background_agents.py
+++ b/python/packages/core/tests/core/test_harness_background_agents.py
@@ -390,7 +390,11 @@ async def counted_wait(*args: Any, **kwargs: Any) -> Any:
         agent_release.set()
         assert await asyncio.gather(*waiters) == ["Task 1 finished with status: completed."] * 2
         assert await _invoke_tool(tools["background_agents_get_task_results"], task_id=1) == "shared result"
-        assert not provider._runtime[session.session_id].in_flight_tasks
+        # Finished executions are retained until cleared or released.
+        runtime = provider._runtime[session.session_id]
+        assert runtime.in_flight_tasks[1].done()
+        assert await _invoke_tool(tools["background_agents_clear_completed_task"], task_id=1) == "Task 1 cleared."
+        assert not runtime.in_flight_tasks
     finally:
         agent_release.set()
         await asyncio.gather(*waiters, return_exceptions=True)
@@ -1042,3 +1046,461 @@ async def fake_drain(*args: Any, **kwargs: Any) -> None:
 
     assert drain_called is False
     assert session.session_id not in provider._runtime
+
+
+# ---------------------------------------------------------------------------
+# Regression tests: finished tasks must survive until cleared or released
+# (Part of #8760 - single-process snapshot/state desync fix)
+# ---------------------------------------------------------------------------
+
+
+async def test_snapshot_restore_returns_real_result_not_lost() -> None:
+    """A session snapshot taken while a task is RUNNING must still return the
+    real result after the task finishes and the snapshot is restored.
+
+    Repro for the single-process gap: _finalize_task previously removed the
+    finished asyncio.Task from in_flight_tasks.  If the host persists session
+    state only after the run completes and a failure (or snapshot restoration)
+    occurs before that, the next refresh would find no local task and mark it
+    LOST, silently discarding the result.
+    """
+    provider = _make_provider(_FakeAgent("Worker", response_text="hello from worker"))
+    session = _make_session()
+    tools = await _get_tools(provider, session)
+
+    # Start a task and immediately snapshot the session state (task is RUNNING).
+    await _invoke_tool(
+        tools["background_agents_start_task"],
+        agent_name="Worker",
+        input="do work",
+        description="regression task",
+    )
+    snapshot = session.to_dict()  # captured while task is still RUNNING
+
+    # Let the task complete deterministically (no sleep).
+    runtime = provider._get_runtime(session)
+    task = runtime.in_flight_tasks[1]
+    await task  # wait for the asyncio.Task to finish
+
+    # Finalize by refreshing (simulates what before_run / get_task_results does).
+    tools2 = await _get_tools(provider, session)
+    result_before_restore = await _invoke_tool(tools2["background_agents_get_task_results"], task_id=1)
+    assert result_before_restore == "hello from worker"
+
+    # Now restore the pre-finalization snapshot into the same provider.
+    # This simulates the host restoring an older session state (e.g. after a
+    # failed persist or a multi-turn rollback).
+    session2 = AgentSession.from_dict(snapshot)
+    tools3 = await _get_tools(provider, session2)
+
+    # The task is still in in_flight_tasks (finished but not yet cleared).
+    # get_task_results must return the real result, not "Task state was lost".
+    result_after_restore = await _invoke_tool(tools3["background_agents_get_task_results"], task_id=1)
+    assert result_after_restore == "hello from worker", f"Expected real result but got: {result_after_restore!r}"
+
+    await provider.release_session(session)
+    await provider.release_session(session2)
+
+
+async def test_snapshot_restore_preserves_error_text_for_failed_task() -> None:
+    """Same as above but for a task that raises; error_text must survive restore."""
+    provider = _make_provider(_FakeAgent("Failer", should_fail=True))
+    session = _make_session()
+    tools = await _get_tools(provider, session)
+
+    await _invoke_tool(
+        tools["background_agents_start_task"],
+        agent_name="Failer",
+        input="trigger failure",
+        description="failure regression",
+    )
+    snapshot = session.to_dict()
+
+    runtime = provider._get_runtime(session)
+    task = runtime.in_flight_tasks[1]
+    with suppress(Exception):
+        await task
+
+    # Finalize via a refresh so task_info gets FAILED status.
+    tools2 = await _get_tools(provider, session)
+    await _invoke_tool(tools2["background_agents_get_task_results"], task_id=1)
+
+    # Restore snapshot and confirm the failure is still readable.
+    session2 = AgentSession.from_dict(snapshot)
+    tools3 = await _get_tools(provider, session2)
+    result = await _invoke_tool(tools3["background_agents_get_task_results"], task_id=1)
+    assert "Agent execution failed" in result, f"Unexpected result: {result!r}"
+
+    await provider.release_session(session)
+    await provider.release_session(session2)
+
+
+async def test_snapshot_restore_preserves_error_text_for_cancelled_task() -> None:
+    """Cancelled task error_text must survive snapshot restoration."""
+    provider = _make_provider(_FakeAgent("Slow", delay=60.0))
+    session = _make_session()
+    tools = await _get_tools(provider, session)
+
+    await _invoke_tool(
+        tools["background_agents_start_task"],
+        agent_name="Slow",
+        input="start slow task",
+        description="cancellation regression",
+    )
+    snapshot = session.to_dict()
+
+    runtime = provider._get_runtime(session)
+    task = runtime.in_flight_tasks[1]
+    task.cancel()
+    with suppress(asyncio.CancelledError):
+        await task
+
+    # Finalize via a refresh.
+    tools2 = await _get_tools(provider, session)
+    await _invoke_tool(tools2["background_agents_get_task_results"], task_id=1)
+
+    # Restore snapshot and confirm cancellation error survives.
+    session2 = AgentSession.from_dict(snapshot)
+    tools3 = await _get_tools(provider, session2)
+    result = await _invoke_tool(tools3["background_agents_get_task_results"], task_id=1)
+    assert "canceled" in result.lower(), f"Unexpected result: {result!r}"
+
+    await provider.release_session(session)
+    await provider.release_session(session2)
+
+
+async def test_clear_completed_task_removes_entry_from_in_flight_tasks() -> None:
+    """clear_completed_task must remove the finished task from in_flight_tasks."""
+    provider = _make_provider(_FakeAgent("Worker", response_text="done"))
+    session = _make_session()
+    tools = await _get_tools(provider, session)
+
+    await _invoke_tool(
+        tools["background_agents_start_task"],
+        agent_name="Worker",
+        input="work",
+        description="clear test",
+    )
+
+    runtime = provider._get_runtime(session)
+    await runtime.in_flight_tasks[1]
+
+    # Finalize and then clear.
+    tools2 = await _get_tools(provider, session)
+    await _invoke_tool(tools2["background_agents_get_task_results"], task_id=1)
+    clear_result = await _invoke_tool(tools2["background_agents_clear_completed_task"], task_id=1)
+
+    assert clear_result == "Task 1 cleared."
+    assert 1 not in runtime.in_flight_tasks, "Task should be removed by clear_completed_task"
+
+    await provider.release_session(session)
+
+
+async def test_release_session_clears_in_flight_tasks() -> None:
+    """release_session must clear all in_flight_tasks including finished ones."""
+    provider = _make_provider(_FakeAgent("Worker", response_text="done"))
+    session = _make_session()
+    tools = await _get_tools(provider, session)
+
+    await _invoke_tool(
+        tools["background_agents_start_task"],
+        agent_name="Worker",
+        input="work",
+        description="release test",
+    )
+
+    runtime = provider._get_runtime(session)
+    await runtime.in_flight_tasks[1]
+
+    # Finalize (task is now finished but still in in_flight_tasks).
+    tools2 = await _get_tools(provider, session)
+    await _invoke_tool(tools2["background_agents_get_task_results"], task_id=1)
+
+    assert 1 in runtime.in_flight_tasks, "Finished task should still be present before release"
+
+    await provider.release_session(session)
+
+    assert session.session_id not in provider._runtime
+    assert not runtime.in_flight_tasks
+
+
+async def test_wait_blocks_for_running_task_and_ignores_finished() -> None:
+    """wait_for_first_completion ignores finished retained tasks and waits for running ones."""
+    provider = _make_provider(
+        _FakeAgent("Fast", response_text="fast done"),
+        _FakeAgent("Slow", response_text="slow done", delay=0.02),
+    )
+    session = _make_session()
+    tools = await _get_tools(provider, session)
+
+    # 1. Start and finish task 1.
+    await _invoke_tool(
+        tools["background_agents_start_task"],
+        agent_name="Fast",
+        input="work",
+        description="task 1",
+    )
+    runtime = provider._get_runtime(session)
+    await runtime.in_flight_tasks[1]
+
+    # Finalize task 1 so its status is COMPLETED and its task is retained.
+    tools2 = await _get_tools(provider, session)
+    await _invoke_tool(tools2["background_agents_get_task_results"], task_id=1)
+
+    # 2. Start task 2 which takes longer.
+    await _invoke_tool(
+        tools2["background_agents_start_task"],
+        agent_name="Slow",
+        input="slow work",
+        description="task 2",
+    )
+
+    # 3. wait([1, 2]) must block for running task 2 and NOT return finished task 1.
+    wait_result = await _invoke_tool(
+        tools2["background_agents_wait_for_first_completion"],
+        task_ids=[1, 2],
+    )
+    assert "Task 2 finished with status: completed." in wait_result
+    assert "Task 1" not in wait_result
+
+    # 4. wait([1]) on finished task returns "not running; current status" text.
+    finished_wait_result = await _invoke_tool(
+        tools2["background_agents_wait_for_first_completion"],
+        task_ids=[1],
+    )
+    assert "Task 1 is not running; current status: completed." in finished_wait_result
+
+    await provider.release_session(session)
+
+
+async def test_clear_completed_task_does_not_clear_in_flight_continuation() -> None:
+    """Restoring an older completed snapshot and clearing task 1 must not clear a running continuation."""
+
+    class _ContinuationAgent(_FakeAgent):
+        def __init__(self, name: str = "Worker") -> None:
+            super().__init__(name, response_text="done")
+            self.release = asyncio.Event()
+            self._runs = 0
+
+        async def run(
+            self, messages: Any = None, *, stream: bool = False, session: Any = None, **kwargs: Any
+        ) -> AgentResponse[Any]:
+            self._runs += 1
+            if self._runs > 1:
+                await self.release.wait()
+            return await super().run(messages, stream=stream, session=session, **kwargs)
+
+    agent = _ContinuationAgent("Worker")
+    provider = _make_provider(agent)
+    session = _make_session()
+    tools = await _get_tools(provider, session)
+
+    # 1. Start task 1 and let it complete.
+    await _invoke_tool(
+        tools["background_agents_start_task"],
+        agent_name="Worker",
+        input="first run",
+        description="task 1",
+    )
+    runtime = provider._get_runtime(session)
+    await runtime.in_flight_tasks[1]
+
+    # Finalize task 1 so its status is COMPLETED.
+    tools2 = await _get_tools(provider, session)
+    await _invoke_tool(tools2["background_agents_get_task_results"], task_id=1)
+
+    # 2. Snapshot the completed state.
+    snapshot = session.to_dict()
+
+    # 3. continue_task with slow agent.
+    continue_msg = await _invoke_tool(tools2["background_agents_continue_task"], task_id=1, text="second run")
+    assert "continued" in continue_msg.lower()
+
+    continuation_task = runtime.in_flight_tasks[1]
+    assert not continuation_task.done()
+
+    # 4. Restore snapshot into a new AgentSession.
+    session2 = AgentSession.from_dict(snapshot)
+    tools3 = await _get_tools(provider, session2)
+
+    # 5. Call clear_completed_task(1) on restored session.
+    clear_msg = await _invoke_tool(tools3["background_agents_clear_completed_task"], task_id=1)
+    assert clear_msg == "Task 1 cleared."
+
+    # 6. Continuation is still in in_flight_tasks.
+    assert 1 in runtime.in_flight_tasks
+    assert runtime.in_flight_tasks[1] is continuation_task
+    assert not continuation_task.done()
+
+    # 7. release_session(cancel_running=True) cancels it.
+    await provider.release_session(session, cancel_running=True)
+    assert continuation_task.cancelled()
+    assert not runtime.in_flight_tasks
+
+
+# ---------------------------------------------------------------------------
+# Regression tests: execution identity (generation) for reused task IDs
+# ---------------------------------------------------------------------------
+
+
+class _ScriptedAgent(_FakeAgent):
+    """Agent whose N-th run returns ``responses[N]`` and optionally waits on a gate or fails."""
+
+    def __init__(self, name: str, responses: list[str], *, fail_runs: set[int] | None = None) -> None:
+        super().__init__(name)
+        self.responses = responses
+        self.fail_runs = fail_runs or set()
+        self.gates: dict[int, asyncio.Event] = {}
+        self.runs = 0
+
+    async def run(
+        self, messages: Any = None, *, stream: bool = False, session: Any = None, **kwargs: Any
+    ) -> AgentResponse[Any]:
+        index = self.runs
+        self.runs += 1
+        gate = self.gates.get(index)
+        if gate is not None:
+            await gate.wait()
+        if index in self.fail_runs:
+            raise RuntimeError(f"run {index} failed")
+        return AgentResponse(messages=[Message(role="assistant", contents=[self.responses[index]])])
+
+
+def test_task_info_generation_round_trip_and_legacy_default() -> None:
+    """generation is persisted, and state written before it existed loads as generation 1."""
+    info = BackgroundTaskInfo(id=1, agent_name="A", description="d", generation=3)
+    assert BackgroundTaskInfo.from_dict(info.to_dict()).generation == 3
+
+    legacy = {"id": 1, "agent_name": "A", "description": "d", "status": "running"}
+    assert BackgroundTaskInfo.from_dict(legacy).generation == 1
+
+
+@pytest.mark.parametrize("continuation_fails", [False, True])
+async def test_restored_running_snapshot_does_not_receive_continuation_outcome(continuation_fails: bool) -> None:
+    """A stale RUNNING snapshot must not be finalized with a later continuation's result or error."""
+    agent = _ScriptedAgent("Worker", ["first result", "second result"], fail_runs={1} if continuation_fails else None)
+    provider = _make_provider(agent)
+    session = _make_session()
+    tools = await _get_tools(provider, session)
+
+    await _invoke_tool(tools["background_agents_start_task"], agent_name="Worker", input="go", description="t")
+    stale_snapshot = session.to_dict()  # generation 1, RUNNING
+    runtime = provider._get_runtime(session)
+    await runtime.in_flight_tasks[1]
+
+    # The live session retrieves generation 1 and continues the task (generation 2).
+    assert await _invoke_tool(tools["background_agents_get_task_results"], task_id=1) == "first result"
+    await _invoke_tool(tools["background_agents_continue_task"], task_id=1, text="again")
+    continuation = runtime.in_flight_tasks[1]
+    with suppress(Exception):
+        await continuation
+
+    # The host restores the snapshot taken while generation 1 was running.
+    restored = AgentSession.from_dict(stale_snapshot)
+    restored_tools = await _get_tools(provider, restored)
+    stale_result = await _invoke_tool(restored_tools["background_agents_get_task_results"], task_id=1)
+    assert stale_result == "Task state was lost (reference unavailable)."
+
+    # The live session still sees its own continuation outcome, and it is still tracked.
+    live_result = await _invoke_tool(tools["background_agents_get_task_results"], task_id=1)
+    assert live_result == ("Task failed: run 1 failed" if continuation_fails else "second result")
+    assert runtime.in_flight_tasks[1] is continuation
+
+    await provider.release_session(session)
+
+
+async def test_restored_running_snapshot_does_not_wait_on_continuation() -> None:
+    """Waiting from a stale snapshot must not block on (or finalize) a newer running continuation."""
+    agent = _ScriptedAgent("Worker", ["first result", "second result"])
+    agent.gates[1] = asyncio.Event()
+    provider = _make_provider(agent)
+    session = _make_session()
+    tools = await _get_tools(provider, session)
+
+    await _invoke_tool(tools["background_agents_start_task"], agent_name="Worker", input="go", description="t")
+    stale_snapshot = session.to_dict()
+    runtime = provider._get_runtime(session)
+    await runtime.in_flight_tasks[1]
+    await _invoke_tool(tools["background_agents_get_task_results"], task_id=1)
+    await _invoke_tool(tools["background_agents_continue_task"], task_id=1, text="again")
+    continuation = runtime.in_flight_tasks[1]
+
+    restored = AgentSession.from_dict(stale_snapshot)
+    restored_tools = await _get_tools(provider, restored)
+    wait_result = await _invoke_tool(restored_tools["background_agents_wait_for_first_completion"], task_ids=[1])
+    assert wait_result == "Task 1 is not running; current status: lost."
+    assert runtime.in_flight_tasks[1] is continuation
+    assert not continuation.done()
+
+    agent.gates[1].set()
+    wait_result = await _invoke_tool(tools["background_agents_wait_for_first_completion"], task_ids=[1])
+    assert wait_result == "Task 1 finished with status: completed."
+    assert await _invoke_tool(tools["background_agents_get_task_results"], task_id=1) == "second result"
+
+    await provider.release_session(session)
+
+
+async def test_continue_from_stale_snapshot_rejected_while_newer_execution_running() -> None:
+    """Continuing an old completed snapshot must not overwrite a newer running continuation."""
+    agent = _ScriptedAgent("Worker", ["first result", "second result", "third result"])
+    agent.gates[1] = asyncio.Event()
+    provider = _make_provider(agent)
+    session = _make_session()
+    tools = await _get_tools(provider, session)
+
+    await _invoke_tool(tools["background_agents_start_task"], agent_name="Worker", input="go", description="t")
+    runtime = provider._get_runtime(session)
+    await runtime.in_flight_tasks[1]
+    await _invoke_tool(tools["background_agents_get_task_results"], task_id=1)
+    completed_snapshot = session.to_dict()  # generation 1, COMPLETED
+    await _invoke_tool(tools["background_agents_continue_task"], task_id=1, text="again")
+    continuation = runtime.in_flight_tasks[1]
+
+    restored = AgentSession.from_dict(completed_snapshot)
+    restored_tools = await _get_tools(provider, restored)
+    result = await _invoke_tool(restored_tools["background_agents_continue_task"], task_id=1, text="stale")
+    assert result.startswith("Error: Task 1 has a newer execution that is still running.")
+    assert runtime.in_flight_tasks[1] is continuation
+    assert runtime.task_generations[1] == 2
+
+    await provider.release_session(session, cancel_running=True)
+    assert continuation.cancelled()
+
+
+async def test_continue_from_stale_snapshot_starts_newer_generation() -> None:
+    """Continuing from an older snapshot after a newer execution finished must use a fresh generation."""
+    agent = _ScriptedAgent("Worker", ["first result", "second result", "third result"])
+    provider = _make_provider(agent)
+    session = _make_session()
+    tools = await _get_tools(provider, session)
+
+    await _invoke_tool(tools["background_agents_start_task"], agent_name="Worker", input="go", description="t")
+    runtime = provider._get_runtime(session)
+    await runtime.in_flight_tasks[1]
+    await _invoke_tool(tools["background_agents_get_task_results"], task_id=1)
+    completed_snapshot = session.to_dict()  # generation 1, COMPLETED
+    await _invoke_tool(tools["background_agents_continue_task"], task_id=1, text="again")
+    await runtime.in_flight_tasks[1]
+    assert await _invoke_tool(tools["background_agents_get_task_results"], task_id=1) == "second result"
+
+    restored = AgentSession.from_dict(completed_snapshot)
+    restored_tools = await _get_tools(provider, restored)
+    await _invoke_tool(restored_tools["background_agents_continue_task"], task_id=1, text="from snapshot")
+    third = runtime.in_flight_tasks[1]
+    assert runtime.task_generations[1] == 3
+    assert restored.state["background_agents"]["tasks"][0]["generation"] == 3
+    await third
+    assert await _invoke_tool(restored_tools["background_agents_get_task_results"], task_id=1) == "third result"
+
+    # The live session (generation 2) clears its record without dropping generation 3.
+    assert await _invoke_tool(tools["background_agents_clear_completed_task"], task_id=1) == "Task 1 cleared."
+    assert runtime.in_flight_tasks[1] is third
+    assert 1 in runtime.background_sessions
+
+    # Clearing the record that owns the latest generation releases the runtime references.
+    assert await _invoke_tool(restored_tools["background_agents_clear_completed_task"], task_id=1) == "Task 1 cleared."
+    assert 1 not in runtime.in_flight_tasks
+    assert 1 not in runtime.task_generations
+    assert 1 not in runtime.background_sessions
+
+    await provider.release_session(session)