diff --git a/docs/decisions/0027-hosting-channels.md b/docs/decisions/0027-hosting-channels.md
index 36d4f14b76a..29d09c045e1 100644
--- a/docs/decisions/0027-hosting-channels.md
+++ b/docs/decisions/0027-hosting-channels.md
@@ -204,7 +204,8 @@ safe to use:
   transient execution.
 
 A `SessionStore` stores `session_id -> AgentSession`, but it does not create sessions. `AgentState` resolves the agent
-target and creates the session on first use:
+target and creates the session on first use. Reads return independent working copies so running from one continuation
+point does not mutate the stored snapshot or another simultaneous branch:
 
 For agent targets:
 
@@ -227,6 +228,11 @@ await state.set_session(response_id, session)
 `agent.run(...)` may update the session object (for example, with service continuation state), so the explicit store call
 belongs after the run, not before it.
 
+Response ids are immutable continuation points, so simultaneous callers can branch from one `previous_response_id` and
+store their completed sessions under different new response ids. A stable `conversation_id` is a mutable head: the app
+must explicitly update it after the run and provide single-writer coordination. The hosting state helper does not lock
+an entire run or resolve concurrent updates to that stable key.
+
 The session id is a partition key, not proof of identity. App or platform code must authenticate and authorize any
 externally supplied key before using it.
 
diff --git a/docs/specs/002-python-hosting-channels.md b/docs/specs/002-python-hosting-channels.md
index 5101982ad8e..e95ec4b1f53 100644
--- a/docs/specs/002-python-hosting-channels.md
+++ b/docs/specs/002-python-hosting-channels.md
@@ -178,6 +178,9 @@ The target may be:
 - `await get_target()`;
 - synchronous `target` only after a target is already available/resolved.
 
+A workflow instance permits one active run. Concurrent hosts use a factory or
+builder with `cache_target=False` to resolve a fresh instance per run.
+
 Workflow checkpointing uses Agent Framework's existing `CheckpointStorage` abstraction directly. Apps that need
 per-session workflow resume should keep an app-owned cursor such as `session_id -> checkpoint_id`. When the app uses
 file-backed cursor storage, the file-based checkpoint storage should share the same app storage root and should be
diff --git a/python/packages/hosting-responses/README.md b/python/packages/hosting-responses/README.md
index 797fcd5ca7d..d08d0c162c5 100644
--- a/python/packages/hosting-responses/README.md
+++ b/python/packages/hosting-responses/README.md
@@ -43,9 +43,18 @@ async def responses(body: dict = Body(...)) -> JSONResponse:
         session=session,
         options=run["options"],
     )
