Fix Windows capture paths and public WebSocket text subscriptions

docs/m7-crypto-l2-capture.md

@@ -91,6 +91,13 @@ operator action.
 
 ## Configuration and CLI
 
+On Windows with long-path support disabled, use extended-length local paths in
+the JSON configuration, for example `"hot_root": "\\\\?\\D:\\capture-hot"`.
+Apply the same form to the archive and restore roots. Journal and temporary
+filenames can exceed the legacy path limit even with a short root. The physical
+volume probe maps this spelling to the same drive device and still checks the
+underlying disks; UNC shares and non-drive device paths remain unsupported.
+
 Example configuration, with operator-selected absolute paths:
 
 ```json

src/quant_data_kit/capture_v2/storage.py

@@ -93,7 +93,14 @@ class VolumeDiskExtents(ctypes.Structure):
         _fields_ = [("count", wintypes.DWORD), ("extents", DiskExtent * 1)]
 
     anchor = Path(path).resolve(strict=True).anchor.rstrip("\\/")
-    if len(anchor) != 2 or anchor[1] != ":":
+    # Extended-length local paths retain this prefix after resolve(). Only the
+    # volume device handle needs the ordinary drive form; keep file paths long.
+    anchor = anchor.removeprefix("\\\\?\\")
+    if (
+        len(anchor) != 2
+        or anchor[0] not in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
+        or anchor[1] != ":"
+    ):
         raise CapturePausedError(f"cannot resolve Windows volume anchor for {path}")
     volume_path = rf"\\.\{anchor}"
     create_file = ctypes.windll.kernel32.CreateFileW

src/quant_data_kit/capture_v2/transport.py

@@ -70,7 +70,9 @@ def __init__(self, connection: object) -> None:
         self._connection = connection
 
     async def send(self, payload: bytes) -> None:
-        await self._connection.send(payload)  # type: ignore[attr-defined]
+        # Subscription JSON is a UTF-8 text message. Sending bytes directly uses
+        # a binary frame, which OKX rejects as an invalid latency probe.
+        await self._connection.send(payload.decode("utf-8"))  # type: ignore[attr-defined]
 
     async def receive(self, *, timeout_seconds: float) -> bytes:
         try:

tests/test_capture_v2_collector_cli_transport.py

@@ -684,10 +684,10 @@ def test_websocket_transport_wrapper_bytes_text_timeout_and_tls(
     class Underlying:
         def __init__(self, values: list[Any]) -> None:
             self.values = values
-            self.sent: list[bytes] = []
+            self.sent: list[str] = []
             self.closed = False
 
-        async def send(self, payload: bytes) -> None:
+        async def send(self, payload: str) -> None:
             self.sent.append(payload)
 
         async def recv(self):
@@ -712,7 +712,7 @@ async def close(self):
         asyncio.run(wrapped.receive(timeout_seconds=0.001))
     asyncio.run(wrapped.send(b"subscribe"))
     asyncio.run(wrapped.close())
-    assert underlying.sent == [b"subscribe"] and underlying.closed
+    assert underlying.sent == ["subscribe"] and underlying.closed
 
     connector = WebsocketsConnector()
     with pytest.raises(ValidationError, match="wss"):

tests/test_capture_websocket_text.py

@@ -0,0 +1,26 @@
+import asyncio
+
+from websockets.asyncio.client import connect
+from websockets.asyncio.server import serve
+
+from quant_data_kit.capture_v2.transport import _WebsocketsConnection
+
+
+def test_subscription_is_a_text_frame_on_the_wire():
+    async def exercise():
+        received = asyncio.get_running_loop().create_future()
+
+        async def handle(connection):
+            received.set_result(await connection.recv())
+
+        async with serve(handle, "127.0.0.1", 0) as server:
+            port = server.sockets[0].getsockname()[1]
+            async with connect(f"ws://127.0.0.1:{port}", proxy=None) as connection:
+                wrapped = _WebsocketsConnection(connection)
+                payload = b'{"op":"subscribe","args":[{"channel":"books","instId":"BTC-USDT"}]}'
+                await wrapped.send(payload)
+                observed = await asyncio.wait_for(received, timeout=3)
+                assert isinstance(observed, str)
+                assert observed.encode("utf-8") == payload
+
+    asyncio.run(exercise())

tests/test_capture_windows_extended_paths.py

@@ -0,0 +1,51 @@
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from quant_data_kit.capture_v2 import storage
+
+
+@pytest.mark.parametrize("anchor", ["D:\\", "\\\\?\\D:\\"])
+def test_local_drive_paths_open_the_same_physical_volume(anchor, monkeypatch):
+    from ctypes import wintypes
+
+    class ResolvedPath:
+        def __init__(self, value):
+            self.anchor = anchor
+
+        def resolve(self, *, strict):
+            assert strict
+            return self
+
+    class OpenVolume:
+        def __call__(self, device, *_args):
+            assert device == "\\\\.\\D:"
+            # An OS denial still fails closed after resolving either spelling.
+            return wintypes.HANDLE(-1).value
+
+    monkeypatch.setattr(storage, "Path", ResolvedPath)
+    monkeypatch.setattr(
+        storage.ctypes,
+        "windll",
+        SimpleNamespace(kernel32=SimpleNamespace(CreateFileW=OpenVolume())),
+        raising=False,
+    )
+    with pytest.raises(storage.CapturePausedError, match="cannot open volume"):
+        storage._windows_volume_identity(Path("unused"))
+
+
+@pytest.mark.parametrize(
+    "anchor", ["\\\\server\\share\\", "\\\\?\\UNC\\server\\share\\", "\\\\?\\Volume{abc}\\", "1:\\"]
+)
+def test_unsupported_windows_anchors_do_not_reach_device_probe(anchor, monkeypatch):
+    class ResolvedPath:
+        def __init__(self, value):
+            self.anchor = anchor
+
+        def resolve(self, *, strict):
+            return self
+
+    monkeypatch.setattr(storage, "Path", ResolvedPath)
+    with pytest.raises(storage.CapturePausedError, match="cannot resolve Windows volume anchor"):
+        storage._windows_volume_identity(Path("unused"))