diff --git a/httpclient/httpclient.py b/httpclient/httpclient.py
index 746cdf2..6938273 100644
--- a/httpclient/httpclient.py
+++ b/httpclient/httpclient.py
@@ -1,5 +1,5 @@
 # /// zerodep
-# version = "0.4.2"
+# version = "0.4.4"
 # deps = []
 # tier = "subsystem"
 # category = "network"
@@ -69,6 +69,8 @@
     "HttpConnectionError",
     "HttpTimeoutError",
     "Socks5Error",
+    # Data structures
+    "CaseInsensitiveDict",
     # Response classes
     "Response",
     "StreamingResponse",
@@ -108,6 +110,145 @@
 DEFAULT_POOL_IDLE_TIMEOUT = 60.0
 
 
+# ── CaseInsensitiveDict ──
+
+
+class CaseInsensitiveDict(dict):
+    """Case-insensitive key lookup ``dict`` subclass that preserves original casing.
+
+    Provides case-insensitive HTTP header storage: ``d["Content-Type"]``
+    and ``d["content-type"]`` resolve to the same slot, but iteration and
+    wire serialisation yield the original casing the caller supplied.
+
+    Internally the underlying ``dict`` stores ``{lowercase_key: value}``
+    for O(1) lookups, while a parallel ``_keys`` mapping records
+    ``{lowercase_key: original_key}`` for casing-preserving iteration.
+
+    This is the type used for ``Response.headers``,
+    ``StreamingResponse.headers``, and the internal ``req_headers`` dict
+    that flows through ``_prepare_request``.  HTTP header names are
+    case-insensitive per :rfc:`7230` \u00a73.2, but the wire format and
+    echo tests expect the casing the caller supplied.
+
+    It is a ``dict`` subclass, so it is accepted everywhere a ``dict``
+    is expected.  Equality is case-insensitive on keys:
+    ``CaseInsensitiveDict({"X-Foo": "bar"}) == {"x-foo": "bar"}``.
+    """
+
+    # Parallel store: lowercase_key → original_key supplied by the caller.
+    # Kept in sync with the underlying dict at all times.
+    _keys: dict[str, str]
+
+    def __init__(
+        self,
+        data: dict | list[tuple[str, str]] | None = None,
+        **kwargs: str,
+    ) -> None:
+        super().__init__()
+        self._keys = {}
+        if data is not None:
+            self.update(data)
+        if kwargs:
+            self.update(kwargs)
+
+    # ------------------------------------------------------------------ #
+    # Core write operations — all funnel through __setitem__ / __delitem__
+    # ------------------------------------------------------------------ #
+
+    def __setitem__(self, key: str, value: str) -> None:  # type: ignore[override]
+        lower = key.lower()
+        self._keys[lower] = key  # preserve (or update) original casing
+        super().__setitem__(lower, value)
+
+    def __delitem__(self, key: str) -> None:
+        lower = key.lower()
+        self._keys.pop(lower, None)
+        super().__delitem__(lower)
+
+    # ------------------------------------------------------------------ #
+    # Read operations — normalise lookup key to lowercase
+    # ------------------------------------------------------------------ #
+
+    def __getitem__(self, key: str) -> str:  # type: ignore[override]
+        return super().__getitem__(key.lower())  # type: ignore[return-value]
+
+    def __contains__(self, key: object) -> bool:
+        return super().__contains__(key.lower() if isinstance(key, str) else key)
+
+    def get(self, key: str, default: str | None = None) -> str | None:  # type: ignore[override]
+        return super().get(key.lower(), default)  # type: ignore[return-value]
+
+    def pop(self, key: str, *args: str) -> str:  # type: ignore[override]
+        lower = key.lower()
+        self._keys.pop(lower, None)
+        return super().pop(lower, *args)  # type: ignore[return-value]
+
+    def setdefault(self, key: str, default: str = "") -> str:  # type: ignore[override]
+        if key.lower() not in self:
+            self[key] = default
+        return self[key]
+
+    def update(  # type: ignore[override]
+        self,
+        data: dict | list[tuple[str, str]] | None = None,
+        **kwargs: str,
+    ) -> None:
+        if data is not None:
+            items = data.items() if hasattr(data, "items") else data
+            for k, v in items:
+                self[k] = v
+        for k, v in kwargs.items():
+            self[k] = v
+
+    # ------------------------------------------------------------------ #
+    # Iteration — yield original casing so wire format is preserved
+    # ------------------------------------------------------------------ #
+
+    def __iter__(self):  # type: ignore[override]
+        for lower in super().__iter__():
+            yield self._keys.get(lower, lower)
+
+    def keys(self):  # type: ignore[override]
+        return list(self.__iter__())
+
+    def values(self):  # type: ignore[override]
+        return list(super().values())
+
+    def items(self):  # type: ignore[override]
+        """Yield ``(original_key, value)`` pairs; preserves casing on the wire."""
+        for lower, value in super().items():
+            yield self._keys.get(lower, lower), value
+
+    # ------------------------------------------------------------------ #
+    # Equality — case-insensitive on keys
+    # ------------------------------------------------------------------ #
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, dict):
+            return NotImplemented
+        if len(self) != len(other):
+            return False
+        # Use our case-insensitive .get() so "X-Foo" == "x-foo" for key lookups
+        return all(self.get(k) == v for k, v in other.items())
+
+    def __hash__(self) -> None:  # type: ignore[override]
+        return None  # dicts are unhashable; satisfy type checkers
+
+    # ------------------------------------------------------------------ #
+    # Miscellaneous
+    # ------------------------------------------------------------------ #
+
+    def copy(self) -> "CaseInsensitiveDict":
+        return CaseInsensitiveDict(self.items())
+
+    def __reduce__(self):  # type: ignore[override]
+        """Ensure pickle/copy reconstructs via __init__ to restore _keys."""
+        return (type(self), (list(self.items()),))
+
+    def __repr__(self) -> str:
+        return f"{type(self).__name__}({dict(self.items())!r})"
+
+
 # ── Exceptions ──
 
 