-    await state.set_session(response_id, session)
+    if body.get("conversation_id") == session_id:
+        # The app must serialize writers that advance this stable id.
+        await state.set_session(session_id, session)
+    else:
+        await state.set_session(response_id, session)
     return JSONResponse(responses_from_run(result, response_id=response_id, session_id=session_id))
 ```
 
+`previous_response_id` identifies an immutable continuation snapshot: multiple
+requests may branch from it and store their results under distinct new response
+ids. `conversation_id` is a mutable head instead; only one caller should
+advance it at a time. These helpers do not provide per-conversation locking.
+
 The base execution-state helpers live in
 [`agent-framework-hosting`](https://pypi.org/project/agent-framework-hosting/).
diff --git a/python/packages/hosting-responses/tests/hosting_responses/test_http_round_trip.py b/python/packages/hosting-responses/tests/hosting_responses/test_http_round_trip.py
index eb8b5b87e34..7ede2c152e5 100644
--- a/python/packages/hosting-responses/tests/hosting_responses/test_http_round_trip.py
+++ b/python/packages/hosting-responses/tests/hosting_responses/test_http_round_trip.py
@@ -13,6 +13,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 from collections.abc import AsyncIterator, Awaitable, Mapping
 from typing import Any, Literal, overload
@@ -56,6 +57,7 @@ class _StubAgent:
     def __init__(self) -> None:
         self.session_ids_seen: list[str | None] = []
         self.turn_counts: dict[str | None, int] = {}
+        self.session_turn_counts_seen: list[int] = []
 
     def create_session(self, *, session_id: str | None = None) -> AgentSession:
         return AgentSession(session_id=session_id)
@@ -97,7 +99,11 @@ def run(
         session_id = session.session_id if session is not None else None
         self.session_ids_seen.append(session_id)
         self.turn_counts[session_id] = self.turn_counts.get(session_id, 0) + 1
-        text = f"turn {self.turn_counts[session_id]} for session {session_id}"
+        session_turn = int(session.state.get("turn_count", 0)) + 1 if session is not None else 1
+        if session is not None:
+            session.state["turn_count"] = session_turn
+        self.session_turn_counts_seen.append(session_turn)
+        text = f"turn {session_turn} for session {session_id}"
 
         if stream:
 
@@ -124,6 +130,7 @@ async def responses(body: dict[str, Any] = Body(...)) -> JSONResponse | Streamin
         except ValueError as exc:
             raise HTTPException(status_code=400, detail=str(exc)) from exc
         session_id = responses_session_id(body)
+        conversation_id = session_id if body.get("conversation_id") == session_id else None
         response_id = create_response_id()
 
         target = await state.get_target()
@@ -142,7 +149,10 @@ async def stream_events() -> AsyncIterator[str]:
                     session_id=session_id,
                 ):
                     yield event
-                await state.set_session(response_id, session)
+                if conversation_id is not None:
+                    await state.set_session(conversation_id, session)
+                else:
+                    await state.set_session(response_id, session)
 
             return StreamingResponse(
                 stream_events(),
@@ -150,7 +160,10 @@ async def stream_events() -> AsyncIterator[str]:
             )
 
         result = await target.run(run["messages"], session=session)
-        await state.set_session(response_id, session)
+        if conversation_id is not None:
+            await state.set_session(conversation_id, session)
+        else:
+            await state.set_session(response_id, session)
         return JSONResponse(responses_from_run(result, response_id=response_id, session_id=session_id))
 
     return app
@@ -246,6 +259,23 @@ async def test_previous_response_id_chain_preserves_session_across_three_turns(s
         assert first_session_id is not None
         assert agent.session_ids_seen == [first_session_id] * 3
         assert agent.turn_counts[first_session_id] == 3
+        assert agent.session_turn_counts_seen == [1, 2, 3]
+
+    async def test_previous_response_id_supports_independent_branches(self) -> None:
+        agent = _StubAgent()
+        app = _build_app(agent)
+
+        root = await _post(app, {"input": "root"})
+        branch_one, branch_two = await asyncio.gather(
+            _post(app, {"input": "branch one", "previous_response_id": root.json()["id"]}),
+            _post(app, {"input": "branch two", "previous_response_id": root.json()["id"]}),
+        )
+        await asyncio.gather(
+            _post(app, {"input": "continue one", "previous_response_id": branch_one.json()["id"]}),
+            _post(app, {"input": "continue two", "previous_response_id": branch_two.json()["id"]}),
+        )
+
+        assert agent.session_turn_counts_seen == [1, 2, 2, 3, 3]
 
     async def test_conversation_id_preserves_session_across_turns(self) -> None:
         agent = _StubAgent()
@@ -258,6 +288,7 @@ async def test_conversation_id_preserves_session_across_turns(self) -> None:
 
         assert agent.session_ids_seen == ["conv_stable", "conv_stable"]
         assert agent.turn_counts["conv_stable"] == 2
+        assert agent.session_turn_counts_seen == [1, 2]
 
     async def test_unrelated_requests_get_independent_sessions(self) -> None:
         agent = _StubAgent()
diff --git a/python/packages/hosting/README.md b/python/packages/hosting/README.md
index 7dd927dfac8..0e376fd45c6 100644
--- a/python/packages/hosting/README.md
+++ b/python/packages/hosting/README.md
@@ -9,10 +9,12 @@ This package keeps Agent Framework state separate from web-framework concerns:
 - `WorkflowState` — resolves a workflow target, including direct `Workflow`
   instances, workflow factories, `WorkflowBuilder`, and orchestration builders.
 
-`SessionStore` is plain storage: `get`/`set`/`delete` by an app-selected id,
-nothing more. It does not know how to create a new value for an id it hasn't
-seen before — use `AgentState.get_or_create_session(...)` for that, since only
-the state object has both the store and the resolved target. Workflow
+`SessionStore` provides `get`/`set`/`delete` by an app-selected id. Each
+successful `get` returns an independent copy, so a run works from a snapshot
+instead of mutating an older continuation point in place. The store does not
+know how to create a new value for an id it hasn't seen before — use
+`AgentState.get_or_create_session(...)` for that, since only the state object
+has both the store and the resolved target. Workflow
 checkpointing should use the existing `CheckpointStorage` abstraction directly;
 if an app needs per-session resume, keep a small app-owned cursor such as
 `session_id -> checkpoint_id`.
@@ -53,6 +55,14 @@ result = await (await state.get_target()).run("Hello", session=session)
 await state.set_session(response_id, session)
 ```
 
