diff --git a/README.md b/README.md
index 8ea2182..fb6a9f0 100644
--- a/README.md
+++ b/README.md
@@ -405,7 +405,8 @@ asyncio.run(main())
 
 ### Listen port — default 9484, auto-increment (v0.7.5+)
 
-The official IICP port **9484** is the default listen port (`IICP_PORT`, `--port`).
+The unassigned project-default port **9484** is the default listen port
+(`IICP_PORT`, `--port`); it is not an IANA-assigned IICP service port.
 The `iicp-node` CLI auto-increments to the next free port when 9484 is already in
 use, so you can run several nodes on one host without picking ports by hand — the
 first binds 9484, the second 9485, the third 9486, and so on. Each node gets its
@@ -416,6 +417,26 @@ give it as-is (no auto-increment at the library level).
 
 ---
 
+### Experimental native TCP boundary
+
+Provider nodes serve the supported HTTP task path by default. Installing the
+`iicp-tcp` extra does not mount or advertise the native TCP draft, which is
+outside the coordinated stable and production support baseline. A direct
+development endpoint requires both the extra and an explicit runtime opt-in:
+
+```bash
+python -m pip install 'iicp-client[iicp-tcp]'
+IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP=1 iicp-node serve --node my-node
+```
+
+Automatic derivation accepts only direct `http://` endpoints and produces
+plaintext `iicp://`. It never rewrites an `https://` endpoint to `iicpsec://`,
+because an HTTPS reverse proxy or Quick Tunnel does not prove a native TLS
+route. Generated launchd and systemd units omit the setting by default and
+preserve it only when explicitly configured.
+
+---
+
 ## Backends
 
 A provider node forwards each task to an inference backend. The backend is selected
diff --git a/pyproject.toml b/pyproject.toml
index 23a6baa..0669a3a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -50,9 +50,9 @@ Documentation = "https://iicp.network/docs"
 metrics = [
     "prometheus-client>=0.20",
 ]
-# Native IICP binary transport (port 9484, spec/iicp-framing.md). Optional
-# because cbor2 has a C-extension build that can fail on some musl-based
-# minimal containers — HTTP-only nodes don't need it.
+# Experimental plaintext native IICP draft (provisional port 9484). Installing
+# this extra does not mount or advertise it; provider nodes also require
+# IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP=1. It is excluded from stable/production.
 iicp-tcp = [
     "cbor2>=5.4",
 ]
diff --git a/src/iicp_client/cli.py b/src/iicp_client/cli.py
index 3753037..66fdb86 100644
--- a/src/iicp_client/cli.py
+++ b/src/iicp_client/cli.py
@@ -91,6 +91,18 @@ def _env_bool(name: str, default: bool = False) -> bool:
     return value.strip().lower() in {"1", "true", "yes"}
 
 
+def _explicit_bool_env(name: str) -> bool | None:
+    value = os.environ.get(name)
+    if value is None:
+        return None
+    normalized = value.strip().lower()
+    if normalized in {"1", "true", "yes"}:
+        return True
+    if normalized in {"0", "false", "no"}:
+        return False
+    raise ValueError(f"{name} must be one of 1/true/yes or 0/false/no")
+
+
 def _managed_operator_decision(args, tunnel_preference, operator) -> tuple[bool, str]:
     """Evaluate the local managed profile before exposure or registration."""
     from iicp_client.operator_profile import ManagedOperatorInput, evaluate_managed_operator
@@ -427,6 +439,10 @@ def _build_parser() -> argparse.ArgumentParser:
         "Dead-state policy: IICP_TUNNEL_DEAD_POLICY=auto|retry|exit|log-only; "
         "generated services set IICP_SUPERVISED=1. env: IICP_TUNNEL=1/0",
     )
