diff --git a/docs/qpk_vnext_n2_isolated_store.md b/docs/qpk_vnext_n2_isolated_store.md
new file mode 100644
index 0000000..b3666aa
--- /dev/null
+++ b/docs/qpk_vnext_n2_isolated_store.md
@@ -0,0 +1,16 @@
+# QPK vNext N2 isolated filesystem store
+
+N2 stores only `qpk-vnext/result/v2` records below an explicitly supplied local
+root. It never reads or scans legacy namespaces, performs no cloud/network I/O,
+and intentionally provides no implicit `latest` operation. Callers must use an
+exact contract key (or the explicit domain/profile/timing selector listing).
+
+Only `persist_mode=durable` is accepted. Ephemeral contracts fail before any
+filesystem side effect. Writes validate through the N1 encode/decode contract,
+derive the complete path from `contract.key`, and use atomic temp-file replace.
+Existing byte-identical canonical JSON is an idempotent no-op; any conflict or
+corruption fails closed without overwrite. The destination is installed with a
+create-only link operation, so concurrent writers cannot overwrite a committed
+record. File and directory metadata are fsynced before success, and temporary
+files are cleaned up on failure. All key paths receive root-containment and
+single-segment checks; selector timing is an exact key segment.
diff --git a/src/quant_platform_kit/strategy_lifecycle/qpk_vnext_n2_store.py b/src/quant_platform_kit/strategy_lifecycle/qpk_vnext_n2_store.py
new file mode 100644
index 0000000..cd19851
--- /dev/null
+++ b/src/quant_platform_kit/strategy_lifecycle/qpk_vnext_n2_store.py
@@ -0,0 +1,166 @@
+"""Filesystem-only store for the clean-slate qpk-vnext/result/v2 contract.
+
+This module deliberately has no legacy fallback, listing-by-latest semantics,
+cloud client, or caller integration.  Keys are validated and contained below
+the explicitly supplied root before any filesystem operation.
+"""
+from __future__ import annotations
+
+import json
+import os
+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:
+    """Write-once local store rooted at an explicit directory."""
+
+    def __init__(self, root: str | os.PathLike[str]) -> None:
+        if not isinstance(root, (str, os.PathLike)):
+            _fail()
+        self.root = Path(root).expanduser().resolve()
+        if not self.root.is_absolute():
+            _fail()
+
+    def _path_for_key(self, key: str) -> Path:
+        if not isinstance(key, str) or not key 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()
+        candidate = (self.root.joinpath(*parts)).resolve()
+        try:
+            candidate.relative_to(self.root)
+        except ValueError:
+            _fail()
+        return candidate
+
+    @staticmethod
+    def _bytes(contract: ResultContract) -> bytes:
+        if contract.persist_mode != "durable":
+            _fail()
+        try:
+            wire = contract.to_wire()
+            checked = decode_wire(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()
+
+    @staticmethod
+    def _sync_directory(directory: Path) -> None:
+        try:
+            fd = os.open(directory, os.O_RDONLY)
+            try:
+                os.fsync(fd)
+            finally:
+                os.close(fd)
+        except OSError:
+            _fail()
+
+    def put(self, contract: ResultContract) -> str:
+        """Atomically create the contract; repeat identical writes are no-ops."""
+        if not isinstance(contract, ResultContract):
+            _fail()
+        payload = self._bytes(contract)
+        path = self._path_for_key(contract.key)
+        if path.exists():
+            try:
+                existing = path.read_bytes()
+            except OSError:
+                _fail()
+            if existing == payload:
+                return "idempotent"
+            _fail()
+        path.parent.mkdir(parents=True, exist_ok=True)
+        self._sync_directory(path.parent)
+        temporary: str | None = None
+        try:
+            fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent)
+            with os.fdopen(fd, "wb") as handle:
+                handle.write(payload)
+                handle.flush()
+                os.fsync(handle.fileno())
+            try:
+                os.link(temporary, path)
+            except FileExistsError:
+                try:
+                    existing = path.read_bytes()
+                except OSError:
+                    _fail()
+                if existing == payload:
+                    return "idempotent"
+                _fail()
+            self._sync_directory(path.parent)
+            os.unlink(temporary)
+            temporary = None
+            return "created"
+        except (OSError, ValueError):
+            _fail()
+        finally:
+            if temporary:
+                try:
+                    os.unlink(temporary)
+                except OSError:
+                    pass
+
+    def get(self, key: str) -> ResultContract:
+        path = self._path_for_key(key)
+        if not path.is_file():
+            _fail()
+        try:
+            data: Any = json.loads(path.read_text(encoding="utf-8"))
+            contract = decode_wire(data)
+        except (OSError, UnicodeError, json.JSONDecodeError, ContractError):
+            _fail()
+        if contract.key != key:
+            _fail()
+        return contract
+
+    def list_keys(self, *, domain: str, profile: str, timing: str) -> tuple[str, ...]:
+        """List exact selector matches; no implicit latest or legacy scan."""
+        try:
+            domain = _segment(domain)
+            profile = _segment(profile)
+        except ContractError:
+            _fail()
+        if timing not in {"next_open", "next_close"}:
+            _fail()
+        prefix = f"{NAMESPACE}/{domain}/{profile}/"
+        base = (self.root / NAMESPACE / domain / profile).resolve()
+        try:
+            base.relative_to(self.root)
+        except ValueError:
+            _fail()
+        if not base.is_dir():
+            return ()
+        found: list[str] = []
+        for path in base.rglob("*.json"):
+            try:
+                key = path.resolve().relative_to(self.root).as_posix()
+                parts = key.split("/")
+                if (len(parts) == 11 and key.startswith(prefix)
+                        and parts[7] == timing and self._path_for_key(key) == path.resolve()):
+                    found.append(key)
+            except ValueError:
+                continue
+        return tuple(sorted(found))
diff --git a/tests/test_qpk_vnext_n2_store.py b/tests/test_qpk_vnext_n2_store.py
new file mode 100644
index 0000000..08fc9ec
--- /dev/null
+++ b/tests/test_qpk_vnext_n2_store.py
@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+import json
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import replace
+
+import pytest
+
+from quant_platform_kit.strategy_lifecycle.qpk_vnext_n1 import ContractError, 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_durable_write_read_and_idempotent(tmp_path):
+    store = IsolatedResultStore(tmp_path)
+    item = contract()
+    assert store.put(item) == "created"
+    assert store.put(item) == "idempotent"
+    assert store.get(item.key) == item
+    assert list(tmp_path.rglob("*.json"))
+
+
+def test_ephemeral_has_no_side_effect(tmp_path):
+    store = IsolatedResultStore(tmp_path)
+    with pytest.raises(StoreError):
+        store.put(replace(contract(), persist_mode="ephemeral"))
+    assert not list(tmp_path.rglob("*"))
+
+
+def test_conflict_corrupt_missing_and_traversal_fail_closed(tmp_path):
+    store = IsolatedResultStore(tmp_path)
+    item = contract()
+    store.put(item)
+    path = tmp_path / item.key
+    path.write_bytes(b"conflict")
+    with pytest.raises(StoreError):
+        store.put(item)
+    with pytest.raises(StoreError):
+        store.get(item.key)
+    for key in ("../x.json", "/tmp/x.json", item.key.replace("strategy", "../x")):
+        with pytest.raises(StoreError):
+            store.get(key)
+    with pytest.raises(StoreError):
+        store.get(item.key + ".missing")
+
+
+def test_selector_listing_is_explicit_and_ignores_legacy(tmp_path):
+    store = IsolatedResultStore(tmp_path)
+    first = contract()
+    second = contract(timing="next_close")
+    store.put(first)
+    store.put(second)
+    (tmp_path / "legacy" / "result.json").parent.mkdir()
+    (tmp_path / "legacy" / "result.json").write_text("{}")
+    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,)
+    with pytest.raises(StoreError):
+        store.list_keys(domain="us_equity", profile="..", timing="next_open")
+    with pytest.raises(StoreError):
+        store.list_keys(domain="us_equity", profile="SOXL/..", timing="next_open")
+    with pytest.raises(StoreError):
+        store.list_keys(domain="us_equity", profile="SOXL", timing="next_openx")
+
+
+def test_atomic_failure_cleans_temp(tmp_path, monkeypatch):
+    store = IsolatedResultStore(tmp_path)
+    def fail_replace(*_args):
+        raise OSError("injected")
+    monkeypatch.setattr("quant_platform_kit.strategy_lifecycle.qpk_vnext_n2_store.os.link", fail_replace)
+    with pytest.raises(StoreError):
+        store.put(contract())
+    assert not list(tmp_path.rglob("*.tmp"))
+
+
+def test_concurrent_put_is_create_only(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
+    assert store.get(item.key) == item