Fix Windows proxy startup and bounded native preface handling

CHANGELOG.md

@@ -9,6 +9,16 @@ in the main repo).
 
 ## [Unreleased]
 
+### Fixed — Windows artifact validation
+
+- Keep proxy startup compatible with redirected cp1252 consoles.
+- Classify fragmented experimental native prefixes without the Windows-
+  incompatible combination of socket peek and wait-all flags; retain a bounded
+  deadline and leave HTTP-only service behavior unchanged.
+- Correct Windows test fixtures for executable discovery, home directories,
+  service-path quoting and corrupt trust-store rejection. Measure streaming
+  flush behavior after input delivery rather than including client setup.
+
 ### Security — bounded stable HTTP task path
 
 - Enforce the protocol's 1 MiB encoded request and response boundary for

src/iicp_client/cli.py

@@ -2783,7 +2783,7 @@ def _cmd_proxy(args: argparse.Namespace) -> int:
     # Flag/env precedence: explicit --host/--port override the TOML/env-loaded config.
     cfg.host = args.host
     cfg.port = args.port
-    print(f"iicp-node proxy → http://{cfg.host}:{cfg.port} (OpenAI/Ollama/Anthropic compat; no directory registration)")
+    print(f"iicp-node proxy -> http://{cfg.host}:{cfg.port} (OpenAI/Ollama/Anthropic compat; no directory registration)")
     uvicorn.run(create_app(cfg), host=cfg.host, port=cfg.port, server_header=False)
     return 0
 

src/iicp_client/native_preface.py

@@ -0,0 +1,18 @@
+"""Bounded, non-consuming classification for experimental native multiplexing."""
+import socket
+import time
+
+
+def peek_protocol_prefix(conn: socket.socket, magic: bytes, *, timeout: float) -> bytes:
+    deadline = time.monotonic() + timeout
+    while True:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise TimeoutError("incomplete protocol prefix")
+        conn.settimeout(remaining)
+        prefix = conn.recv(len(magic), socket.MSG_PEEK)
+        if not prefix or len(prefix) == len(magic) or not magic.startswith(prefix):
+            return prefix
+        # Peeking leaves partial bytes readable, so select alone would spin.
+        # Bound both the wait and retry rate until the entire magic arrives.
+        time.sleep(min(0.005, remaining))

src/iicp_client/node.py

@@ -40,6 +40,7 @@
 from iicp_client.effective_capability import EffectiveCapability, effective_capability_to_dict
 from iicp_client.idempotency import IdempotencyGuard
 from iicp_client.iicp_tcp import IICP_MAGIC, IicpTcpServer  # #457 single-port multiplexer
+from iicp_client.native_preface import peek_protocol_prefix
 from iicp_client.peer_manager import PeerManager
 from iicp_client.scheduler import QUEUE_WAIT_S, is_queue_eligible
 
@@ -1942,10 +1943,9 @@ async def _handle_native_conn(conn: socket.socket) -> None:
 
         def _route_conn(conn: socket.socket, addr: Any) -> None:
             try:
-                conn.settimeout(10.0)
-                # Wait for the full 4-byte prefix without consuming it; the chosen consumer
-                # then parses from the start. MSG_WAITALL avoids misrouting on a fragmented magic.
-                prefix = conn.recv(4, socket.MSG_PEEK | socket.MSG_WAITALL)
+                # Preserve fragmented magic without the non-portable
+                # MSG_PEEK|MSG_WAITALL combination (WSAEOPNOTSUPP on Windows).
+                prefix = peek_protocol_prefix(conn, IICP_MAGIC, timeout=10.0)
             except OSError:
                 try:
                     conn.close()

tests/test_backends.py

@@ -97,8 +97,12 @@ async def test_openai_compat_streaming_handler_flushes_at_utf8_byte_bound():
 
 
 class _DelayedSseStream(httpx.AsyncByteStream):
+    def __init__(self, delivered):
+        self.delivered = delivered
+
     async def __aiter__(self):
         yield b'data: {"choices":[{"delta":{"content":"timed"}}]}\n\n'
+        self.delivered.set()
         # Keep the next transport chunk well beyond the assertion deadline so
         # the test proves timer-driven flushing without depending on a 75 ms
         # scheduler window on loaded Windows builders.
@@ -108,8 +112,9 @@ async def __aiter__(self):
 
 @respx.mock
 async def test_openai_compat_streaming_handler_flushes_after_25_ms():
+    delivered = asyncio.Event()
     respx.post("http://localhost:11434/v1/chat/completions").mock(
-        return_value=httpx.Response(200, stream=_DelayedSseStream())
+        return_value=httpx.Response(200, stream=_DelayedSseStream(delivered))
     )
     handler = openai_compat_streaming_handler(model="qwen")
     events = handler(
@@ -119,9 +124,16 @@ async def test_openai_compat_streaming_handler_flushes_after_25_ms():
             "payload": {"messages": []},
         }
     )
-    first = await asyncio.wait_for(anext(events), timeout=0.25)
-    assert first == {"status": "partial", "result": "timed"}
-    await events.aclose()
+    # Measure flushing after input delivery, not Windows TLS/client setup.
+    first_task = asyncio.create_task(anext(events))
+    try:
+        await asyncio.wait_for(delivered.wait(), timeout=5.0)
+        first = await asyncio.wait_for(first_task, timeout=0.25)
+        assert first == {"status": "partial", "result": "timed"}
+    finally:
+        first_task.cancel()
+        await asyncio.gather(first_task, return_exceptions=True)
+        await events.aclose()
 
 
 @respx.mock

