From 04d65375dfbc00690fa124dad293e32ef36474d8 Mon Sep 17 00:00:00 2001
From: PureSaber <88314620+PureSaber@users.noreply.github.com>
Date: Tue, 29 Sep 2026 16:34:47 +0800
Subject: [PATCH 1/4] feat: add PIT-safe action bridge and verified historical
 halt calendar

---
 docs/PUBLIC_CORPORATE_ACTIONS.md              |   8 ++
 src/quant_data_kit/providers/action_bridge.py | 115 ++++++++++++++++++
 src/quant_data_kit/providers/halt_calendar.py |  73 +++++++++++
 tests/test_action_bridge.py                   |  93 ++++++++++++++
 tests/test_halt_calendar.py                   |  45 +++++++
 5 files changed, 334 insertions(+)
 create mode 100644 src/quant_data_kit/providers/action_bridge.py
 create mode 100644 src/quant_data_kit/providers/halt_calendar.py
 create mode 100644 tests/test_action_bridge.py
 create mode 100644 tests/test_halt_calendar.py

diff --git a/docs/PUBLIC_CORPORATE_ACTIONS.md b/docs/PUBLIC_CORPORATE_ACTIONS.md
index f4bf283..e2bb062 100644
--- a/docs/PUBLIC_CORPORATE_ACTIONS.md
+++ b/docs/PUBLIC_CORPORATE_ACTIONS.md
@@ -6,6 +6,14 @@ Ordinary and special CASH dividends on the same ex-date are added only when thei
 
 `providers.tradability.fetch_tradability` captures the current risk-warning board and the requested date's suspension records. `no_reported_restriction` means only that these two responses report no restriction. It does not certify historical listing status, delistings, the next session or price-limit execution. Missing schemas fail rather than turning into an empty restriction list.
 
