feat: add descriptor-relative qpk vnext n2 store

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

docs/qpk_vnext_n2_descriptor_store.md

@@ -0,0 +1,13 @@
+# QPK-N2 descriptor-relative store
+
+This fresh reslice is local-only and consumes only the merged N1 vNext contract.
+It opens and traverses every path component relative to verified directory file
+descriptors with `O_NOFOLLOW`; pathname `resolve()` checks are not used as a
+security boundary. Missing directories are created one segment at a time and
+new ancestor descriptors are fsynced. Payload files are fsynced and installed
+with create-only linking, preserving write-once/idempotent semantics under
+concurrency. Selector listing decodes every candidate before returning it.
+
+The store accepts durable records only, rejects ephemeral before filesystem I/O,
+has exact reads and explicit selector listing, and defers latest. Legacy paths,
+network/S3, callers, orchestrators, exporters, and live behavior are excluded.

src/quant_platform_kit/strategy_lifecycle/qpk_vnext_n2_store.py

@@ -0,0 +1,251 @@
+"""Descriptor-relative local store for the clean-slate qpk-vnext/result/v2 contract."""
+from __future__ import annotations
+
+import errno
+import json
+import os
+import stat
+import tempfile
+from pathlib import Path
+from typing import Any
+
+from quant_platform_kit.strategy_lifecycle.qpk_vnext_n1 import (
+    NAMESPACE, ContractError, ResultContract, _segment, decode_wire,
+)
+
+
+class StoreError(ValueError):
+    """Sanitized isolated-store failure."""
+
+
+def _fail() -> None:
+    raise StoreError("invalid qpk-vnext isolated store operation")
+
+
+class IsolatedResultStore:
+    """Filesystem store using descriptor-relative, no-follow operations only."""
+
+    _DIR_FLAGS = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
+    _FILE_FLAGS = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+
+    def __init__(self, root: str | os.PathLike[str]) -> None:
+        if not isinstance(root, (str, os.PathLike)) or not os.name == "posix":
+            _fail()
+        self.root = Path(root).expanduser()
+        if not self.root.is_absolute():
+            _fail()
+
+    @staticmethod
+    def _fsync(fd: int) -> None:
+        try:
+            os.fsync(fd)
+        except OSError:
+            _fail()
+
+    def _root_fd(self) -> int:
+        try:
+            return os.open(self.root, self._DIR_FLAGS)
+        except OSError:
+            _fail()
+
+    @staticmethod
+    def _validate_key(key: str) -> list[str]:
+        if not isinstance(key, str) or "\\" in key:
+            _fail()
+        parts = key.split("/")
+        if len(parts) != 11 or parts[:3] != NAMESPACE.split("/") or not parts[-1].endswith(".json"):
+            _fail()
+        if any(not part or part in {".", ".."} for part in parts):
+            _fail()
+        for part in parts[3:-1]:
+            try:
+                _segment(part.removeprefix("i").removeprefix("p") if part[:1] in {"i", "p"} else part)
+            except ContractError:
+                _fail()
+        return parts
+
+    @classmethod
+    def _open_child_dir(cls, parent: int, name: str, *, create: bool, created: list[int]) -> int:
+        try:
+            return os.open(name, cls._DIR_FLAGS, dir_fd=parent)
+        except FileNotFoundError:
+            if not create:
+                _fail()
+            try:
+                os.mkdir(name, mode=0o700, dir_fd=parent)
+                cls._fsync(parent)
+            except FileExistsError:
+                pass
+            except OSError:
+                _fail()
+            try:
+                child = os.open(name, cls._DIR_FLAGS, dir_fd=parent)
+            except OSError:
+                _fail()
+            created.append(child)
+            return child
+        except OSError:
+            _fail()
+
+    @classmethod
+    def _walk(cls, root_fd: int, parts: list[str], *, create: bool) -> tuple[int, list[int]]:
+        current = root_fd
+        opened: list[int] = []
+        for part in parts:
+            current = cls._open_child_dir(current, part, create=create, created=opened)
+        return current, opened
+
+    @staticmethod
+    def _wire_bytes(contract: ResultContract) -> bytes:
+        if not isinstance(contract, ResultContract) or contract.persist_mode != "durable":
+            _fail()
+        try:
+            checked = decode_wire(contract.to_wire())
+            return json.dumps(checked.to_wire(), ensure_ascii=False, sort_keys=True,
+                              separators=(",", ":"), allow_nan=False).encode("utf-8")
+        except (ContractError, TypeError, ValueError, UnicodeError):
+            _fail()
+
+    def put(self, contract: ResultContract) -> str:
+        payload = self._wire_bytes(contract)
+        parts = self._validate_key(contract.key)
+        root_fd = self._root_fd()
+        parent_fd = None
+        created: list[int] = []
+        temp_name: str | None = None
+        try:
+            parent_fd, created = self._walk(root_fd, parts[:-1], create=True)
+            target = parts[-1]
+            try:
+                existing_fd = os.open(target, self._FILE_FLAGS, dir_fd=parent_fd)
+            except FileNotFoundError:
+                existing_fd = None
+            except OSError:
+                _fail()
+            if existing_fd is not None:
+                try:
+                    existing = os.read(existing_fd, len(payload) + 1)
+                finally:
+                    os.close(existing_fd)
+                if existing == payload:
+                    return "idempotent"
+                _fail()
+            for _ in range(8):
+                candidate = f".{target}.{os.getpid()}.{next(tempfile._RandomNameSequence())}.tmp"
+                try:
+                    fd = os.open(candidate, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o600, dir_fd=parent_fd)
+                    temp_name = candidate
+                    break
+                except FileExistsError:
+                    continue
+                except OSError:
+                    _fail()
+            else:
+                _fail()
+            with os.fdopen(fd, "wb") as handle:
+                handle.write(payload)
+                handle.flush()
+                os.fsync(handle.fileno())
+            try:
+                os.link(temp_name, target, src_dir_fd=parent_fd, dst_dir_fd=parent_fd, follow_symlinks=False)
+            except FileExistsError:
+                try:
+                    existing_fd = os.open(target, self._FILE_FLAGS, dir_fd=parent_fd)
+                    existing = os.read(existing_fd, len(payload) + 1)
+                    os.close(existing_fd)
+                except OSError:
+                    _fail()
+                if existing != payload:
+                    _fail()
+                return "idempotent"
+            self._fsync(parent_fd)
+            for fd in reversed(created):
+                self._fsync(fd)
+            os.unlink(temp_name, dir_fd=parent_fd)
+            temp_name = None
+            self._fsync(parent_fd)
+            return "created"
+        except (OSError, ValueError):
+            _fail()
+        finally:
+            if temp_name and parent_fd is not None:
+                try:
+                    os.unlink(temp_name, dir_fd=parent_fd)
+                except OSError:
+                    pass
+            for fd in reversed(created):
+                try:
+                    os.close(fd)
+                except OSError:
+                    pass
+            os.close(root_fd)
+
+    def get(self, key: str) -> ResultContract:
+        parts = self._validate_key(key)
+        root_fd = self._root_fd()
+        parent_fd = None
+        try:
+            parent_fd, opened = self._walk(root_fd, parts[:-1], create=False)
+            fd = os.open(parts[-1], self._FILE_FLAGS, dir_fd=parent_fd)
+            try:
+                data: Any = json.loads(os.read(fd, 10_000_000).decode("utf-8"))
+            finally:
+                os.close(fd)
+            item = decode_wire(data)
+            if item.key != key:
+                _fail()
+            return item
+        except (OSError, UnicodeError, json.JSONDecodeError, ContractError):
+            _fail()
+        finally:
+            for fd in reversed(opened if 'opened' in locals() else []):
+                os.close(fd)
+            os.close(root_fd)
+
+    def list_keys(self, *, domain: str, profile: str, timing: str) -> tuple[str, ...]:
+        try:
+            domain, profile = _segment(domain), _segment(profile)
+        except ContractError:
+            _fail()
+        if timing not in {"next_open", "next_close"}:
+            _fail()
+        root_fd = self._root_fd()
+        profile_fd = None
+        try:
+            profile_fd, opened = self._walk(root_fd, NAMESPACE.split("/") + [domain, profile], create=False)
+            found: list[str] = []
+
+            def scan(fd: int, rel: list[str]) -> None:
+                try:
+                    names = os.listdir(fd)
+                except OSError:
+                    _fail()
+                for name in names:
+                    if not name or name in {".", ".."}:
+                        continue
+                    try:
+                        info = os.stat(name, dir_fd=fd, follow_symlinks=False)
+                    except OSError:
+                        _fail()
+                    if stat.S_ISLNK(info.st_mode):
+                        _fail()
+                    if stat.S_ISDIR(info.st_mode):
+                        child = os.open(name, self._DIR_FLAGS, dir_fd=fd)
+                        try:
+                            scan(child, rel + [name])
+                        finally:
+                            os.close(child)
+                    elif name.endswith(".json"):
+                        key = "/".join(NAMESPACE.split("/") + [domain, profile] + rel + [name])
+                        parts = self._validate_key(key)
+                        if parts[7] != timing:
+                            continue
+                        item = self.get(key)
+                        found.append(item.key)
+
+            scan(profile_fd, [])
+            return tuple(sorted(set(found)))
+        finally:
+            for fd in reversed(opened if 'opened' in locals() else []):
+                os.close(fd)
+            os.close(root_fd)

