diff --git a/python/packages/azure-cosmos/agent_framework_azure_cosmos/_history_provider.py b/python/packages/azure-cosmos/agent_framework_azure_cosmos/_history_provider.py
index b62c5467ad8..0a6e2167bf6 100644
--- a/python/packages/azure-cosmos/agent_framework_azure_cosmos/_history_provider.py
+++ b/python/packages/azure-cosmos/agent_framework_azure_cosmos/_history_provider.py
@@ -300,12 +300,6 @@ async def _ensure_container_proxy(self) -> None:
 
     @staticmethod
     def _session_partition_key(session_id: str | None) -> str:
-        if session_id:
-            return session_id
-
-        generated_session_id = str(uuid.uuid4())
-        logger.warning(
-            "Received empty session_id; generated temporary session id '%s' for Cosmos partition key.",
-            generated_session_id,
-        )
-        return generated_session_id
+        if not session_id:
+            raise ValueError("session_id must be a non-empty string.")
+        return session_id
diff --git a/python/packages/azure-cosmos/tests/test_cosmos_history_provider.py b/python/packages/azure-cosmos/tests/test_cosmos_history_provider.py
index 58461cd094a..3451f7b304f 100644
--- a/python/packages/azure-cosmos/tests/test_cosmos_history_provider.py
+++ b/python/packages/azure-cosmos/tests/test_cosmos_history_provider.py
@@ -189,26 +189,16 @@ async def test_empty_returns_empty(self, mock_container: MagicMock) -> None:
 
         assert messages == []
 
-    async def test_none_session_id_generates_guid_partition_key(
-        self, mock_container: MagicMock, caplog: pytest.LogCaptureFixture
+    @pytest.mark.parametrize("session_id", [None, ""], ids=["none", "empty"])
+    async def test_missing_session_id_raises_value_error(
+        self, mock_container: MagicMock, session_id: str | None
     ) -> None:
-        mock_container.query_items.return_value = _to_async_iter([])
-
         provider = CosmosHistoryProvider(source_id="mem", container_client=mock_container)
-        with caplog.at_level("WARNING"):
-            await provider.get_messages(None)
 
-        query_kwargs = mock_container.query_items.call_args.kwargs
-        session_key = query_kwargs["partition_key"]
-        assert isinstance(session_key, str)
-        assert session_key != ""
-        assert session_key != "default"
-        uuid.UUID(session_key)
-        assert query_kwargs["parameters"] == [
-            {"name": "@session_id", "value": session_key},
-            {"name": "@source_id", "value": "mem"},
-        ]
-        assert "Received empty session_id" in caplog.text
+        with pytest.raises(ValueError, match="session_id"):
+            await provider.get_messages(session_id)
+
+        mock_container.query_items.assert_not_called()
 
     async def test_skips_non_dict_message_payload(self, mock_container: MagicMock) -> None:
         mock_container.query_items.return_value = _to_async_iter([{"message": "bad"}, {"message": None}])