Reject conflicted native task types before body reads

src/iicp_client/iicp_tcp.py

@@ -50,6 +50,26 @@
 _READ_CHUNK = 4096
 MAX_FRAME_PAYLOAD = 16 * 1024 * 1024  # Length-field payload bytes; header excluded.
 
+_STABLE_TASK_MESSAGE_TYPES = frozenset((*range(0x01, 0x0B), 0x0D, 0x0E))
+
+
+def stable_task_message_type_error(msg_type: int) -> str | None:
+    """Return the fail-closed stable-task disposition for one type byte.
+
+    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.
+    """
+    if msg_type in _STABLE_TASK_MESSAGE_TYPES:
+        return None
+    if msg_type in {0x00, 0xFF}:
+        return "invalid_type"
+    if msg_type in {0x0B, 0x0C}:
+        return "conflicted_type"
+    if 0xF0 <= msg_type <= 0xFE:
+        return "unsupported_extension"
+    return "unknown_type"
+
 
 class MsgType(IntEnum):
     """spec/iicp-framing.md §3 — core message types 0x01–0x0E.
@@ -404,6 +424,10 @@ async def _session(
             if version != FRAMING_VERSION:
                 logger.warning("Unsupported IICP framing version — closing")
                 return
+            type_error = stable_task_message_type_error(msg_type)
+            if type_error is not None:
+                logger.warning("Rejected native task frame type 0x%02x: %s", msg_type, type_error)
+                return
             if payload_len > MAX_FRAME_PAYLOAD:
                 logger.warning("IICP frame payload exceeds limit — closing")
                 return
@@ -936,6 +960,9 @@ async def _read_frame(self, timeout_s: float | None = None) -> tuple[int, bytes]
             raise IicpTcpClientError(f"bad magic in response: {magic!r}")
         if version != FRAMING_VERSION:
             raise IicpTcpClientError(f"unsupported framing version {version} in response")
+        type_error = stable_task_message_type_error(mt)
+        if type_error is not None:
+            raise IicpTcpClientError(f"rejected native task frame type 0x{mt:02x}: {type_error}")
         if payload_len > MAX_FRAME_PAYLOAD:
             raise IicpTcpClientError(f"response frame payload too large: {payload_len} > {MAX_FRAME_PAYLOAD}")
         payload = await asyncio.wait_for(self._reader.readexactly(payload_len), timeout=t) if payload_len else 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 only; dispatch, TLS, lifecycle, experimental relay opcodes, logical fragmentation and unsupported QUIC behavior are outside this fixture.",
+  "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.",
   "frame": {
     "framing_version": 1,
     "header_bytes": 12,
@@ -48,6 +48,33 @@
     "max_payload_bytes": 16777216,
     "length_semantics": "payload_bytes_excluding_12_byte_header"
   },
+  "stable_task_profile": {
+    "status": "implementation_backed_pre_ratification",
+    "accepted_message_types": [
+      1,
+      2,
+      3,
+      4,
+      5,
+      6,
+      7,
+      8,
+      9,
+      10,
+      13,
+      14
+    ],
+    "conflicted_message_types": [
+      11,
+      12
+    ],
+    "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",
+    "plaintext_scope": "development_only",
+    "stable_claim": "not_admitted"
+  },
   "scenarios": [
     {
       "name": "ping_empty",
@@ -137,5 +164,76 @@
         "reason": "payload_too_large"
       }
     }
+  ],
+  "stable_task_type_scenarios": [
+    {
+      "name": "init_is_in_stable_task_profile",
+      "message_type": 1,
+      "expected": {
+        "outcome": "accept"
+      }
+    },
+    {
+      "name": "call_is_in_stable_task_profile",
+      "message_type": 5,
+      "expected": {
+        "outcome": "accept"
+      }
+    },
+    {
+      "name": "observe_is_in_stable_task_profile",
+      "message_type": 13,
+      "expected": {
+        "outcome": "accept"
+      }
+    },
+    {
+      "name": "zero_is_invalid",
+      "message_type": 0,
+      "expected": {
+        "outcome": "reject",
+        "reason": "invalid_type"
+      }
+    },
+    {
+      "name": "control_relay_bind_collision_is_rejected",
+      "message_type": 11,
+      "expected": {
+        "outcome": "reject",
+        "reason": "conflicted_type"
+      }
+    },
+    {
+      "name": "advertise_relay_ack_collision_is_rejected",
+      "message_type": 12,
+      "expected": {
+        "outcome": "reject",
+        "reason": "conflicted_type"
+      }
+    },
+    {
+      "name": "future_core_type_is_unknown",
+      "message_type": 15,
+      "expected": {
+        "outcome": "reject",
+        "reason": "unknown_type"
+      }
+    },
+    {
+      "name": "unnegotiated_private_extension_is_rejected",
+      "message_type": 240,
+      "expected": {
+        "outcome": "reject",
+        "reason": "unsupported_extension"
+      }
+    },
+    {
+      "name": "maximum_type_is_invalid",
+      "message_type": 255,
+      "expected": {
+        "outcome": "reject",
+        "reason": "invalid_type"
+      }
+    }
   ]
 }

tests/test_iicp_tcp.py

@@ -280,6 +280,29 @@ async def test_server_rejects_oversized_length_before_body_read(server_port):
         await writer.wait_closed()
 
 
+async def test_server_rejects_conflicted_type_before_body_read(server_port):
+    reader, writer = await asyncio.open_connection("127.0.0.1", server_port)
+    try:
+        writer.write(_frame(MsgType.INIT, cbor2.dumps({1: FRAMING_VERSION}, canonical=True)))
+        await writer.drain()
+        await _read_frame(reader)
+        writer.write(
+            _HEADER.pack(
+                IICP_MAGIC,
+                FRAMING_VERSION,
+                0x0B,
+                0,
+                0,
+                MAX_FRAME_PAYLOAD,
+            )
+        )
+        await writer.drain()
+        assert await asyncio.wait_for(reader.read(1), timeout=TIMEOUT) == b""
+    finally:
+        writer.close()
+        await writer.wait_closed()
+
+
 async def test_payload_bearing_frame_does_not_close_session(server_port):
     """Regression guard for the iter-1410 adapter bug — pre-fix the session loop
     closed on every frame with a non-empty CBOR payload because IicpFrame.decode