tests/test_qpk_vnext_n2_store.py

@@ -0,0 +1,88 @@
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import replace
+
+import pytest
+
+from quant_platform_kit.strategy_lifecycle.qpk_vnext_n1 import ResultContract
+from quant_platform_kit.strategy_lifecycle.qpk_vnext_n2_store import IsolatedResultStore, StoreError
+
+
+def contract(**changes):
+    values = dict(domain="us_equity", profile="SOXL", timing="next_open", identity_version=1,
+                  persist_mode="durable", strategy_id="strategy", run_id="run-1",
+                  param_set_id="baseline", param_version=1, source_revision="rev1",
+                  computed_at="2026-07-15T00:00:00.000000Z", params={"x": 1})
+    values.update(changes)
+    return ResultContract(**values)
+
+
+def test_write_read_idempotent_and_ephemeral_side_effect_free(tmp_path):
+    store = IsolatedResultStore(tmp_path)
+    item = contract()
+    assert store.put(item) == "created"
+    assert store.put(item) == "idempotent"
+    assert store.get(item.key) == item
+    before = sorted(tmp_path.rglob("*"))
+    with pytest.raises(StoreError):
+        store.put(replace(item, persist_mode="ephemeral"))
+    assert sorted(tmp_path.rglob("*")) == before
+
+
+def test_symlink_root_and_key_segments_fail_closed(tmp_path):
+    target = tmp_path / "target"
+    target.mkdir()
+    link = tmp_path / "link"
+    link.symlink_to(target, target_is_directory=True)
+    store = IsolatedResultStore(link)
+    with pytest.raises(StoreError):
+        store.get(contract().key)
+    store = IsolatedResultStore(target)
+    with pytest.raises(StoreError):
+        store.get("qpk-vnext/result/v2/us_equity/SOXL/../run/next_open/i1/p1/x.json")
+    item = contract()
+    store.put(item)
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    leaf = target.joinpath(*item.key.split("/")[:-1])
+    leaf.rename(target / "leaf-backup")
+    leaf.symlink_to(outside, target_is_directory=True)
+    with pytest.raises(StoreError):
+        store.get(item.key)
+
+
+def test_concurrent_identical_and_conflicting_writers(tmp_path):
+    store = IsolatedResultStore(tmp_path)
+    item = contract()
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        outcomes = list(pool.map(lambda _: store.put(item), range(8)))
+    assert outcomes.count("created") == 1
+    assert outcomes.count("idempotent") == 7
+    with pytest.raises(StoreError):
+        store.put(contract(computed_at="2026-07-16T00:00:00.000000Z"))
+
+
+def test_validated_listing_exact_timing_and_corrupt_rejection(tmp_path):
+    store = IsolatedResultStore(tmp_path)
+    first = contract()
+    second = contract(timing="next_close")
+    store.put(first)
+    store.put(second)
+    assert store.list_keys(domain="us_equity", profile="SOXL", timing="next_open") == (first.key,)
+    assert store.list_keys(domain="us_equity", profile="SOXL", timing="next_close") == (second.key,)
+    path = tmp_path / second.key
+    path.write_bytes(b"{}")
+    with pytest.raises(StoreError):
+        store.list_keys(domain="us_equity", profile="SOXL", timing="next_close")
+    for profile in ("..", "SOXL/../x"):
+        with pytest.raises(StoreError):
+            store.list_keys(domain="us_equity", profile=profile, timing="next_open")
+
+
+def test_temp_cleanup_on_atomic_failure(tmp_path, monkeypatch):
+    store = IsolatedResultStore(tmp_path)
+    monkeypatch.setattr("quant_platform_kit.strategy_lifecycle.qpk_vnext_n2_store.os.link", lambda *_a, **_k: (_ for _ in ()).throw(OSError("injected")))
+    with pytest.raises(StoreError):
+        store.put(contract())
+    assert not list(tmp_path.rglob("*.tmp"))