tests/test_dispatch_ticket_trust_store.py

@@ -87,13 +87,9 @@ def test_corruption_permissions_and_orphan_temp_are_fail_closed(tmp_path: Path)
 
     path.write_text("{not-json", encoding="utf-8")
     os.chmod(path, 0o600)
-    if os.name == "posix":
-        with pytest.raises(TrustBundleStoreCorrupt):
-            store.load()
-    else:
-        # Windows chmod cannot broaden an ACL; it only toggles the read-only
-        # attribute. The state remains protected by its owner-only directory.
-        assert store.load() is not None
+    # JSON corruption is invalid on every OS, independently of ACL semantics.
+    with pytest.raises(TrustBundleStoreCorrupt):
+        store.load()
 
     recovered = store.recover(
         _bundle("v1"), AdminRecoveryAuthorization("repair-corrupt-test", 1)

tests/test_native_preface.py

@@ -0,0 +1,43 @@
+import socket
+import threading
+import time
+from unittest.mock import Mock, patch
+
+import pytest
+
+from iicp_client.native_preface import peek_protocol_prefix
+
+
+def test_fragmented_magic_is_not_consumed(monkeypatch):
+    monkeypatch.delattr(socket, 'MSG_WAITALL', raising=False)
+    reader, writer = socket.socketpair()
+    def send():
+        for byte in b'IICP':
+            writer.sendall(bytes([byte]))
+            time.sleep(0.01)
+    thread = threading.Thread(target=send)
+    thread.start()
+    try:
+        assert peek_protocol_prefix(reader, b'IICP', timeout=2) == b'IICP'
+        assert reader.recv(4) == b'IICP'
+    finally:
+        thread.join()
+        reader.close()
+        writer.close()
+
+
+def test_partial_prefix_timeout_is_bounded():
+    conn = Mock()
+    conn.recv.return_value = b'I'
+    with patch('iicp_client.native_preface.time.monotonic', side_effect=[0, 0, 2]):
+        with pytest.raises(TimeoutError):
+            peek_protocol_prefix(conn, b'IICP', timeout=1)
+    assert conn.recv.call_count == 1
+
+
+@pytest.mark.parametrize('prefix', [b'', b'GET ', b'P'])
+def test_eof_and_http_prefixes_return_without_wait(prefix):
+    conn = Mock()
+    conn.recv.return_value = prefix
+    assert peek_protocol_prefix(conn, b'IICP', timeout=1) == prefix
+    conn.recv.assert_called_once_with(4, socket.MSG_PEEK)

tests/test_proxy_cli_encoding.py

@@ -0,0 +1,18 @@
+"""Redirected Windows consoles must not prevent proxy startup."""
+import io
+from argparse import Namespace
+from unittest.mock import Mock
+
+from iicp_client import cli
+
+
+def test_proxy_startup_supports_cp1252_stdout(monkeypatch, tmp_path):
+    run = Mock()
+    monkeypatch.setattr('uvicorn.run', run)
+    buffer = io.BytesIO()
+    output = io.TextIOWrapper(buffer, encoding='cp1252', errors='strict')
+    monkeypatch.setattr(cli.sys, 'stdout', output)
+    assert cli._cmd_proxy(Namespace(config=str(tmp_path / "absent.toml"), host='127.0.0.1', port=9484)) == 0
+    output.flush()
+    assert b'iicp-node proxy ->' in buffer.getvalue()
+    run.assert_called_once()

tests/test_runtime_health.py

@@ -28,6 +28,7 @@ def test_healthcheck_cli_exit_semantics(tmp_path, monkeypatch, capsys):
     from iicp_client.cli import _cmd_healthcheck
 
     monkeypatch.setenv("HOME", str(tmp_path))
+    monkeypatch.setenv("USERPROFILE", str(tmp_path))
     health = RuntimeHealth()
     health.mark_running()
     health.advance_runtime()

tests/test_service.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import shlex
+
 from iicp_client import cli
 from iicp_client.service import render_launchd, render_systemd
 
@@ -97,7 +99,7 @@ def test_service_preserves_only_explicit_tunnel_policy_and_resolved_binary(monke
     resolved = str(binary.resolve())
     assert f"<key>IICP_CLOUDFLARED_PATH</key><string>{resolved}</string>" in launchd.content
     assert "<key>IICP_TUNNEL</key><string>1</string>" in launchd.content
-    assert f"Environment=IICP_CLOUDFLARED_PATH={resolved}" in systemd.content
+    assert f"Environment=IICP_CLOUDFLARED_PATH={shlex.quote(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

tests/test_tunnel.py

@@ -53,7 +53,7 @@
 
 
 def test_cloudflared_override_is_absolute_executable_and_authoritative(monkeypatch, tmp_path):
-    binary = tmp_path / "cloudflared"
+    binary = tmp_path / ("cloudflared.exe" if sys.platform == "win32" else "cloudflared")
     binary.write_text("#!/bin/sh\nexit 0\n")
     binary.chmod(0o700)