+This response-keyed pattern supports simultaneous branches: callers may read
+the same `previous_response_id`, receive independent working copies, and store
+each result under a different new response id. A stable `conversation_id` is
+different: explicitly write the completed session back under that same id to
+advance its mutable head, and ensure only one caller advances it at a time.
+`AgentState` does not provide that application-level locking or optimistic
+concurrency control.
+
 Targets can be direct instances, synchronous factories, asynchronous factories,
 or awaitables:
 
@@ -73,6 +83,10 @@ result = await (await state.get_target()).run("Hello", checkpoint_storage=storag
 latest = await storage.get_latest(workflow_name=(await state.get_target()).name)
 ```
 
+A `Workflow` instance allows only one active run. Hosts that need simultaneous
+workflow runs should pass a factory or builder and set `cache_target=False` so
+each request receives a fresh workflow instance.
+
 `WorkflowState` also accepts an unbuilt workflow builder directly:
 
 ```python
diff --git a/python/packages/hosting/agent_framework_hosting/_state.py b/python/packages/hosting/agent_framework_hosting/_state.py
index ee0c6195a49..148221e6b44 100644
--- a/python/packages/hosting/agent_framework_hosting/_state.py
+++ b/python/packages/hosting/agent_framework_hosting/_state.py
@@ -20,6 +20,7 @@
 from __future__ import annotations
 
 import asyncio
+import copy
 import inspect
 from collections.abc import Awaitable, Callable, Mapping
 from typing import Any, Generic, Protocol, TypedDict, TypeVar, cast, runtime_checkable
@@ -50,6 +51,15 @@ class SessionStore:
     database, ...), you are responsible for that store's own TTL/eviction
     policy; this in-memory reference implementation does not model that
     concern.
+
+    The ``get`` method creates a copy of the session in order to ensure multiple
+    callers using the same response id can continue the session.
+    The behavior for that is controlled by the developer.
+
+    So if there should be branching, then make sure to store the session with the new
+    session id, if the conversation should continue, then store them with the same ID.
+    Ensuring that multiple callers cannot simultaneously overwrite the same session is
+    the responsibility of the developer.
     """
 
     def __init__(self) -> None:
@@ -57,7 +67,9 @@ def __init__(self) -> None:
         self._sessions: dict[str, AgentSession] = {}
 
     async def get(self, session_id: str) -> AgentSession | None:
-        """Return the stored session for ``session_id``, or ``None`` if absent.
+        """Return a copy of the stored session for ``session_id``, or ``None`` if absent.
+
+        When overriding this method ensure the semantics stay the same and a copy is returned.
 
         Args:
             session_id: Opaque app-selected session id.
@@ -67,7 +79,8 @@ async def get(self, session_id: str) -> AgentSession | None:
         """
         if not session_id:
             raise ValueError("session_id must be a non-empty string")
-        return self._sessions.get(session_id)
+        session = self._sessions.get(session_id)
+        return copy.deepcopy(session) if session is not None else None
 
     async def set(self, session_id: str, session: AgentSession) -> None:
         """Store ``session`` under ``session_id``, replacing any existing entry.
@@ -226,7 +239,8 @@ async def get_or_create_session(self, session_id: str) -> AgentSession:
             session_id: Opaque app-selected session id.
 
         Returns:
-            The stored or newly created ``AgentSession``.
+            An independent working copy of the stored or newly created
+            ``AgentSession``.
         """
         if not session_id:
             raise ValueError("session_id must be a non-empty string")
@@ -235,8 +249,9 @@ async def get_or_create_session(self, session_id: str) -> AgentSession:
             session = await self._session_store.get(session_id)
             if session is None:
                 target = await self.get_target()
-                session = target.create_session(session_id=session_id)
-                await self._session_store.set(session_id, session)
+                stored_session = target.create_session(session_id=session_id)
+                await self._session_store.set(session_id, stored_session)
+                session = copy.deepcopy(stored_session)
             return session
 
     async def set_session(self, session_id: str, session: AgentSession) -> None:
diff --git a/python/packages/hosting/tests/hosting/test_state.py b/python/packages/hosting/tests/hosting/test_state.py
index b0c8a0c68f4..deb166fc966 100644
--- a/python/packages/hosting/tests/hosting/test_state.py
+++ b/python/packages/hosting/tests/hosting/test_state.py
@@ -106,13 +106,23 @@ async def test_get_returns_none_for_missing_id(self) -> None:
 
         assert await store.get("session-1") is None
 
-    async def test_set_then_get_returns_stored_session(self) -> None:
+    async def test_set_then_get_returns_session_copy(self) -> None:
         store = SessionStore()
         session = AgentSession(session_id="session-1")