@@ -193,7 +334,7 @@ class Response:
     def __init__(
         self,
         status_code: int,
-        headers: dict[str, str],
+        headers: CaseInsensitiveDict,
         content: bytes,
         url: str,
     ) -> None:
@@ -255,7 +396,7 @@ def __repr__(self) -> str:
         return f"<Response [{self.status_code}]>"
 
 
-def _guess_encoding_from_headers(headers: dict[str, str]) -> str:
+def _guess_encoding_from_headers(headers: CaseInsensitiveDict) -> str:
     """Extract charset from Content-Type header, default utf-8."""
     ct = headers.get("content-type", "")
     for part in ct.split(";"):
@@ -474,7 +615,7 @@ class StreamingResponse:
     )
 
     status_code: int
-    headers: dict[str, str]
+    headers: CaseInsensitiveDict
     url: str
     _encoding: str
     _decompressor: zlib._Decompress | None
@@ -495,7 +636,7 @@ def __init__(self) -> None:
     def _from_sync(
         cls,
         status_code: int,
-        headers: dict[str, str],
+        headers: CaseInsensitiveDict,
         url: str,
         resp: http.client.HTTPResponse,
         conn: http.client.HTTPConnection,
@@ -524,7 +665,7 @@ def _from_sync(
     def _from_async(
         cls,
         status_code: int,
-        headers: dict[str, str],
+        headers: CaseInsensitiveDict,
         url: str,
         reader: asyncio.StreamReader,
         writer: asyncio.StreamWriter,
@@ -1270,6 +1411,30 @@ def _parse_url(url: str) -> tuple[str, str, int, str, bool]:
 # -- Shared request preparation helpers --
 
 
+def _headers_set_default(
+    req_headers: CaseInsensitiveDict, key: str, value: str
+) -> None:
+    """Set *key*/*value* only when the key is not already present.
+
+    With ``CaseInsensitiveDict``, ``setdefault`` already handles case
+    normalisation.  This thin wrapper keeps the call-site readable.
+    """
+    req_headers.setdefault(key, value)
+
+
+def _headers_merge_user(
+    req_headers: CaseInsensitiveDict, user_headers: dict[str, str] | None
+) -> None:
+    """Merge *user_headers* into *req_headers*; user values always win.
+
+    With ``CaseInsensitiveDict``, ``update`` handles case-insensitive
+    collision automatically — setting ``user-agent`` overwrites
+    ``User-Agent`` in the same slot.
+    """
+    if user_headers:
+        req_headers.update(user_headers)
+
+
 def _prepare_request(
     method: str,
     url: str,
@@ -1279,30 +1444,45 @@ def _prepare_request(
     files: dict[str, Any] | list[tuple[str, Any]] | None,
     params: dict[str, Any] | None,
     auth: tuple[str, str] | Auth | None,
-) -> tuple[str, bytes | None, dict[str, str], Auth | None]:
+) -> tuple[str, bytes | None, CaseInsensitiveDict, Auth | None]:
     """Build URL, encode body, assemble headers, and normalize auth.
 
     Shared by _sync_request and _async_request (Phases 1-3).
 
+    Header precedence (highest → lowest):
+      1. auth headers (digest/basic — must override everything)
+      2. user-supplied *headers*
+      3. body-derived defaults (Content-Type, Content-Length)
+      4. library defaults (User-Agent, Accept-Encoding)
+
+    All keys are normalised to lowercase via :class:`CaseInsensitiveDict`;
+    no duplicate header names are ever emitted.
+
     Returns:
         (final_url, body_bytes, request_headers, auth_object).
     """
     url = _build_url(url, params)
     body, content_type = _prepare_body(data, json_data, files)
 
-    req_headers: dict[str, str] = {
-        "User-Agent": DEFAULT_USER_AGENT,
-        "Accept-Encoding": "gzip, deflate",
-    }
+    req_headers: CaseInsensitiveDict = CaseInsensitiveDict()
+
+    # Library defaults — only applied when the user hasn't already set them
+    _headers_set_default(req_headers, "User-Agent", DEFAULT_USER_AGENT)
+    _headers_set_default(req_headers, "Accept-Encoding", "gzip, deflate")
+
+    # Body-derived headers — set as defaults so user can override
     if content_type:
-        req_headers["Content-Type"] = content_type
+        _headers_set_default(req_headers, "Content-Type", content_type)
     if body is not None:
-        req_headers["Content-Length"] = str(len(body))
-    req_headers.update(headers or {})
+        _headers_set_default(req_headers, "Content-Length", str(len(body)))
+
+    # User headers win over all of the above
+    _headers_merge_user(req_headers, headers)
 
     auth_obj = _normalize_auth(auth)
     if isinstance(auth_obj, BasicAuth):
-        req_headers.update(auth_obj.auth_headers(method, url))
+        # Auth headers have highest priority — always override
+        _headers_merge_user(req_headers, auth_obj.auth_headers(method, url))
 
     return url, body, req_headers, auth_obj
 
@@ -1651,7 +1831,7 @@ def _sync_request(
             try:
                 conn.request(method, request_path, body=body, headers=req_headers)
                 resp = conn.getresponse()
-                resp_headers = {k.lower(): v for k, v in resp.getheaders()}
+                resp_headers = CaseInsensitiveDict(resp.getheaders())
                 status = resp.status
 
                 if _is_redirect(status, resp_headers):
@@ -1718,7 +1898,7 @@ def _sync_request(
 async def _async_read_response_headers(
     reader: asyncio.StreamReader,
     timeout: float,
-) -> tuple[int, dict[str, str]]:
+) -> tuple[int, CaseInsensitiveDict]:
     """Read HTTP status line and headers from an asyncio StreamReader.
 
     Does NOT consume the body -- the reader is left positioned at the
@@ -1736,7 +1916,7 @@ async def _async_read_response_headers(
     status_code = int(parts[1])
 
     # Headers until empty line
-    headers: dict[str, str] = {}
+    headers: CaseInsensitiveDict = CaseInsensitiveDict()
     while True:
         line = await asyncio.wait_for(reader.readline(), timeout=timeout)
         decoded = line.decode("latin-1").rstrip("\r\n")
@@ -1744,7 +1924,7 @@ async def _async_read_response_headers(
             break
         if ":" in decoded:
             k, v = decoded.split(":", 1)
-            headers[k.strip().lower()] = v.strip()
+            headers[k.strip()] = v.strip()
 
     return status_code, headers
 
@@ -2001,10 +2181,12 @@ def _build_raw_http_request(
         Encoded HTTP/1.1 request bytes (without body).
     """
     request_line = f"{method} {request_path} HTTP/1.1\r\n"
-    header_lines = f"Host: {host}\r\n"
+    # Emit Host first (RFC 7230 §5.4). req_headers is a CaseInsensitiveDict
+    # so the 'in' check is O(1) and case-insensitive.
+    header_lines = "" if "host" in req_headers else f"Host: {host}\r\n"
     for k, v in req_headers.items():
         header_lines += f"{k}: {v}\r\n"
-    if not use_pool or use_proxy:
+    if (not use_pool or use_proxy) and "connection" not in req_headers:
         header_lines += "Connection: close\r\n"
     header_lines += "\r\n"
     return (request_line + header_lines).encode("latin-1")
@@ -2346,13 +2528,13 @@ def _encode_multipart(
 def _merge_headers(
     base: dict[str, str] | None,
     extra: dict[str, str] | None,
-) -> dict[str, str]:
-    """Merge header dicts (case-insensitive merge, last wins)."""
-    merged: dict[str, str] = {}
-    for h in (base, extra):
-        if h:
-            for k, v in h.items():
-                merged[k] = v
+) -> CaseInsensitiveDict:
+    """Merge header dicts into a :class:`CaseInsensitiveDict`; *extra* wins."""
+    merged = CaseInsensitiveDict()
+    if base:
+        merged.update(base)
+    if extra:
+        merged.update(extra)
     return merged
 
 
@@ -2509,6 +2691,13 @@ def close(self) -> None:
         """Close all pooled connections."""
         self._pool.close_all()
 
+    # Async-style alias so callers can use the same name for both clients
+    # in generic code (``await client.aclose()`` works for AsyncClient;
+    # ``client.aclose()`` works here as a plain synchronous no-op wrapper).
+    async def aclose(self) -> None:  # type: ignore[misc]
+        """Async-compatible alias for :meth:`close` (parity with AsyncClient)."""
+        self.close()
+
     def __enter__(self) -> Client:
         return self
 
@@ -2588,6 +2777,21 @@ async def aclose(self) -> None:
         """Close all pooled connections."""
         await self._pool.close_all()
 
+    # Sync-style alias for interface parity with Client.
+    def close(self) -> None:
+        """Emit a warning and do nothing — use ``await aclose()`` instead.
+
+        ``AsyncClient`` manages async resources; calling synchronous ``close()``
+        cannot safely await the pool teardown coroutine.  This method exists
+        solely for interface parity with :class:`Client` so that type-annotated
+        code that calls ``client.close()`` does not raise ``AttributeError``.
+        Always prefer :meth:`aclose` inside async code.
+        """
+        logger.warning(
+            "AsyncClient.close() is a no-op — use 'await client.aclose()' "
+            "to properly close async connections."
+        )
+
     async def __aenter__(self) -> AsyncClient:
         return self
 
diff --git a/httpclient/test_httpclient_correctness.py b/httpclient/test_httpclient_correctness.py
index 83f20a7..8343aaf 100644
--- a/httpclient/test_httpclient_correctness.py
+++ b/httpclient/test_httpclient_correctness.py
@@ -1,9 +1,11 @@
 """Correctness tests: zerodep HTTP client vs httpx."""
 
 import asyncio
+import io
 import os
 import sys
 import time
+import unittest.mock
 from concurrent.futures import ThreadPoolExecutor
 
 import pytest
@@ -13,6 +15,7 @@
 from httpclient import (
     AsyncClient,
     BasicAuth,
+    CaseInsensitiveDict,
     Client,
     DigestAuth,
     HttpConnectionError,
@@ -261,7 +264,6 @@ def test_upload_with_data(self, httpbin_url):
         assert "file" in body["files"]
 
     def test_upload_file_object(self, httpbin_url):
-        import io
 
         buf = io.BytesIO(b"file object content")
         buf.name = "buffer.txt"
@@ -836,3 +838,353 @@ def do_request(i):
 
         client.close()
         assert not errors, f"Thread errors: {errors}"
+
+
+# ── Unit tests: header deduplication (no network) ──
+# These tests verify the fix for issue #116:
+# - _prepare_request: case-insensitive merge; user headers win over defaults
+# - _build_raw_http_request: no duplicate Host when user provides one
+
+
+sys.path.insert(0, os.path.dirname(__file__))
+
+from httpclient import (
+    _build_raw_http_request,
+    _merge_headers,
+    _prepare_request,
+)
+
+
+class TestCaseInsensitiveDict:
+    """CaseInsensitiveDict: O(1) case-insensitive key operations."""
+
+    def test_set_and_get_same_case(self):
+        d = CaseInsensitiveDict()
+        d["Content-Type"] = "application/json"
+        assert d["Content-Type"] == "application/json"
+
+    def test_get_different_case(self):
+        d = CaseInsensitiveDict()
+        d["Content-Type"] = "application/json"
+        assert d["content-type"] == "application/json"
+        assert d["CONTENT-TYPE"] == "application/json"
+
+    def test_no_duplicate_keys_on_overwrite(self):
+        d = CaseInsensitiveDict()
+        d["User-Agent"] = "a"
+        d["user-agent"] = "b"
+        assert len(d) == 1
+        assert d["user-agent"] == "b"
+
+    def test_contains_case_insensitive(self):
+        d = CaseInsensitiveDict({"Host": "example.com"})
+        assert "host" in d
+        assert "HOST" in d
+        assert "Host" in d
+
+    def test_delitem_case_insensitive(self):
+        d = CaseInsensitiveDict({"Content-Type": "text/plain"})
+        del d["content-type"]
+        assert "content-type" not in d
+
+    def test_get_with_default(self):
+        d = CaseInsensitiveDict()
+        assert d.get("Missing") is None
+        assert d.get("Missing", "fallback") == "fallback"
+
+    def test_setdefault_does_not_overwrite(self):
+        d = CaseInsensitiveDict({"X-Foo": "original"})
+        d.setdefault("x-foo", "new")
+        assert d["x-foo"] == "original"
+
+    def test_setdefault_inserts_when_absent(self):
+        d = CaseInsensitiveDict()
+        d.setdefault("X-Bar", "default")
+        assert d["x-bar"] == "default"
+
+    def test_pop_case_insensitive(self):
+        d = CaseInsensitiveDict({"Authorization": "Bearer token"})
+        val = d.pop("authorization")
+        assert val == "Bearer token"
+        assert len(d) == 0
+
+    def test_update_from_dict(self):
+        d = CaseInsensitiveDict({"User-Agent": "old"})
+        d.update({"user-agent": "new", "Accept": "*/*"})
+        assert len(d) == 2
+        assert d["user-agent"] == "new"
+        assert d["accept"] == "*/*"
+
+    def test_init_from_list_of_tuples(self):
+        d = CaseInsensitiveDict([("Content-Type", "text/html"), ("X-Foo", "bar")])
+        assert d["content-type"] == "text/html"
+        assert d["x-foo"] == "bar"
+
+    def test_iteration_preserves_original_casing(self):
+        """keys()/items() must yield the casing the caller supplied, not lowercase."""
+        d = CaseInsensitiveDict({"Content-Type": "text/plain", "HOST": "example.com"})
+        # Original casing is preserved in iteration (important for wire format)
+        assert set(d.keys()) == {"Content-Type", "HOST"}
+        # But lookups are still case-insensitive
+        assert d["content-type"] == "text/plain"
+        assert d["host"] == "example.com"
+
+    def test_is_dict_subclass(self):
+        d = CaseInsensitiveDict()
+        assert isinstance(d, dict)
+
+    def test_wire_format_preserves_casing(self):
+        """items() must yield original casing — critical for http.client wire format."""
+        d = CaseInsensitiveDict(
+            {"Content-Type": "application/json", "X-Custom": "test"}
+        )
+        wire_keys = [k for k, _ in d.items()]
+        assert "Content-Type" in wire_keys
+        assert "X-Custom" in wire_keys
+        # Lowercase lookup still works
+        assert d["content-type"] == "application/json"
+
+    def test_sigv4_headers_no_duplicate_host(self):
+        """Key use case: SigV4 headers merged with library defaults."""
+        d = CaseInsensitiveDict()
+        d.setdefault("User-Agent", "zerodep/1")
+        d.setdefault("Accept-Encoding", "gzip")
+        d.update(
+            {
+                "host": "mybucket.s3.amazonaws.com",
+                "x-amz-date": "20240101T120000Z",
+                "Authorization": "AWS4-HMAC-SHA256 ...",
+            }
+        )
+        # No duplicate host key
+        host_keys = [k for k in d if k.lower() == "host"]
+        assert len(host_keys) == 1
+        # Original casing preserved in items()
+        item_keys = [k for k, _ in d.items()]
+        assert "User-Agent" in item_keys  # setdefault preserves casing
+
+
+class TestPrepareRequestHeaderDedup:
+    """_prepare_request: case-insensitive merge, user headers always win."""
+
+    def _prepare(self, headers=None, data=None):
+        _, _, req_headers, _ = _prepare_request(
+            "GET",
+            "https://example.com/",
+            headers,
+            data,
+            None,
+            None,
+            None,
+            None,
+        )
+        return req_headers
+
+    def test_default_user_agent_present(self):
+        h = self._prepare()
+        assert any(k.lower() == "user-agent" for k in h)
+
+    def test_user_agent_override_same_case(self):
+        h = self._prepare({"User-Agent": "my-agent/1.0"})
+        ua_values = [v for k, v in h.items() if k.lower() == "user-agent"]
+        assert ua_values == ["my-agent/1.0"], f"got: {h}"
+
+    def test_user_agent_override_lowercase(self):
+        """user-agent (lowercase) must override the default User-Agent."""
+        h = self._prepare({"user-agent": "custom/2.0"})
+        ua_values = [v for k, v in h.items() if k.lower() == "user-agent"]
+        assert len(ua_values) == 1, f"duplicate user-agent: {h}"
+        assert ua_values[0] == "custom/2.0"
+
+    def test_accept_encoding_override(self):
+        h = self._prepare({"Accept-Encoding": "identity"})
+        ae_values = [v for k, v in h.items() if k.lower() == "accept-encoding"]
+        assert ae_values == ["identity"]
+
+    def test_accept_encoding_override_lowercase(self):
+        h = self._prepare({"accept-encoding": "identity"})
+        ae_values = [v for k, v in h.items() if k.lower() == "accept-encoding"]
+        assert len(ae_values) == 1, f"duplicate accept-encoding: {h}"
+        assert ae_values[0] == "identity"
+
+    def test_no_duplicate_keys_with_mixed_case(self):
+        h = self._prepare({"user-agent": "x", "accept-encoding": "identity"})
+        lower_keys = [k.lower() for k in h]
+        assert len(lower_keys) == len(set(lower_keys)), f"duplicate keys: {lower_keys}"
+
+    def test_user_provided_content_type_wins(self):
+        """User content-type overrides the body-derived one."""
+        h = self._prepare(
+            headers={"Content-Type": "application/x-custom"},
+            data=b"body",
+        )
+        ct_values = [v for k, v in h.items() if k.lower() == "content-type"]
+        assert ct_values == ["application/x-custom"], f"got: {h}"
+
+    def test_no_duplicate_content_type(self):
+        h = self._prepare(
+            headers={"content-type": "text/plain"},
+            data=b"hello",
+        )
+        ct_values = [v for k, v in h.items() if k.lower() == "content-type"]
+        assert len(ct_values) == 1
+
+    def test_sigv4_style_headers_preserved(self):
+        """Simulate a SigV4-signed headers dict; no defaults should clobber them."""
+        sigv4_headers = {
+            "host": "mybucket.s3.amazonaws.com",
+            "x-amz-date": "20240101T120000Z",
+            "x-amz-content-sha256": "e3b0c44298fc1c149afbf4c8996fb924"
+            "27ae41e4649b934ca495991b7852b855",
+            "Authorization": "AWS4-HMAC-SHA256 Credential=AKIA.../...",
+        }
+        h = self._prepare(sigv4_headers)
+        # host must appear exactly once
+        host_values = [v for k, v in h.items() if k.lower() == "host"]
+        assert host_values == ["mybucket.s3.amazonaws.com"], f"got: {h}"
+        # SigV4 headers must survive intact
+        assert "x-amz-date" in h or "X-Amz-Date" in h
+
+
+class TestBuildRawHttpRequestHostDedup:
+    """_build_raw_http_request: no duplicate Host when user already provides one."""
+
+    def _build(self, req_headers, host="example.com", use_pool=False, use_proxy=False):
+        # _build_raw_http_request always receives a CaseInsensitiveDict in
+        # production (from _prepare_request); wrap here to match reality.
+        if not isinstance(req_headers, CaseInsensitiveDict):
+            req_headers = CaseInsensitiveDict(req_headers)
+        raw = _build_raw_http_request(
+            "GET", "/path", host, req_headers, use_pool, use_proxy
+        )
+        return raw.decode("latin-1")
+
+    def test_host_added_when_absent(self):
+        raw = self._build({})
+        host_count = sum(
+            1 for line in raw.splitlines() if line.lower().startswith("host:")
+        )
+        assert host_count == 1
+
+    def test_no_duplicate_host_when_user_provides_host(self):
+        """User-supplied 'host' must not result in two Host lines."""
+        raw = self._build({"host": "mybucket.s3.amazonaws.com"})
+        host_lines = [ln for ln in raw.splitlines() if ln.lower().startswith("host:")]
+        assert len(host_lines) == 1, f"duplicate Host lines: {host_lines}"
+
+    def test_no_duplicate_host_uppercase(self):
+        raw = self._build({"Host": "custom.example.com"})
+        host_lines = [ln for ln in raw.splitlines() if ln.lower().startswith("host:")]
+        assert len(host_lines) == 1, f"duplicate Host lines: {host_lines}"
+
+    def test_user_host_value_preserved(self):
+        raw = self._build({"host": "mybucket.s3.amazonaws.com"})
+        host_lines = [ln for ln in raw.splitlines() if ln.lower().startswith("host:")]
+        assert "mybucket.s3.amazonaws.com" in host_lines[0]
+
+    def test_connection_close_added_when_no_pool(self):
+        raw = self._build({}, use_pool=False)
+        assert "Connection: close" in raw
+
+    def test_no_duplicate_connection_when_user_sets_it(self):
+        raw = self._build({"Connection": "keep-alive"}, use_pool=False)
+        conn_lines = [
+            ln for ln in raw.splitlines() if ln.lower().startswith("connection:")
+        ]
+        assert len(conn_lines) == 1
+        assert "keep-alive" in conn_lines[0]
+
+    def test_sigv4_full_roundtrip(self):
+        """Simulate what AsyncS3Client would pass; verify no duplicates."""
+        sha256_empty = (
+            "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
+        )
+        auth_val = (
+            "AWS4-HMAC-SHA256 Credential=minioadmin/20240101/us-east-1/s3/aws4_request,"
+            " SignedHeaders=host;x-amz-content-sha256;x-amz-date, Signature=abc123"
+        )
+        sigv4_headers = {
+            "host": "127.0.0.1:9000",
+            "x-amz-date": "20240101T120000Z",
+            "x-amz-content-sha256": sha256_empty,
+            "Authorization": auth_val,
+        }
+        raw = self._build(sigv4_headers, host="127.0.0.1:9000")
+        host_lines = [ln for ln in raw.splitlines() if ln.lower().startswith("host:")]
+        assert len(host_lines) == 1, f"duplicate Host: {host_lines}"
+        assert "Authorization" in raw
+        assert "x-amz-date" in raw
+
+
+class TestMergeHeadersCaseInsensitive:
+    """_merge_headers: case-insensitive dedup, extra wins."""
+
+    def test_extra_overrides_base_same_case(self):
+        merged = _merge_headers({"User-Agent": "base/1"}, {"User-Agent": "extra/2"})
+        ua = [v for k, v in merged.items() if k.lower() == "user-agent"]
+        assert ua == ["extra/2"]
+
+    def test_extra_overrides_base_different_case(self):
+        merged = _merge_headers({"User-Agent": "base/1"}, {"user-agent": "extra/2"})
+        ua = [v for k, v in merged.items() if k.lower() == "user-agent"]
+        assert len(ua) == 1, f"duplicate: {merged}"
+        assert ua[0] == "extra/2"
+
+    def test_no_duplicate_keys(self):
+        merged = _merge_headers(
+            {"Content-Type": "application/json"},
+            {"content-type": "text/plain"},
+        )
+        keys_lower = [k.lower() for k in merged]
+        assert len(keys_lower) == len(set(keys_lower)), f"duplicates: {merged}"
+
+    def test_none_base(self):
+        merged = _merge_headers(None, {"X-Foo": "bar"})
+        assert merged == {"X-Foo": "bar"}  # __eq__ is case-insensitive on keys
+
+    def test_none_extra(self):
+        merged = _merge_headers({"X-Foo": "bar"}, None)
+        assert merged == {"X-Foo": "bar"}
+
+    def test_both_none(self):
+        assert _merge_headers(None, None) == {}
+
+
+class TestClientInterfaceParity:
+    """Client and AsyncClient expose the same interface."""
+
+    def test_client_has_aclose(self):
+        c = Client()
+        assert callable(getattr(c, "aclose", None)), "Client missing aclose"
+        c.close()
+
+    def test_async_client_close_is_noop_with_warning(self):
+        """AsyncClient.close() must exist, be callable, and log a warning."""
+        import logging
+
+        ac = AsyncClient()
+        assert callable(getattr(ac, "close", None)), "AsyncClient missing close"
+        with unittest.mock.patch.object(
+            logging.getLogger("httpclient"), "warning"
+        ) as mock_warn:
+            ac.close()
+            mock_warn.assert_called_once()
+            assert "aclose" in mock_warn.call_args[0][0]
+
+    def test_client_init_params_match(self):
+        import inspect
+
+        cp = inspect.signature(Client.__init__).parameters
+        acp = inspect.signature(AsyncClient.__init__).parameters
+        # Exclude 'self'; both should accept the same keyword args
+        assert set(cp) == set(acp), (
+            f"param mismatch: Client={set(cp)} AsyncClient={set(acp)}"
+        )
+
+    def test_client_methods_match(self):
+        sync_methods = {m for m in dir(Client) if not m.startswith("_")}
+        async_methods = {m for m in dir(AsyncClient) if not m.startswith("_")}
+        assert sync_methods == async_methods, (
+            f"only in Client: {sync_methods - async_methods}\n"
+            f"only in AsyncClient: {async_methods - sync_methods}"
+        )
diff --git a/manifest.json b/manifest.json
index 8a23774..f3d7800 100644
--- a/manifest.json
+++ b/manifest.json
@@ -1,6 +1,6 @@
 {
   "version": "1",
-  "generated": "2026-07-04T02:18:17.969946+00:00",
+  "generated": "2026-07-04T05:34:55.974821+00:00",
   "modules": {
     "a2a": {
       "description": "A2A (Agent-to-Agent Protocol) - Zero-dependency Python implementation",
@@ -163,12 +163,12 @@
       "files": [
         "httpclient/httpclient.py"
       ],
-      "version": "0.4.2",
+      "version": "0.4.4",
       "deps": [],
       "tier": "subsystem",
       "category": "network",
-      "last_updated": "2026-06-01T14:08:24-05:00",
-      "content_hash": "4878a4c5f1659b2dfd91cdc22e9276671e249c6cbe250dc576898c32a9b82877"
+      "last_updated": "2026-07-04T00:03:12-05:00",
+      "content_hash": "5621cb6b10e8c9b79750aa3790c6a9d6a5792649db7bc5117c01fd5aa61f6ef1"
     },
     "httpserver": {
       "description": "Zero-dependency async HTTP server with decorator-based routing",
@@ -364,7 +364,7 @@
       "deps": [],
       "tier": "subsystem",
       "category": "storage",
-      "last_updated": "2026-07-03T18:22:04-05:00",
+      "last_updated": "2026-07-03T21:18:24-05:00",
       "content_hash": "326e91696bf03f0c5ecd70541c94dbc522968252fd7ae702c3a4c5e3f32e4988"
     },
     "scheduler": {