+    serve.epilog = (
+        "Experimental native TCP is disabled by default and excluded from stable/production claims. "
+        "For a direct development endpoint only: IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP=1."
+    )
     serve.add_argument(
         "--relay-capable",
         action=argparse.BooleanOptionalAction,
@@ -1818,7 +1834,7 @@ async def _serve(args: argparse.Namespace) -> int:
         return 2
 
     # Resolve the actual listen port before NAT detection: start at the
-    # requested port (default 9484, the official IICP port) and auto-increment
+    # requested port (default 9484, the unassigned project convention) and auto-increment
     # to the next free port. This keeps one port per node (multiple models on
     # one node share it) while N nodes on one host each get a distinct port →
     # distinct pinhole. Skipped when the operator supplies an explicit
@@ -2131,17 +2147,23 @@ async def _serve(args: argparse.Namespace) -> int:
         logger.error(str(exc))
         return 2
 
-    # #457 / ADR-040 — advertise the native IICP binary transport. serve() multiplexes it
-    # onto the SAME socket as HTTP (first-byte detection), so transport_endpoint shares the
-    # endpoint's host:port with the iicp:// scheme. Derived from the FINAL endpoint (after NAT
-    # profile application); register() only sends it when registering (skip_registration gates
-    # the non-routable case) → advertise-when-reachable. Opt out: IICP_DISABLE_NATIVE_TRANSPORT=1.
-    if not args.skip_registration and os.environ.get("IICP_DISABLE_NATIVE_TRANSPORT") != "1":
+    # Native TCP remains a development-only draft outside the coordinated
+    # stable support baseline. It must be explicitly enabled, and HTTPS must
+    # never be rewritten to iicpsec because this server has no native TLS path.
+    if _explicit_bool_env("IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP") is True:
         from iicp_client.node import derive_native_endpoint
 
         _native_ep = derive_native_endpoint(node._cfg.endpoint)
         if _native_ep:
             node._cfg.transport_endpoint = _native_ep
+            logger.warning(
+                "Experimental plaintext native TCP enabled; it is excluded from stable and production claims."
+            )
+        else:
+            logger.warning(
+                "Experimental native TCP was requested but no direct HTTP endpoint can be derived; "
+                "HTTPS/tunnel endpoints are not native TLS routes, so the listener remains disabled."
+            )
 
     # #404 — register with bounded backoff retry. On persistent failure, pass an
     # empty token (NOT None) so the heartbeat loop still starts and re-registers on
@@ -2621,7 +2643,7 @@ def _check_dependencies(backend_url: str) -> list[_DepIssue]:
 
     # 2) Optional Python deps mapped to pip extras
     optional = [
-        ("cbor2", "iicp-tcp", "native IICP TCP transport (port 9484)"),
+        ("cbor2", "iicp-tcp", "experimental native IICP TCP draft (disabled by default; not stable/production)"),
         ("upnpclient", "nat", "UPnP NAT detection + IPv6 firewall pinhole"),
         ("ifaddr", "nat", "interface enumeration for NAT detection"),
         ("prometheus_client", "metrics", "/metrics endpoint"),
diff --git a/src/iicp_client/iicp_tcp.py b/src/iicp_client/iicp_tcp.py
index a61be55..b50a6c8 100644
--- a/src/iicp_client/iicp_tcp.py
+++ b/src/iicp_client/iicp_tcp.py
@@ -1,5 +1,9 @@
 # SPDX-License-Identifier: Apache-2.0
-"""Native IICP binary transport (port 9484) — server + framing + cbor payloads.
+"""Experimental native IICP binary transport — server + framing + CBOR payloads.
+
+Provider nodes require the separate ``IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP=1``
+runtime opt-in before mounting or advertising this plaintext development
+binding. It is excluded from stable and production support claims.
 
 Implements the wire side of spec/iicp-framing.md so a hybrid-client SDK node
 can answer task CALLs over the native binary transport instead of (or in
@@ -58,7 +62,7 @@ def stable_task_message_type_error(msg_type: int) -> str | None:
 
     Relay experiments retain 0x0B/0x0C on their dedicated transport. Those
     bytes conflict with the inherited CONTROL/ADVERTISE registry and therefore
-    cannot enter a stable native task session.
+    cannot enter the bounded experimental native task profile.
     """
     if msg_type in _STABLE_TASK_MESSAGE_TYPES:
         return None
diff --git a/src/iicp_client/nat_detection.py b/src/iicp_client/nat_detection.py
index d997d28..97ef920 100644
--- a/src/iicp_client/nat_detection.py
+++ b/src/iicp_client/nat_detection.py
@@ -136,8 +136,8 @@ async def detect_nat(
         external_ip_probe_url: opt-in WAN-IP probe URL (e.g. api.ipify.org).
             Used as a fallback when UPnP AddPortMapping succeeds but the IGD
             refuses GetExternalIPAddress (FRITZ!Box auth-restricted case).
-        transport_port: optional native IICP TCP port (default 9484 per
-            spec/iicp-dir.md v0.7.0). When set AND distinct from `bind_port`,
+        transport_port: optional experimental native IICP TCP port. The stable
+            default is disabled. When set AND distinct from `bind_port`,
             the detector asks UPnP to map BOTH ports and returns a
             transport_endpoint URL alongside the HTTP public_endpoint.
     """
diff --git a/src/iicp_client/node.py b/src/iicp_client/node.py
index a07e219..b92b562 100644
--- a/src/iicp_client/node.py
+++ b/src/iicp_client/node.py
@@ -121,17 +121,15 @@ async def _post_cip_receipt(
 
 
 def derive_native_endpoint(endpoint: str) -> str | None:
-    """#457 / ADR-040 — derive the native binary transport_endpoint from the HTTP `endpoint`.
+    """Derive the experimental plaintext native endpoint from direct HTTP.
 
-    They share one host:port (serve() multiplexes both planes on one socket via first-byte
-    detection), so the native URI is the same authority with the ``iicp`` scheme (``iicpsec``
-    for TLS). Returns None if `endpoint` is not a parseable http(s) URL.
+    The maintained server has no native TLS terminator. An HTTPS reverse proxy
+    or tunnel therefore cannot be assumed to carry the binary protocol and is
+    never rewritten to ``iicpsec://`` automatically.
     """
     parts = urlsplit(endpoint)
     if parts.scheme == "http" and parts.netloc:
         return f"iicp://{parts.netloc}"
-    if parts.scheme == "https" and parts.netloc:
-        return f"iicpsec://{parts.netloc}"
     return None
 
 
@@ -280,10 +278,10 @@ class NodeConfig:
     max_concurrent: int = 4
     tokens_per_min: int = 10000
     max_tokens: int = 8192
-    # spec/iicp-dir.md v0.7.0 — optional native IICP binary endpoint (ADR-040).
-    # Scheme MUST be iicp:// (plaintext) or iicpsec:// (TLS); default port 9484.
-    # When set, the directory persists it and clients SHOULD prefer it over
-    # `endpoint` for task CALLs. Leave None for HTTP-only operation.
+    # Experimental native IICP binary endpoint (ADR-040), disabled by default.
+    # iicp:// is plaintext development use; iicpsec:// requires a real native
+    # TLS terminator. Port 9484 is an unassigned project convention. When set,
+    # the directory persists it for explicitly enabled experimental peers.
     transport_endpoint: str | None = None
     # #331 Phase A.1 / ADR-041 — NAT-traversal observability fields surfaced
     # to the directory in the register payload. Populated automatically by
@@ -1856,15 +1854,15 @@ def _json_response(self, status: int, body: bytes, cors: bool = True) -> None:
                 self.end_headers()
                 self.wfile.write(body)
 
-        # #457 / ADR-040 — single-port multiplexer: the HTTP control plane and the native
-        # IICP binary transport share ONE socket. Each accepted connection's first 4 bytes
-        # are peeked (MSG_PEEK, non-consuming): the IICP frame magic "IICP" routes to the
-        # native handler (the SAME backend task handler as HTTP), anything else (an HTTP
-        # request line) to the BaseHTTPRequestHandler above. One socket ⇒ one pinhole ⇒
-        # native is reachable exactly when HTTP is (advertise-when-reachable); a CGNAT node
-        # needs no second hole. bind_and_activate=False: we own the listening socket.
+        # The experimental native multiplexer is mounted only when the
+        # operator explicitly configured transport_endpoint. Ordinary nodes
+        # remain HTTP-only; compiling or importing the draft is not consent to
+        # expose it on the public listener.
+        native_enabled = bool(self._cfg.transport_endpoint)
         server = ThreadingHTTPServer((host, port), _Handler, bind_and_activate=False)
-        native_server = IicpTcpServer(host=host, port=port, node_id=self._cfg.node_id, handler=handler)
+        native_server = (
+            IicpTcpServer(host=host, port=port, node_id=self._cfg.node_id, handler=handler) if native_enabled else None
+        )
         # Bind to the address family implied by `host` — the CLI defaults host to
         # "::" (IPv6), which a hardcoded AF_INET socket cannot bind (gaierror).
         family = _listen_family(host, port)
@@ -1887,6 +1885,9 @@ def _json_response(self, status: int, body: bytes, cors: bool = True) -> None:
         mux_stop = threading.Event()
 
         async def _handle_native_conn(conn: socket.socket) -> None:
+            if native_server is None:
+                conn.close()
+                return
             try:
                 conn.setblocking(False)
                 reader, writer = await asyncio.open_connection(sock=conn)
@@ -1911,7 +1912,7 @@ def _route_conn(conn: socket.socket, addr: Any) -> None:
                     pass
                 return
             conn.settimeout(None)
-            if prefix == IICP_MAGIC:
+            if native_enabled and prefix == IICP_MAGIC:
                 asyncio.run_coroutine_threadsafe(_handle_native_conn(conn), loop)
             else:
                 # ThreadingHTTPServer.process_request threads the request; _Handler reads the
@@ -1929,12 +1930,16 @@ def _accept_loop() -> None:
                 # Peek+route off the accept thread so a slow client can't block new connections.
                 threading.Thread(target=_route_conn, args=(conn, addr), daemon=True).start()
 
-        logger.info(
-            "IICP node %s listening on %s:%d (HTTP + native IICP, single port)",
-            self._cfg.node_id,
-            host,
-            port,
-        )
+        if native_enabled:
+            logger.warning(
+                "IICP node %s enabled experimental plaintext native TCP on %s:%d; "
+                "it is excluded from stable and production claims",
+                self._cfg.node_id,
+                host,
+                port,
+            )
+        else:
+            logger.info("IICP node %s listening on %s:%d (HTTP)", self._cfg.node_id, host, port)
         self._runtime_health.mark_running()
 
         bg_tasks: list[asyncio.Task] = []
diff --git a/src/iicp_client/proxy/routing/router.py b/src/iicp_client/proxy/routing/router.py
index 5dc4426..fd4aeb1 100644
--- a/src/iicp_client/proxy/routing/router.py
+++ b/src/iicp_client/proxy/routing/router.py
@@ -16,9 +16,11 @@
     - ADR-008 — directory score ordering is authoritative; the proxy preserves it
     - spec/iicp-core.md §10 — retry/idempotency semantics for client implementations
 """
+
 from __future__ import annotations
 
 import logging
+import os
 from typing import Any
 from uuid import UUID
 
@@ -29,6 +31,23 @@
 logger = logging.getLogger(__name__)
 
 
+def _experimental_native_enabled() -> bool:
+    """Return whether directory native routes may enter this proxy path.
+
+    The stable proxy ignores ``transport_endpoint`` by default. Direct callers
+    can still construct ``NodeClient`` with an endpoint for bounded experiments.
+    """
+    raw = os.environ.get("IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP")
+    if raw is None:
+        return False
+    value = raw.strip().lower()
+    if value in {"1", "true", "yes"}:
+        return True
+    if value not in {"0", "false", "no"}:
+        logger.warning("Ignoring invalid IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP value in proxy routing")
+    return False
+
+
 class TaskRouter:
     """Dispatch a task to one node, respecting retry + circuit-breaker policy.
 
@@ -72,7 +91,7 @@ async def route(
         """
         node_id = node["node_id"]
         endpoint = node["endpoint"]
-        transport_endpoint = node.get("transport_endpoint")  # spec v0.7.0 dual-endpoint fallback
+        transport_endpoint = node.get("transport_endpoint") if _experimental_native_enabled() else None
 
         if not _is_ssrf_safe(endpoint):
             logger.warning(
@@ -80,9 +99,7 @@ async def route(
                 node_id[:8] if len(node_id) > 8 else node_id,
                 endpoint,
             )
-            raise ValueError(
-                f"Node endpoint '{endpoint}' is not publicly routable (SSRF guard)"
-            )
+            raise ValueError(f"Node endpoint '{endpoint}' is not publicly routable (SSRF guard)")
 
         self._circuit.check(node_id)
 
@@ -90,7 +107,10 @@ async def route(
 
         async def attempt() -> dict[str, Any]:
             return await client.submit_task(
-                task_id, intent, payload, timeout_ms,
+                task_id,
+                intent,
+                payload,
+                timeout_ms,
                 cip_envelope=cip_envelope,
                 source_node_id=source_node_id,
             )
diff --git a/src/iicp_client/service.py b/src/iicp_client/service.py
index 6e39260..238282c 100644
--- a/src/iicp_client/service.py
+++ b/src/iicp_client/service.py
@@ -151,6 +151,15 @@ def _supervisor_tunnel_environment() -> dict[str, str]:
         result["IICP_CLOUDFLARED_PATH"] = binary
     if normalized is not None:
         result["IICP_TUNNEL"] = normalized
+    native = os.environ.get("IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP")
+    if native is not None:
+        value = native.strip().lower()
+        if value in {"1", "true", "yes"}:
+            result["IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP"] = "1"
+        elif value in {"0", "false", "no"}:
+            result["IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP"] = "0"
+        else:
+            raise ValueError("IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP must be one of 1/true/yes or 0/false/no")
     return result
 
 
diff --git a/tests/fixtures/native-framing-v1.json b/tests/fixtures/native-framing-v1.json
index 46b9daf..53495de 100644
--- a/tests/fixtures/native-framing-v1.json
+++ b/tests/fixtures/native-framing-v1.json
@@ -1,7 +1,7 @@
 {
   "fixture_version": "1.0.0-draft",
   "status": "implementation-backed-pre-ratification",
-  "purpose": "Cross-implementation native framing vectors for the current ordered-stream binding. They cover bounded frame decoding and the fail-closed stable task-session type boundary. Dispatch direction, TLS, lifecycle, dedicated experimental relay sessions, logical fragmentation and unsupported QUIC behavior remain outside this fixture.",
+  "purpose": "Cross-implementation native framing vectors for the current experimental ordered-stream binding. They cover bounded frame decoding and the fail-closed task-session type boundary. Passing these vectors does not admit native TCP to the coordinated stable or production baseline. Dispatch direction, TLS, lifecycle, dedicated experimental relay sessions, logical fragmentation and unsupported QUIC behavior remain outside this fixture.",
   "frame": {
     "framing_version": 1,
     "header_bytes": 12,
@@ -71,7 +71,7 @@
     "conflict": "0x0B/0x0C are CONTROL/ADVERTISE in the draft registry but RELAY_BIND/RELAY_ACK in maintained relay experiments; stable task sessions reject both bytes before proportional allocation.",
     "relay_boundary": "Experimental relay sessions remain isolated on their dedicated transport and do not count as stable task-profile conformance.",
     "extension_boundary": "0xF0-0xFE require negotiated extension state; the current stable task profile negotiates none and rejects them before proportional allocation.",
-    "production_security_disposition": "open_qualify_or_exclude",
+    "production_security_disposition": "excluded_from_stable_baseline",
     "plaintext_scope": "development_only",
     "stable_claim": "not_admitted"
   },
diff --git a/tests/proxy/test_router.py b/tests/proxy/test_router.py
index 3a4dcc8..c54ed52 100644
--- a/tests/proxy/test_router.py
+++ b/tests/proxy/test_router.py
@@ -1,4 +1,5 @@
 """Unit tests for TaskRouter — circuit breaker integration + retry wiring."""
+
 from __future__ import annotations
 
 from unittest.mock import AsyncMock, patch
@@ -38,6 +39,7 @@ def _make_router(
 # Happy path
 # ---------------------------------------------------------------------------
 
+
 @pytest.mark.asyncio
 async def test_route_returns_result_on_success():
     """PROXY-ROUTE-01: Router discovers nodes via directory and routes; returns backend result on success."""
@@ -74,6 +76,7 @@ async def test_route_records_success_on_circuit_breaker():
 # Circuit breaker integration
 # ---------------------------------------------------------------------------
 
+
 @pytest.mark.asyncio
 async def test_route_raises_circuit_open_when_breaker_tripped():
     """Router raises CircuitOpenError immediately when circuit is open."""
@@ -105,6 +108,7 @@ async def test_route_records_failure_on_exception():
 # Retry integration
 # ---------------------------------------------------------------------------
 
+
 @pytest.mark.asyncio
 async def test_route_retries_on_transient_error():
     """Router retries on httpx.ConnectError (retriable per RetryManager) then succeeds."""
@@ -143,6 +147,7 @@ async def test_route_raises_after_all_retries_exhausted():
 # Token forwarding
 # ---------------------------------------------------------------------------
 
+
 @pytest.mark.asyncio
 async def test_route_uses_configured_node_token():
     """Router creates NodeClient with the configured node_token."""
@@ -162,10 +167,51 @@ async def submit_task(self, *args, **kwargs) -> dict:
     assert captured_token == ["my-secret-token"]
 
 
+@pytest.mark.asyncio
+async def test_route_ignores_experimental_transport_endpoint_by_default(monkeypatch):
+    """Stable proxy routing must not opt itself into the native draft."""
+    monkeypatch.delenv("IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP", raising=False)
+    captured_transport: list[str | None] = []
+
+    class CapturingClient:
+        def __init__(self, endpoint: str, token: str, transport_endpoint: str | None = None):
+            captured_transport.append(transport_endpoint)
+
+        async def submit_task(self, *args, **kwargs) -> dict:
+            return {"status": "success"}
+
+    node = {**NODE, "transport_endpoint": "iicp://1.2.3.4:9484"}
+    with patch("iicp_client.proxy.routing.router.NodeClient", CapturingClient):
+        await _make_router().route(node, TASK_ID, INTENT, PAYLOAD, TIMEOUT_MS)
+
+    assert captured_transport == [None]
+
+
+@pytest.mark.asyncio
+async def test_route_preserves_explicit_experimental_transport_opt_in(monkeypatch):
+    monkeypatch.setenv("IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP", "1")
+    captured_transport: list[str | None] = []
+
+    class CapturingClient:
+        def __init__(self, endpoint: str, token: str, transport_endpoint: str | None = None):
+            captured_transport.append(transport_endpoint)
+
+        async def submit_task(self, *args, **kwargs) -> dict:
+            return {"status": "success"}
+
+    native = "iicp://1.2.3.4:9484"
+    node = {**NODE, "transport_endpoint": native}
+    with patch("iicp_client.proxy.routing.router.NodeClient", CapturingClient):
+        await _make_router().route(node, TASK_ID, INTENT, PAYLOAD, TIMEOUT_MS)
+
+    assert captured_transport == [native]
+
+
 # ---------------------------------------------------------------------------
 # CIP-CALL-01: cip envelope passthrough (S.12 §4.1, §10.4)
 # ---------------------------------------------------------------------------
 
+
 @pytest.mark.asyncio
 async def test_route_passes_cip_envelope_to_submit_task():
     """CIP-CALL-01: cip_envelope provided to route() must reach NodeClient.submit_task().
diff --git a/tests/test_serve_multiplex.py b/tests/test_serve_multiplex.py
index 9d3d689..8651795 100644
--- a/tests/test_serve_multiplex.py
+++ b/tests/test_serve_multiplex.py
@@ -1,11 +1,5 @@
 # SPDX-License-Identifier: Apache-2.0
-"""#457 / ADR-040 — `iicp-node serve` multiplexes the HTTP control plane and the native
-IICP binary transport on ONE port (first-byte detection). Proves BOTH planes answer on the
-same socket, and that transport_endpoint derives from the HTTP endpoint.
-
-Fails without the fix: pre-#457 serve() bound only an HTTP server on the port, so a native
-IICP CALL would hit the HTTP parser and never get a RESPONSE.
-"""
+"""The native TCP draft is mounted only after an explicit endpoint opt-in."""
 
 from __future__ import annotations
 
@@ -15,9 +9,11 @@
 from http.client import HTTPConnection
 from typing import Any
 
+import pytest
+
 from iicp_client import IicpNode, NodeConfig
 from iicp_client._confidentiality import decrypt_response, encrypt_payload_with_context
-from iicp_client.iicp_tcp import IicpTcpClient
+from iicp_client.iicp_tcp import IicpTcpClient, IicpTcpClientError
 from iicp_client.node import derive_native_endpoint
 
 CHAT = "urn:iicp:intent:llm:chat:v1"
@@ -58,9 +54,7 @@ async def _wait_port(port: int) -> None:
     loop = asyncio.get_event_loop()
     for _ in range(50):
         try:
-            await loop.run_in_executor(
-                None, lambda: socket.create_connection(("127.0.0.1", port), timeout=0.1).close()
-            )
+            await loop.run_in_executor(None, lambda: socket.create_connection(("127.0.0.1", port), timeout=0.1).close())
             return
         except OSError:
             await asyncio.sleep(0.05)
@@ -75,12 +69,11 @@ async def test_http_and_native_call_share_one_port() -> None:
         region="test-region",
         model="test-model",
         max_concurrent=4,
+        transport_endpoint="iicp://127.0.0.1:9484",
     )
     node = IicpNode(cfg)
     port = _free_port()
-    serve_task = asyncio.create_task(
-        node.serve(_echo, host="127.0.0.1", port=port, node_token=None)
-    )
+    serve_task = asyncio.create_task(node.serve(_echo, host="127.0.0.1", port=port, node_token=None))
     try:
         await _wait_port(port)
 
@@ -103,10 +96,34 @@ async def test_http_and_native_call_share_one_port() -> None:
 
 def test_derive_native_endpoint() -> None:
     assert derive_native_endpoint("http://203.0.113.5:9484") == "iicp://203.0.113.5:9484"
-    assert derive_native_endpoint("https://node.example:9484") == "iicpsec://node.example:9484"
+    assert derive_native_endpoint("https://node.example:9484") is None
     assert derive_native_endpoint("not-a-url") is None
 
 
+async def test_native_call_is_not_mounted_without_explicit_endpoint() -> None:
+    cfg = NodeConfig(
+        node_id="http-only-node",
+        endpoint="http://test-node.local",
+        intent=CHAT,
+        region="test-region",
+        model="test-model",
+        max_concurrent=4,
+    )
+    assert cfg.transport_endpoint is None
+    node = IicpNode(cfg)
+    port = _free_port()
+    serve_task = asyncio.create_task(node.serve(_echo, host="127.0.0.1", port=port, node_token=None))
+    try:
+        await _wait_port(port)
+        async with IicpTcpClient("127.0.0.1", port) as client:
+            with pytest.raises((asyncio.IncompleteReadError, TimeoutError, IicpTcpClientError)):
+                await asyncio.wait_for(client.handshake(), timeout=2.0)
+    finally:
+        serve_task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await serve_task
+
+
 async def test_http_task_decrypts_iicp_conf(monkeypatch, tmp_path) -> None:
     monkeypatch.setenv("IICP_CX_KEY_DIR", str(tmp_path / "cx"))
     cfg = NodeConfig(
@@ -166,12 +183,15 @@ async def test_http_rejects_required_encrypted_response_without_encrypted_reques
         await _wait_port(port)
         status, body = await asyncio.get_event_loop().run_in_executor(
             None,
-            lambda: _http_task(port, {
-                "task_id": "cx-plain-required",
-                "intent": CHAT,
-                "payload": {"secret": True},
-                "cx_response_encryption": "required",
-            }),
+            lambda: _http_task(
+                port,
+                {
+                    "task_id": "cx-plain-required",
+                    "intent": CHAT,
+                    "payload": {"secret": True},
+                    "cx_response_encryption": "required",
+                },
+            ),
         )
         assert status == 400
         assert body["error"]["code"] == "IICP-CX-03"
diff --git a/tests/test_service.py b/tests/test_service.py
index 3d6d67c..d3d775b 100644
--- a/tests/test_service.py
+++ b/tests/test_service.py
@@ -8,6 +8,7 @@ def test_launchd_unit_runs_foreground_serve_with_hourly_auto_update(monkeypatch,
     monkeypatch.setenv("HOME", str(tmp_path))
     monkeypatch.delenv("IICP_AUTO_UPDATE", raising=False)
     monkeypatch.delenv("IICP_AUTO_UPDATE_INTERVAL_S", raising=False)
+    monkeypatch.delenv("IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP", raising=False)
     unit = render_launchd("mynode")
 
     assert unit.platform == "launchd"
@@ -20,6 +21,7 @@ def test_launchd_unit_runs_foreground_serve_with_hourly_auto_update(monkeypatch,
     assert "<key>IICP_SUPERVISED</key><string>1</string>" in unit.content
     assert "<key>IICP_TUNNEL_DEAD_POLICY</key><string>auto</string>" in unit.content
     assert "<key>KeepAlive</key><true/>" in unit.content
+    assert "<key>IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP</key>" not in unit.content
     assert "--daemon" not in unit.content
 
 
@@ -27,6 +29,7 @@ def test_systemd_unit_runs_foreground_serve_with_hourly_auto_update(monkeypatch,
     monkeypatch.setenv("HOME", str(tmp_path))
     monkeypatch.delenv("IICP_AUTO_UPDATE", raising=False)
     monkeypatch.delenv("IICP_AUTO_UPDATE_INTERVAL_S", raising=False)
+    monkeypatch.delenv("IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP", raising=False)
     unit = render_systemd("mynode")
 
     assert unit.platform == "systemd"
@@ -37,6 +40,7 @@ def test_systemd_unit_runs_foreground_serve_with_hourly_auto_update(monkeypatch,
     assert "Environment=IICP_SUPERVISED=1" in unit.content
     assert "Environment=IICP_TUNNEL_DEAD_POLICY=auto" in unit.content
     assert "Restart=on-failure" in unit.content
+    assert "Environment=IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP=" not in unit.content
     assert "--daemon" not in unit.content
 
 
@@ -86,6 +90,7 @@ def test_service_preserves_only_explicit_tunnel_policy_and_resolved_binary(monke
     binary.chmod(0o700)
     monkeypatch.setenv("IICP_CLOUDFLARED_PATH", str(binary))
     monkeypatch.setenv("IICP_TUNNEL", "yes")
+    monkeypatch.setenv("IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP", "yes")
 
     launchd = render_launchd("mynode")
     systemd = render_systemd("mynode")
@@ -94,6 +99,8 @@ def test_service_preserves_only_explicit_tunnel_policy_and_resolved_binary(monke
     assert "<key>IICP_TUNNEL</key><string>1</string>" in launchd.content
     assert f"Environment=IICP_CLOUDFLARED_PATH={resolved}" in systemd.content
     assert "Environment=IICP_TUNNEL=1" in systemd.content
+    assert "<key>IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP</key><string>1</string>" in launchd.content
+    assert "Environment=IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP=1" in systemd.content
 
     monkeypatch.delenv("IICP_TUNNEL")
     automatic = render_launchd("mynode")
@@ -113,3 +120,12 @@ def test_service_refuses_invalid_or_unavailable_forced_tunnel(monkeypatch, tmp_p
     monkeypatch.setenv("IICP_TUNNEL", "1")
     with pytest.raises(ValueError, match="requires cloudflared"):
         render_systemd("mynode")
+
+
+def test_service_refuses_invalid_experimental_native_setting(monkeypatch, tmp_path):
+    import pytest
+
+    monkeypatch.setenv("HOME", str(tmp_path))
+    monkeypatch.setenv("IICP_ENABLE_EXPERIMENTAL_NATIVE_TCP", "sometimes")
+    with pytest.raises(ValueError, match="must be one of"):
+        render_launchd("mynode")