+        session.state["nested"] = {"values": ["original"]}
 
         await store.set("session-1", session)
 
-        assert await store.get("session-1") is session
+        stored = await store.get("session-1")
+        assert stored is not None
+        assert stored is not session
+        assert stored.session_id == session.session_id
+        assert stored.state == session.state
+
+        stored.state["nested"]["values"].append("changed")
+        reread = await store.get("session-1")
+        assert reread is not None
+        assert reread.state["nested"]["values"] == ["original"]
 
     async def test_set_can_store_same_session_under_additional_id(self) -> None:
         store = SessionStore()
@@ -121,8 +131,13 @@ async def test_set_can_store_same_session_under_additional_id(self) -> None:
         await store.set("resp_1", session)
         await store.set("resp_2", session)
 
-        assert await store.get("resp_1") is session
-        assert await store.get("resp_2") is session
+        first = await store.get("resp_1")
+        second = await store.get("resp_2")
+        assert first is not None
+        assert second is not None
+        assert first is not session
+        assert second is not session
+        assert first is not second
 
     async def test_set_replaces_existing_entry(self) -> None:
         store = SessionStore()
@@ -132,7 +147,10 @@ async def test_set_replaces_existing_entry(self) -> None:
         await store.set("session-1", first)
         await store.set("session-1", second)
 
-        assert await store.get("session-1") is second
+        stored = await store.get("session-1")
+        assert stored is not None
+        assert stored is not second
+        assert stored.session_id == second.session_id
 
     async def test_delete_forgets_session(self) -> None:
         store = SessionStore()
@@ -247,8 +265,9 @@ async def test_get_or_create_session_creates_and_stores_once(self) -> None:
         first = await state.get_or_create_session("session-1")
         second = await state.get_or_create_session("session-1")
 
-        assert first is second
+        assert first is not second
         assert first.session_id == "session-1"
+        assert second.session_id == "session-1"
         assert len(agent.created_sessions) == 1
 
     async def test_get_or_create_session_creates_once_for_concurrent_callers(self) -> None:
@@ -266,7 +285,8 @@ async def set(self, session_id: str, session: AgentSession) -> None:
 
         sessions = await asyncio.gather(*(state.get_or_create_session("session-1") for _ in range(20)))
 
-        assert all(session is sessions[0] for session in sessions)
+        assert len({id(session) for session in sessions}) == len(sessions)
+        assert all(session.session_id == "session-1" for session in sessions)
         assert len(agent.created_sessions) == 1
 
     async def test_get_or_create_session_reuses_a_session_set_on_the_state(self) -> None:
@@ -277,7 +297,8 @@ async def test_get_or_create_session_reuses_a_session_set_on_the_state(self) ->
 
         session = await state.get_or_create_session("session-1")
 
-        assert session is pre_existing
+        assert session is not pre_existing
+        assert session.session_id == pre_existing.session_id
         assert len(agent.created_sessions) == 0
 
 
@@ -332,6 +353,15 @@ async def test_workflow_builder_is_built_once_and_cached_by_default(self) -> Non
 
         assert first is second
 