@@ -348,6 +371,33 @@ async def oversized_ack(reader: asyncio.StreamReader, writer: asyncio.StreamWrit
         await server.wait_closed()
 
 
+async def test_client_rejects_conflicted_response_type_before_body_read():
+    async def conflicted_ack(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        await _read_frame(reader)
+        writer.write(
+            _HEADER.pack(
+                IICP_MAGIC,
+                FRAMING_VERSION,
+                0x0B,
+                0,
+                0,
+                MAX_FRAME_PAYLOAD,
+            )
+        )
+        await writer.drain()
+        writer.close()
+
+    server = await asyncio.start_server(conflicted_ack, "127.0.0.1", 0)
+    port = server.sockets[0].getsockname()[1]
+    try:
+        async with IicpTcpClient("127.0.0.1", port) as client:
+            with pytest.raises(IicpTcpClientError, match="conflicted_type"):
+                await client.handshake()
+    finally:
+        server.close()
+        await server.wait_closed()
+
+
 async def test_client_rejects_ack_with_wrong_negotiated_version():
     async def wrong_ack(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
         await _read_frame(reader)

tests/test_native_framing_fixture.py

@@ -5,7 +5,13 @@
 import json
 from pathlib import Path
 
-from iicp_client.iicp_tcp import FRAME_HEADER_LEN, MAX_FRAME_PAYLOAD, IicpFrame, MsgType
+from iicp_client.iicp_tcp import (
+    FRAME_HEADER_LEN,
+    MAX_FRAME_PAYLOAD,
+    IicpFrame,
+    MsgType,
+    stable_task_message_type_error,
+)
 
 FIXTURE = Path(__file__).parent / "fixtures" / "native-framing-v1.json"
 
@@ -56,3 +62,11 @@ def test_native_frame_encoder_rejects_payload_above_the_declared_limit() -> None
         assert "frame payload too large" in str(error)
     else:
         raise AssertionError("oversized payload must be rejected")
+
+
+def test_stable_task_type_boundary_matches_canonical_vectors() -> None:
+    data = json.loads(FIXTURE.read_text())
+    for scenario in data["stable_task_type_scenarios"]:
+        expected = scenario["expected"]
+        actual = stable_task_message_type_error(scenario["message_type"])
+        assert actual == expected.get("reason"), scenario["name"]