+`providers.action_bridge.distribution_terms` turns one normalized distribution into separate dividend entitlement, dividend payment and split terms. Rights, mergers and spin-offs are not read from the dividend feed. `explicit_terms` accepts only a complete evidenced record for those kinds.
+
+The bridge preserves `captured_at` as `available_at`; ex/payment dates are economic event dates, not evidence of historical data availability. A later capture must not be backdated for PIT replay.
+
+`providers.halt_calendar.freeze_halt_calendar` stores supplied symbol-session observations and a SHA-256 of the canonical payload. Lookup of a missing session fails. The current Eastmoney ST/halt response cannot be frozen as this calendar.
+
+Every lookup verifies the stored SHA-256 before returning a status. The digest detects changes to the frozen payload; it does not authenticate the source or certify historical completeness.
+
 Primary API field references: [AKShare stock documentation](https://akshare.akfamily.xyz/data/stock/stock.html), sections `stock_dividend_cninfo`, `stock_zh_a_st_em`, and `stock_tfp_em`.
 
 2026-09-19 integration observation: CNInfo returned 109 normalized distribution records across the four example symbols. Eastmoney's current ST endpoint disconnected; no replacement status was fabricated. Tencent's adjusted series and CNInfo cash amounts disagree for the 000333 June 2026 distribution (price adjustment 3.72 versus cash 3.80 per share). Consumers must keep this discrepancy visible; the daily simulation refuses an unexplained adjustment window.
diff --git a/src/quant_data_kit/providers/action_bridge.py b/src/quant_data_kit/providers/action_bridge.py
new file mode 100644
index 0000000..5b11da9
--- /dev/null
+++ b/src/quant_data_kit/providers/action_bridge.py
@@ -0,0 +1,115 @@
+"""Turn evidenced corporate-action rows into execution ActionTerms.
+
+CNInfo distributions already carry cash, bonus shares and transfers. Rights,
+mergers and spin-offs are not inferred from that feed: they enter only through
+an explicit record that satisfies ActionTerms.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from decimal import Decimal
+
+import pandas as pd
+
+from quant_data_kit.financial.actions import KINDS, ActionTerms
+
+_EXPLICIT_KINDS = {"merger", "spin_off", "rights_distribution", "rights_exercise"}
+
+
+def _utc(value, field: str) -> str:
+    stamp = pd.to_datetime(value, errors="coerce")
+    if pd.isna(stamp):
+        raise ValueError(f"corporate action is missing {field}")
+    if stamp.tzinfo is not None:
+        stamp = stamp.tz_convert("UTC")
+    else:
+        stamp = stamp.tz_localize("UTC")
+    return stamp.strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def _date(value, field: str) -> str:
+    stamp = pd.to_datetime(value, errors="coerce")
+    if pd.isna(stamp):
+        raise ValueError(f"corporate action is missing {field}")
+    return pd.Timestamp(stamp).strftime("%Y-%m-%d")
+
+
+def distribution_terms(row, *, instrument_id: str, currency: str = "CNY") -> list[ActionTerms]:
+    """Split one normalized CNInfo distribution into dividend and split terms."""
+    cash = Decimal(str(row.cash_per_share))
+    ratio = Decimal(str(row.share_ratio))
+    if not cash.is_finite() or not ratio.is_finite() or cash < 0 or ratio <= 0:
+        raise ValueError("distribution amounts must be non-negative with a positive share ratio")
+    if cash == 0 and ratio == 1:
+        raise ValueError("distribution has neither cash nor a share change")
+    event_id = str(row.event_id)
+    source = str(row.source)
+    record_date = _date(row.record_date, "record_date")
+    ex_at = _utc(row.ex_date, "ex_date")
+    # A historical date in a feed is not evidence that we had the row then.
+    available_at = _utc(row.captured_at, "captured_at")
+    terms: list[ActionTerms] = []
+    if cash > 0:
+        pay_at = _utc(row.pay_date, "pay_date")
+        common = {
+            "instrument_id": instrument_id,
+            "currency": currency,
+            "source": source,
+            "cash_per_unit": str(cash),
+            "entitlement_date": record_date,
+        }
+        terms.append(
+            ActionTerms(
+                event_id=event_id + ":entitlement",
+                kind="dividend_entitlement",
+                effective_at=ex_at,
+                available_at=available_at,
+                evidence_id=event_id,
+                **common,
+            )
+        )
+        terms.append(
+            ActionTerms(
+                event_id=event_id + ":payment",
+                kind="dividend_payment",
+                effective_at=pay_at,
+                available_at=available_at,
+                evidence_id=event_id,
+                **common,
+            )
+        )
+    if ratio != 1:
+        available = _date(row.shares_available_date, "shares_available_date")
+        ex_date = _date(row.ex_date, "ex_date")
+        if available != ex_date:
+            raise ValueError("deferred share delivery is not bridged without a receivables ledger")
+        terms.append(
+            ActionTerms(
+                event_id=event_id + ":split",
+                instrument_id=instrument_id,
+                kind="split",
+                effective_at=ex_at,
+                available_at=available_at,
+                currency=currency,
+                source=source,
+                evidence_id=event_id,
+                ratio=str(ratio),
+            )
+        )
+    return terms
+
+
+def explicit_terms(fields: Mapping) -> ActionTerms:
+    """Accept rights, merger and spin-off terms only when every field is supplied."""
+    if not isinstance(fields, Mapping):
+        raise TypeError("explicit corporate action must be a mapping")
+    kind = fields.get("kind")
+    if kind not in _EXPLICIT_KINDS:
+        raise ValueError(
+            "rights, merger and spin-off require an explicit evidenced record; "
+            f"supported kinds: {sorted(_EXPLICIT_KINDS)}"
+        )
+    if kind not in KINDS:
+        raise ValueError(f"unsupported action kind: {kind}")
+    return ActionTerms(**dict(fields))
diff --git a/src/quant_data_kit/providers/halt_calendar.py b/src/quant_data_kit/providers/halt_calendar.py
new file mode 100644
index 0000000..ed7f690
--- /dev/null
+++ b/src/quant_data_kit/providers/halt_calendar.py
@@ -0,0 +1,73 @@
+"""Content-addressed historical halt calendar.
+
+Sessions that are not in the supplied observations fail closed. A current
+Eastmoney ST/halt snapshot cannot be stored as this calendar.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+
+import pandas as pd
+
+_STATUSES = {"tradable", "suspended", "risk_warning", "delisted"}
+_NOT_TRADABLE = {"suspended", "delisted"}
+_SCHEMA = "quant.halt-calendar/v1"
+
+
+def freeze_halt_calendar(observations: pd.DataFrame, *, source: str) -> dict:
+    if not isinstance(source, str) or not source.strip():
+        raise ValueError("halt calendar source is required")
+    source = source.strip()
+    if source.startswith("akshare.eastmoney.current"):
+        raise ValueError("current ST/halt snapshot cannot certify a historical calendar")
+    required = {"symbol", "session", "status"}
+    if not required.issubset(observations.columns):
+        raise ValueError("halt calendar requires symbol, session and status")
+    rows = []
+    seen: set[tuple[str, str]] = set()
+    for record in observations.to_dict("records"):
+        if pd.isna(record["symbol"]):
+            raise ValueError("halt observation has an empty symbol, date or status")
+        symbol = str(record["symbol"]).strip()
+        session = pd.to_datetime(record["session"], errors="coerce")
+        status = str(record["status"])
+        if not symbol or pd.isna(session) or status not in _STATUSES:
+            raise ValueError("halt observation has an empty symbol, date or status")
+        day = pd.Timestamp(session).strftime("%Y-%m-%d")
+        key = (symbol, day)
+        if key in seen:
+            raise ValueError(f"duplicate halt observation for {symbol} on {day}")
+        seen.add(key)
+        rows.append({"symbol": symbol, "session": day, "status": status})
+    rows.sort(key=lambda item: (item["symbol"], item["session"]))
+    payload = {"schema": _SCHEMA, "source": source, "observations": rows}
+    encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    payload["sha256"] = hashlib.sha256(encoded.encode()).hexdigest()
+    return payload
+
+
+def lookup(calendar: dict, symbol: str, session) -> str:
+    if calendar.get("schema") != _SCHEMA or "sha256" not in calendar:
+        raise ValueError("halt calendar schema is not recognized")
+    payload = {key: value for key, value in calendar.items() if key != "sha256"}
+    encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    if hashlib.sha256(encoded.encode()).hexdigest() != calendar["sha256"]:
+        raise ValueError("halt calendar SHA-256 mismatch")
+    day = pd.Timestamp(session).strftime("%Y-%m-%d")
+    for row in calendar.get("observations", []):
+        if row["symbol"] == symbol and row["session"] == day:
+            if row["status"] not in _STATUSES:
+                raise ValueError("unsupported halt status")
+            return row["status"]
+    raise ValueError(f"session is not covered by the halt calendar: {symbol} {day}")
+
+
+def is_tradable(calendar: dict, symbol: str, session) -> bool:
+    status = lookup(calendar, symbol, session)
+    if status in _NOT_TRADABLE:
+        return False
+    if status in {"tradable", "risk_warning"}:
+        return True
+    raise ValueError(f"unsupported halt status: {status}")
diff --git a/tests/test_action_bridge.py b/tests/test_action_bridge.py
new file mode 100644
index 0000000..b7fcf96
--- /dev/null
+++ b/tests/test_action_bridge.py
@@ -0,0 +1,93 @@
+from decimal import Decimal
+
+import pandas as pd
+import pytest
+
+from quant_data_kit.providers.action_bridge import distribution_terms, explicit_terms
+from quant_data_kit.providers.corporate_actions import normalize_actions
+
+
+def _cash_row():
+    raw = pd.DataFrame(
+        [
+            {
+                "实施方案公告日期": "2026-07-01",
+                "股权登记日": "2026-07-09",
+                "除权日": "2026-07-10",
+                "派息日": "2026-07-12",
+                "股份到账日": None,
+                "送股比例": None,
+                "转增比例": None,
+                "派息比例": 10,
+                "实施方案分红说明": "10派10元(含税)",
+            }
+        ]
+    )
+    return normalize_actions(raw, "600036", "2026-09-19T00:00:00Z").iloc[0]
+
+
+def test_cash_distribution_becomes_entitlement_and_payment():
+    terms = distribution_terms(_cash_row(), instrument_id="600036.SH")
+    assert [item.kind for item in terms] == ["dividend_entitlement", "dividend_payment"]
+    assert terms[0].cash_per_unit == "1"
+    assert terms[0].entitlement_date == "2026-07-09"
+    assert terms[1].effective_at.startswith("2026-07-12")
+    assert all(term.available_at == "2026-09-19T00:00:00Z" for term in terms)
+
+
+def test_bonus_shares_become_a_split_and_deferred_delivery_is_refused():
+    raw = pd.DataFrame(
+        [
+            {
+                "实施方案公告日期": "2026-07-01",
+                "股权登记日": "2026-07-09",
+                "除权日": "2026-07-10",
+                "派息日": None,
+                "股份到账日": "2026-07-10",
+                "送股比例": 5,
+                "转增比例": None,
+                "派息比例": None,
+                "实施方案分红说明": "10送5股",
+            }
+        ]
+    )
+    row = normalize_actions(raw, "600036", "2026-09-19T00:00:00Z").iloc[0]
+    terms = distribution_terms(row, instrument_id="600036.SH")
+    assert len(terms) == 1 and terms[0].kind == "split"
+    assert Decimal(terms[0].ratio) == Decimal("1.5")
+    assert terms[0].available_at == "2026-09-19T00:00:00Z"
+    late = row.copy()
+    late["shares_available_date"] = pd.Timestamp("2026-07-20")
+    with pytest.raises(ValueError, match="receivables"):
+        distribution_terms(late, instrument_id="600036.SH")
+
+
+def test_rights_and_mergers_require_an_explicit_record():
+    with pytest.raises(ValueError, match="explicit"):
+        explicit_terms({"kind": "split"})
+    terms = explicit_terms(
+        {
+            "event_id": "rights-1",
+            "instrument_id": "600036.SH",
+            "kind": "rights_distribution",
+            "effective_at": "2026-07-10T01:00:00Z",
+            "available_at": "2026-07-10T01:00:00Z",
+            "currency": "CNY",
+            "source": "prospectus",
+            "evidence_id": "doc-1",
+            "ratio": "0.3",
+            "target_id": "600036.SH.RIGHT",
+            "cost_fraction": "0",
+            "target_mark": "1",
+            "parent_mark": "10",
+        }
+    )
+    assert terms.kind == "rights_distribution"
+
+
+@pytest.mark.parametrize("value", ["NaN", "Infinity", "-1"])
+def test_invalid_cash_amount_is_rejected(value):
+    row = _cash_row().copy()
+    row["cash_per_share"] = value
+    with pytest.raises(ValueError, match="distribution amounts"):
+        distribution_terms(row, instrument_id="600036.SH")
diff --git a/tests/test_halt_calendar.py b/tests/test_halt_calendar.py
new file mode 100644
index 0000000..064c5f6
--- /dev/null
+++ b/tests/test_halt_calendar.py
@@ -0,0 +1,45 @@
+import pandas as pd
+import pytest
+
+from quant_data_kit.providers.halt_calendar import freeze_halt_calendar, is_tradable, lookup
+
+
+def _frame():
+    return pd.DataFrame(
+        [
+            {"symbol": "600036", "session": "2026-07-10", "status": "suspended"},
+            {"symbol": "600036", "session": "2026-07-11", "status": "tradable"},
+            {"symbol": "000001", "session": "2026-07-10", "status": "risk_warning"},
+        ]
+    )
+
+
+def test_calendar_is_hashed_and_uncovered_sessions_fail():
+    first = freeze_halt_calendar(_frame(), source="exchange.halt.history")
+    second = freeze_halt_calendar(_frame().iloc[::-1], source="exchange.halt.history")
+    assert first["sha256"] == second["sha256"]
+    assert is_tradable(first, "600036", "2026-07-10") is False
+    assert is_tradable(first, "600036", "2026-07-11") is True
+    assert lookup(first, "000001", "2026-07-10") == "risk_warning"
+    with pytest.raises(ValueError, match="not covered"):
+        is_tradable(first, "600036", "2026-07-12")
+
+
+def test_current_snapshot_cannot_be_frozen_as_history():
+    with pytest.raises(ValueError, match="current"):
+        freeze_halt_calendar(_frame(), source="akshare.eastmoney.current_st_and_halts")
+
+
+def test_tampered_calendar_is_rejected():
+    calendar = freeze_halt_calendar(_frame(), source="exchange.halt.history")
+    calendar["observations"][1]["status"] = "tradable"
+    with pytest.raises(ValueError, match="SHA-256"):
+        lookup(calendar, "600036", "2026-07-10")
+
+
+@pytest.mark.parametrize("symbol", [None, float("nan"), " "])
+def test_missing_symbol_is_rejected(symbol):
+    frame = _frame()
+    frame.loc[0, "symbol"] = symbol
+    with pytest.raises(ValueError, match="empty symbol"):
+        freeze_halt_calendar(frame, source="exchange.halt.history")

From 72e2ee2f52964b8ec4aec44ac97f36b300289fe2 Mon Sep 17 00:00:00 2001
From: PureSaber <88314620+PureSaber@users.noreply.github.com>
Date: Tue, 29 Sep 2026 17:02:46 +0800
Subject: [PATCH 2/4] Fix Windows staging lock races and archive path round
 trips

---
 .github/workflows/ci.yml             |  6 ++++-
 README.md                            |  6 +++++
 src/quant_data_kit/data_lake.py      |  8 ++++++-
 src/quant_data_kit/normalized_v3.py  | 36 ++++++++++++++++++----------
 tests/conftest.py                    | 15 ++++++++++++
 tests/test_integrity_branch_gates.py |  3 +++
 tests/test_normalized_v3_failures.py | 15 ++++++++++++
 7 files changed, 75 insertions(+), 14 deletions(-)
 create mode 100644 tests/conftest.py

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index b2aff9d..d0f11f7 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -11,10 +11,14 @@ permissions:
 
 jobs:
   test:
-    runs-on: ubuntu-latest
+    runs-on: ${{ matrix.os }}
     strategy:
       matrix:
+        os: [ubuntu-latest]
         python-version: ["3.10", "3.11", "3.12"]
+        include:
+          - os: windows-latest
+            python-version: "3.12"
 
     steps:
       - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
diff --git a/README.md b/README.md
index 68842e2..78bccb0 100644
--- a/README.md
+++ b/README.md
@@ -1,5 +1,11 @@
 # quant-data-kit
 
+On Windows, deep lake partitions can exceed `MAX_PATH`. If system long paths are
+disabled, supply an absolute extended-length lake root (for example
+`Path(r"\\\\?\\F:\\lake")`). The pytest temporary-directory fixture uses this form
+automatically, including in Windows CI; no registry settings are changed.
+Archive file URIs preserve extended-length and UNC roots and literal percent escapes.
+
 研究可信度升级：接口、使用示例、验收及限制见 [11–20 使用说明](docs/SELECTION_LABELS.md)。
 
 Hong Kong daily research uses the explicit `quant_data_kit.hong_kong` module
diff --git a/src/quant_data_kit/data_lake.py b/src/quant_data_kit/data_lake.py
index c57477e..4dbf96a 100644
--- a/src/quant_data_kit/data_lake.py
+++ b/src/quant_data_kit/data_lake.py
@@ -1085,7 +1085,13 @@ def _local_archive_path(archive_uri: str) -> Path:
         path = direct_path
     elif archive_uri.lower().startswith("file:"):
         parsed = urlparse(archive_uri)
-        raw_path = url2pathname(unquote(parsed.path))
+        authority = unquote(parsed.netloc)
+        uri_path = parsed.path
+        if authority and authority.lower() != "localhost":
+            uri_path = "//" + authority + uri_path
+        # url2pathname already percent-decodes. A second decode would turn a
+        # literal %xx filename into a different path and discard UNC/device roots.
+        raw_path = url2pathname(uri_path)
         if os.name == "nt" and re.match(r"^/[A-Za-z]:", raw_path):
             raw_path = raw_path[1:]
         path = Path(raw_path)
diff --git a/src/quant_data_kit/normalized_v3.py b/src/quant_data_kit/normalized_v3.py
index 951133e..d2d936a 100644
--- a/src/quant_data_kit/normalized_v3.py
+++ b/src/quant_data_kit/normalized_v3.py
@@ -9,7 +9,7 @@
 import shutil
 import uuid
 from collections.abc import Iterable, Iterator, Mapping, Sequence
-from contextlib import contextmanager
+from contextlib import ExitStack, contextmanager
 from dataclasses import asdict, dataclass, field
 from datetime import date, datetime
 from functools import cache, lru_cache
@@ -526,19 +526,31 @@ def _streaming_stage(root: Path) -> Iterator[Path]:
                     owner.unlink()
             except TimeoutError:
                 continue
-    operation = uuid.uuid4().hex
-    stage = staging_root / f"normalized-batch-stream-{operation}"
-    owner = owners_root / f"{stage.name}.lock"
-    with process_file_lock(owner):
-        stage.mkdir(exist_ok=False)
+        operation = uuid.uuid4().hex
+        stage = staging_root / f"normalized-batch-stream-{operation}"
+        owner = owners_root / f"{stage.name}.lock"
+        ownership = ExitStack()
+        ownership.enter_context(process_file_lock(owner))
         try:
-            yield stage
-        finally:
-            if stage.exists():
-                shutil.rmtree(stage)
-    with process_file_lock(gc_lock):
-        if owner.exists():
+            stage.mkdir(exist_ok=False)
+        except BaseException:
+            ownership.close()
             owner.unlink()
+            raise
+    try:
+        yield stage
+    finally:
+        # Creation and retirement must be atomic with respect to the collector.
+        # Otherwise it can see an owner with no stage, open its lock file, and
+        # race unlink (WinError 32) or split lock identity on POSIX.
+        with process_file_lock(gc_lock):
+            try:
+                if stage.exists():
+                    shutil.rmtree(stage)
+            finally:
+                ownership.close()
+            if owner.exists():
+                owner.unlink()
 
 
 def _sql_text(value: str | Path) -> str:
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..a043458
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,15 @@
+"""Exercise deep lake partitions on Windows without a machine-wide registry change."""
+
+import os
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def tmp_path(tmp_path: Path) -> Path:
+    if os.name == "nt":
+        path = str(tmp_path.resolve())
+        if not path.startswith("\\\\?\\"):
+            return Path("\\\\?\\UNC\\" + path[2:] if path.startswith("\\\\") else "\\\\?\\" + path)
+    return tmp_path
diff --git a/tests/test_integrity_branch_gates.py b/tests/test_integrity_branch_gates.py
index 76703b8..51e7a1b 100644
--- a/tests/test_integrity_branch_gates.py
+++ b/tests/test_integrity_branch_gates.py
@@ -517,6 +517,9 @@ def test_raw_path_staging_and_archive_negative_guards(
     archive = tmp_path.parent / f"{tmp_path.name}-archive.bin"
     archive.write_bytes(b"archive")
     assert lake_module._local_archive_path(archive.as_uri()) == archive.resolve()
+    escaped = tmp_path.parent / f"{tmp_path.name}-literal%20archive.bin"
+    escaped.write_bytes(b"archive with a literal percent escape")
+    assert lake_module._local_archive_path(escaped.as_uri()) == escaped.resolve()
     with pytest.raises(ValidationError, match="non-empty"):
         lake_module._local_archive_path(" ")
     with pytest.raises(ValidationError, match="Remote archive cleanup"):
diff --git a/tests/test_normalized_v3_failures.py b/tests/test_normalized_v3_failures.py
index b89fa5c..e010e0f 100644
--- a/tests/test_normalized_v3_failures.py
+++ b/tests/test_normalized_v3_failures.py
@@ -174,6 +174,7 @@ def test_streaming_stage_serializes_owner_file_cleanup(
     root = lake_module._resolved_lake_root(tmp_path / "lake", create=True)
     real_lock = normalized_v3.process_file_lock
     real_unlink = Path.unlink
+    real_mkdir = Path.mkdir
     held_locks: set[Path] = set()
 
     @contextmanager
@@ -189,14 +190,28 @@ def tracked_lock(path: Path, *, timeout_seconds: float = 60.0):
     def guarded_unlink(path: Path, *args: object, **kwargs: object) -> None:
         if path.parent.name == ".stage-owners" and path.name.startswith("normalized-batch-stream-"):
             assert any(item.name == ".gc.lock" for item in held_locks)
+            assert path not in held_locks
         real_unlink(path, *args, **kwargs)
 
+    def guarded_mkdir(path: Path, *args: object, **kwargs: object) -> None:
+        if path.name.startswith("normalized-batch-stream-"):
+            assert any(item.name == ".gc.lock" for item in held_locks)
+        real_mkdir(path, *args, **kwargs)
+
     monkeypatch.setattr(normalized_v3, "process_file_lock", tracked_lock)
     monkeypatch.setattr(Path, "unlink", guarded_unlink)
+    monkeypatch.setattr(Path, "mkdir", guarded_mkdir)
     with normalized_v3._streaming_stage(root) as stage:
         assert stage.is_dir()
     owners_root = root / "normalized" / ".stage-owners"
     assert sorted(item.name for item in owners_root.iterdir()) == [".gc.lock"]
+    with (
+        pytest.raises(RuntimeError, match="producer failed"),
+        normalized_v3._streaming_stage(root),
+    ):
+        raise RuntimeError("producer failed")
+    assert sorted(item.name for item in owners_root.iterdir()) == [".gc.lock"]
+    assert not list((root / "normalized" / "staging").iterdir())
 
 
 def _rewrite_index_manifest(

From 6aac97a50231776d429d9029fde08524e5faa17d Mon Sep 17 00:00:00 2001
From: PureSaber <88314620+PureSaber@users.noreply.github.com>
Date: Tue, 29 Sep 2026 17:09:22 +0800
Subject: [PATCH 3/4] Preserve required Linux check names when adding Windows
 CI

---
 .github/workflows/ci.yml | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index d0f11f7..7ef1a89 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -11,6 +11,7 @@ permissions:
 
 jobs:
   test:
+    name: test (${{ matrix.test-label || matrix.python-version }})
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
@@ -19,6 +20,7 @@ jobs:
         include:
           - os: windows-latest
             python-version: "3.12"
+            test-label: Windows, 3.12
 
     steps:
       - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1

From 13823f2e3caa07fbd3fe4298ed16ee941e638d49 Mon Sep 17 00:00:00 2001
From: PureSaber <88314620+PureSaber@users.noreply.github.com>
Date: Tue, 29 Sep 2026 17:15:33 +0800
Subject: [PATCH 4/4] Clarify extended-length Windows path example

---
 README.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/README.md b/README.md
index 78bccb0..947a4a9 100644
--- a/README.md
+++ b/README.md
@@ -2,7 +2,7 @@
 
 On Windows, deep lake partitions can exceed `MAX_PATH`. If system long paths are
 disabled, supply an absolute extended-length lake root (for example
-`Path(r"\\\\?\\F:\\lake")`). The pytest temporary-directory fixture uses this form
+`Path("//?/F:/lake")`). The pytest temporary-directory fixture uses this form
 automatically, including in Windows CI; no registry settings are changed.
 Archive file URIs preserve extended-length and UNC roots and literal percent escapes.