+    async def test_workflow_builder_returns_fresh_targets_when_cache_disabled(self) -> None:
+        builder = _workflow_fixture("echo_workflow_builder")()
+        state: WorkflowState[Workflow] = WorkflowState(builder, cache_target=False)
+
+        first = await state.get_target()
+        second = await state.get_target()
+
+        assert first is not second
+
     async def test_accepts_orchestration_style_builder_without_importing_orchestrations(self) -> None:
         """``SupportsBuild`` is structural: any object with a zero-arg ``build() -> Workflow``
         is accepted, matching ``agent_framework_orchestrations``' builders without this
diff --git a/python/samples/04-hosting/af-hosting/local_responses/README.md b/python/samples/04-hosting/af-hosting/local_responses/README.md
index ab0c67fa286..9b976be1c4b 100644
--- a/python/samples/04-hosting/af-hosting/local_responses/README.md
+++ b/python/samples/04-hosting/af-hosting/local_responses/README.md
@@ -24,9 +24,9 @@ What the route demonstrates:
   turn.
 - Produces the AF messages, options, and session id that the route passes to
   `agent.run(...)`.
-- **Stores** each newly minted response id for the session it was just
-  resolved from, via `state.set_session(response_id, session)` after
-  `agent.run(...)` has updated the session.
+- **Stores** each newly minted response id for response-keyed continuation,
+  via `state.set_session(response_id, session)` after `agent.run(...)` has
+  updated the session.
   OpenAI's `previous_response_id` rotates every turn *by design* — it lets a
   caller continue from any earlier response, not just the latest one — so
   every response id needs to stay independently resolvable, not just the
@@ -34,6 +34,14 @@ What the route demonstrates:
 - Treats an unknown `conversation_id` as a request to create a new local
   session. Your app can choose a stricter policy, such as requiring a separate
   API to create new conversations before callers can continue them.
+- Explicitly advances a supplied `conversation_id` after each completed run.
+  A conversation id is a mutable head, so only one caller should advance it at
+  a time. The sample and `AgentState` do not provide that locking; production
+  apps must serialize writers or use optimistic concurrency. These requests
+  store the updated session only under the stable conversation id.
+- Treats each `previous_response_id` as an immutable snapshot. Multiple callers
+  can branch from the same response concurrently because each receives a
+  session copy and stores its result under a newly minted response id.
 
 `app:app` is a module-level FastAPI ASGI app; recommended local launch is
 Hypercorn.
@@ -47,7 +55,9 @@ the FastAPI app layer, or inside the route body.
 Session continuation deserves particular care: treat `previous_response_id` and
 `conversation_id` as untrusted request values, authorize the caller before
 loading or storing a session for those ids, and partition any durable session
-store by tenant/user as appropriate for your application.
+store by tenant/user as appropriate for your application. Also coordinate
+writers for each stable `conversation_id`; this sample does not do so out of
+the box.
 
 ## Run
 
diff --git a/python/samples/04-hosting/af-hosting/local_responses/app.py b/python/samples/04-hosting/af-hosting/local_responses/app.py
index da3ae51bda5..88c22315c71 100644
--- a/python/samples/04-hosting/af-hosting/local_responses/app.py
+++ b/python/samples/04-hosting/af-hosting/local_responses/app.py
@@ -118,6 +118,7 @@ async def responses(body: dict[str, Any] = Body(...)) -> JSONResponse | Streamin
     except ValueError as exc:
         raise HTTPException(status_code=400, detail=str(exc)) from exc
     session_id = responses_session_id(body)
+    conversation_id = session_id if body.get("conversation_id") == session_id else None
     response_id = create_response_id()
 
     # App-specific policy: allow only the request options this route is willing
@@ -151,10 +152,15 @@ async def stream_events() -> AsyncIterator[str]:
             ):
                 yield event
             # `agent.run(..., stream=True)` updates the session while the stream
-            # is consumed/finalized. Store it under the newly minted response id
-            # after finalization so a later `previous_response_id` can restore
-            # this exact continuation point.
-            await state.set_session(response_id, session)
+            # is consumed/finalized. Persist the selected continuation only
+            # after finalization.
+            if conversation_id is not None:
+                # A stable conversation id is a mutable head. Apps must ensure
+                # only one caller advances it at a time; AgentState does not
+                # serialize concurrent runs for the same id.
+                await state.set_session(conversation_id, session)
+            else:
+                await state.set_session(response_id, session)
 
         return StreamingResponse(
             stream_events(),
@@ -166,10 +172,14 @@ async def stream_events() -> AsyncIterator[str]:
         session=session,
         options=options_for_run,
     )
-    # `agent.run(...)` updates the session. Store it under the newly minted
-    # response id after the run so `previous_response_id=response_id` continues
-    # from this exact point.
-    await state.set_session(response_id, session)
+    # `agent.run(...)` updates the session. Persist the selected continuation
+    # only after the run completes.
+    if conversation_id is not None:
+        # Preserve sequential conversation continuity. Production apps must
+        # provide their own per-conversation single-writer coordination.
+        await state.set_session(conversation_id, session)
+    else:
+        await state.set_session(response_id, session)
     return JSONResponse(
         responses_from_run(
             result,
diff --git a/python/samples/04-hosting/af-hosting/local_responses_workflow/README.md b/python/samples/04-hosting/af-hosting/local_responses_workflow/README.md
index f5961cc36b0..d394ec92d6c 100644
--- a/python/samples/04-hosting/af-hosting/local_responses_workflow/README.md
+++ b/python/samples/04-hosting/af-hosting/local_responses_workflow/README.md
@@ -19,6 +19,10 @@ stored under the sample's local `storage/` root. Checkpoints are scoped into
 per-continuation buckets so a "latest checkpoint" lookup cannot cross
 conversations.
 
+The sample passes a builder with `cache_target=False`, producing a fresh
+workflow instance for every request. This is required for simultaneous runs
+because one `Workflow` instance intentionally permits only one active run.
+
 ## Production readiness
 
 This is not a full-fledged production deployment. Before exposing this pattern