fix(update): treat a timed paidAccess gate with no recorded end as active

CivitAI only returns a non-null paidAccess for an *active* gate: a lapsed gate
stays in the database as a tombstone and is filtered out server-side, so
{"permanent": false, "endsAt": null} — a timed gate whose window end has not
been recorded yet — is still enforced. Verified live: on model 1802980 that
version reports canDownload: false while its lapsed siblings report true.

Both the update service and the download gate dropped that shape, so such
versions read as free and "Hide Early Access Updates" missed them — the class
of bug reported in #1060.

The interpretation now lives in py/utils/paid_access.py and is shared, so the
badge, the update filter and the download warning cannot disagree.

py/services/download_manager.py

@@ -33,6 +33,13 @@ from ..utils.constants import (
     VALID_OTHER_CIVITAI_TYPES,
 )
 from ..utils.civitai_utils import normalize_civitai_download_url, rewrite_preview_url
+from ..utils.paid_access import (
+    is_early_access_deadline_active,
+    is_gate_active,
+    is_permanent_paid,
+    normalize_paid_access,
+    parse_civitai_timestamp,
+)
 from ..utils.file_utils import calculate_sha256, calculate_autov3
 from ..utils.preview_selection import resolve_mature_threshold, select_preview_media
 from ..utils.utils import calculate_filename_for_model, sanitize_folder_name
@@ -1985,40 +1992,32 @@ class DownloadManager:
                 os.makedirs(save_dir, exist_ok=True)
 
             # Check if this is a paid or early access model
-            paid_access = version_info.get("paidAccess")
-            if isinstance(paid_access, str):
-                # Some providers (e.g. CivArchive fallback) carry the DTO as JSON text
-                try:
-                    parsed = json.loads(paid_access)
-                    paid_access = parsed if isinstance(parsed, dict) else None
-                except (TypeError, ValueError):
-                    paid_access = None
-            if not isinstance(paid_access, dict):
-                paid_access = None
-            # An empty DTO ({"permanent": false, "endsAt": null}) is not a gate
-            if paid_access and not paid_access.get("permanent") and not paid_access.get("endsAt"):
-                paid_access = None
-            if version_info.get("earlyAccessEndsAt") or paid_access:
-                permanent_paid = bool(paid_access.get("permanent")) if paid_access else False
+            # CivitAI reports a non-null paidAccess only for an ACTIVE gate, so
+            # {"permanent": false, "endsAt": null} (a timed gate whose end is not
+            # recorded yet) still counts as gated here.
+            paid_access = normalize_paid_access(version_info.get("paidAccess"))
+            legacy_ea_ends_at = version_info.get("earlyAccessEndsAt")
+            gate_active = is_gate_active(paid_access) or is_early_access_deadline_active(
+                legacy_ea_ends_at
+            )
+            if gate_active:
+                permanent_paid = is_permanent_paid(paid_access)
                 if permanent_paid:
                     early_access_msg = (
                         "This model requires payment. Please ensure you have "
                         "purchased access and are logged in to Civitai."
                     )
                 else:
-                    early_access_date = version_info.get("earlyAccessEndsAt")
+                    early_access_date = legacy_ea_ends_at
                     if not early_access_date and paid_access:
                         early_access_date = paid_access.get("endsAt")
                     if not early_access_date:
                         early_access_date = ""
                     # Convert to a readable date if possible
                     try:
-                        from datetime import datetime
-
-                        date_obj = datetime.fromisoformat(
-                            early_access_date.replace("Z", "+00:00")
-                        )
-                        formatted_date = date_obj.strftime("%Y-%m-%d")
+                        formatted_date = parse_civitai_timestamp(
+                            early_access_date
+                        ).strftime("%Y-%m-%d")
                         early_access_msg = (
                             f"This model requires payment (until {formatted_date}). "
                         )

py/services/model_update_service.py

@@ -20,6 +20,10 @@ from .settings_manager import get_settings_manager
 from ..utils.cache_paths import CacheType, resolve_cache_path_with_migration
 from ..utils.constants import MODEL_WEIGHT_FILE_TYPES
 from ..utils.civitai_utils import rewrite_preview_url
+from ..utils.paid_access import (
+    is_early_access_deadline_active,
+    normalize_paid_access as _normalize_paid_access_payload,
+)
 from ..utils.preview_selection import resolve_mature_threshold, select_preview_media
 
 logger = logging.getLogger(__name__)
@@ -172,16 +176,10 @@ class ModelUpdateRecord:
         if version.is_paid and not version.early_access_ends_at:
             return False
 
-        # Phase 2: Precise check with exact end time
+        # Phase 2: Precise check with exact end time (None when the gate is timed
+        # but its window end has not been recorded yet -> treated as active below)
         if version.early_access_ends_at:
-            try:
-                ea_date = datetime.fromisoformat(
-                    version.early_access_ends_at.replace("Z", "+00:00")
-                )
-                return ea_date > datetime.now(timezone.utc)
-            except (ValueError, AttributeError):
-                # If date parsing fails, treat as active EA (conservative)
-                return True
+            return is_early_access_deadline_active(version.early_access_ends_at)
 
         # Phase 1: Basic EA flag from bulk API
         return version.is_early_access
@@ -1762,6 +1760,9 @@ class ModelUpdateService:
 
         # CivitAI's paidAccess DTO ({"permanent": bool, "endsAt": ISO|null})
         # gates versions behind a paid tier while availability stays "Public".
+        # A non-null DTO from the public API is always an ACTIVE gate: lapsed
+        # (tombstone) gates come back as null. That includes the timed gate whose
+        # end is not recorded yet, {"permanent": false, "endsAt": null}.
         paid_access = self._normalize_paid_access(entry.get("paidAccess"))
         paid_access_json = json.dumps(paid_access) if paid_access else None
         is_paid = bool(paid_access.get("permanent")) if paid_access else False
@@ -1769,7 +1770,7 @@ class ModelUpdateService:
             early_access_ends_at = _normalize_string(paid_access.get("endsAt"))
         # Only timed gates are early access; permanent paid versions are not
         # (consumers filter them via is_paid), so the stored flag stays accurate.
-        if not is_early_access and paid_access and paid_access.get("endsAt"):
+        if not is_early_access and paid_access and not paid_access.get("permanent"):
             is_early_access = True
 
         return ModelVersionRecord(
@@ -1797,24 +1798,12 @@ class ModelUpdateService:
         Accepts a dict, None, or a JSON string (as carried by the by-hash
         enrichment path) and returns ``{"permanent": bool, "endsAt": str|None}``
         or None when the input carries no paid-access signal.
+
+        Delegates to :mod:`py.utils.paid_access` so the update service and the
+        download gate cannot disagree about what counts as a gate.
         """
-        if value is None:
-            return None
-        if isinstance(value, str):
-            try:
-                parsed = json.loads(value)
-            except (TypeError, ValueError):
-                return None
-            if not isinstance(parsed, dict):
-                return None
-            value = parsed
-        if not isinstance(value, Mapping):
-            return None
-        permanent = bool(value.get("permanent"))
-        ends_at = _normalize_string(value.get("endsAt"))
-        if not permanent and ends_at is None:
-            return None
-        return {"permanent": permanent, "endsAt": ends_at}
+
+        return _normalize_paid_access_payload(value)
 
     @staticmethod
     def _extract_file_count(files) -> Optional[int]:

py/utils/paid_access.py

@@ -0,0 +1,158 @@
+"""Shared interpretation of CivitAI's ``paidAccess`` DTO.
+
+One rule for the whole backend: the update service (badges, update filters, price
+tracking) and the download gate both ask these helpers, so a version can never be
+"paid" for one consumer and "free" for another.
+
+Why the rules look like this
+----------------------------
+CivitAI's public v1 API only returns a non-null DTO for an *active* gate: a lapsed
+gate stays in the database as a tombstone but is filtered out server-side
+(``toPublicPaidAccessDto`` in civitai's ``server/services/paid-access.service.ts``).
+A timed gate whose end time has not been recorded yet is reported as
+``{"permanent": false, "endsAt": null}`` and is still enforced — such a version
+reports ``canDownload: false`` on the model page — so it must NOT be discarded.
+Dropping it (the previous behaviour) hid real gates, which is the class of bug
+reported in issue #1060.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from typing import Any, Mapping, Optional
+
+__all__ = [
+    "normalize_paid_access",
+    "is_permanent_paid",
+    "is_gate_active",
+    "is_early_access_deadline_active",
+    "parse_civitai_timestamp",
+]
+
+# A DTO that carries neither key is not a gate signal at all (defensive: CivitAI
+# never emits a bare ``{}``, but other metadata sources might).
+_PAID_ACCESS_KEYS = ("permanent", "endsAt")
+
+
+def parse_civitai_timestamp(value: Any) -> Optional[datetime]:
+    """Parse a CivitAI ISO-8601 timestamp into an aware UTC datetime.
+
+    Returns None for anything that is not a parsable string. Naive timestamps are
+    assumed to be UTC, matching CivitAI's serialization.
+    """
+
+    if not isinstance(value, str):
+        return None
+    text = value.strip()
+    if not text:
+        return None
+    if text.endswith("Z"):
+        text = f"{text[:-1]}+00:00"
+    try:
+        parsed = datetime.fromisoformat(text)
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        parsed = parsed.replace(tzinfo=timezone.utc)
+    return parsed
+
+
+def normalize_paid_access(value: Any) -> Optional[dict]:
+    """Normalize a ``paidAccess`` value into ``{"permanent": bool, "endsAt": str|None}``.
+
+    Accepts a mapping, a JSON string (the by-hash enrichment path carries the DTO as
+    text), or None. Returns None when the value carries no gate signal.
+
+    Note that ``{"permanent": False, "endsAt": None}`` is a *gate*: CivitAI reports
+    it for a timed gate whose window end is not set yet, and enforces it.
+    """
+
+    if value is None:
+        return None
+
+    if isinstance(value, str):
+        try:
+            parsed = json.loads(value)
+        except (TypeError, ValueError):
+            return None
+        if not isinstance(parsed, dict):
+            return None
+        value = parsed
+
+    if not isinstance(value, Mapping):
+        return None
+
+    if not any(key in value for key in _PAID_ACCESS_KEYS):
+        return None
+
+    ends_at = value.get("endsAt")
+    normalized_ends_at = ends_at.strip() if isinstance(ends_at, str) and ends_at.strip() else None
+    return {
+        "permanent": bool(value.get("permanent")),
+        "endsAt": normalized_ends_at,
+    }
+
+
+def is_permanent_paid(paid_access: Optional[Mapping[str, Any]]) -> bool:
+    """True when the gate never expires (a permanent paid version)."""
+
+    return bool(paid_access and paid_access.get("permanent"))
+
+
+def is_gate_active(
+    paid_access: Optional[Mapping[str, Any]],
+    *,
+    now: Optional[datetime] = None,
+) -> bool:
+    """True when a normalized ``paidAccess`` gate is currently in force.
+
+    Active means permanent, or a timed gate whose end is either still in the future
+    or not recorded yet (CivitAI enforces the latter too). An unparsable end time is
+    treated as active rather than free: the download would fail anyway, so the
+    conservative reading matches what the user will experience.
+    """
+
+    if not paid_access:
+        return False
+    if paid_access.get("permanent"):
+        return True
+
+    ends_at = paid_access.get("endsAt")
+    if not ends_at:
+        # Timed gate with no recorded end — CivitAI still gates the download.
+        return True
+
+    parsed = parse_civitai_timestamp(ends_at)
+    if parsed is None:
+        return True
+
+    reference = now or datetime.now(timezone.utc)
+    if reference.tzinfo is None:
+        reference = reference.replace(tzinfo=timezone.utc)
+    return parsed > reference
+
+
+def is_early_access_deadline_active(
+    ends_at: Any,
+    *,
+    now: Optional[datetime] = None,
+) -> bool:
+    """True when a legacy ``earlyAccessEndsAt`` deadline is still in the future.
+
+    Kept separate from :func:`is_gate_active` because the legacy field is a bare
+    timestamp rather than a DTO. A present-but-unparsable value is treated as active,
+    matching the previous conservative behaviour.
+    """
+
+    if not ends_at:
+        return False
+
+    parsed = parse_civitai_timestamp(ends_at)
+    if parsed is None:
+        return True
+
+    reference = now or datetime.now(timezone.utc)
+    if reference.tzinfo is None:
+        reference = reference.replace(tzinfo=timezone.utc)
+    return parsed > reference

tests/services/test_model_update_service.py

@@ -826,6 +826,32 @@ def test_extract_single_version_paid_access_permanent(tmp_path):
     assert version.paid_access is not None
 
 
+def test_extract_single_version_paid_access_pending_end(tmp_path):
+    """A timed gate whose window end is not recorded yet
+    ({"permanent": false, "endsAt": null}) is still an active gate, so it is
+    early access with no known end date rather than a free version."""
+    db_path = tmp_path / "updates.sqlite"
+    service = ModelUpdateService(str(db_path))
+
+    entry = {
+        "id": 42,
+        "name": "v1 paid",
+        "availability": "Public",
+        "paidAccess": {"permanent": False, "endsAt": None},
+        "files": [],
+        "images": [],
+    }
+
+    version = service._extract_single_version(entry, index=0)
+
+    assert version is not None
+    assert version.is_early_access is True
+    assert version.is_paid is False
+    assert version.early_access_ends_at is None
+    assert version.paid_access == '{"permanent": false, "endsAt": null}'
+    assert ModelUpdateRecord._is_early_access_active(version) is True
+
+
 def test_normalize_paid_access_accepts_json_string():
     """The by-hash enrichment path may hand paidAccess to _normalize_paid_access
     as a JSON string; both the permanent and timed shapes must normalize."""
@@ -841,14 +867,21 @@ def test_normalize_paid_access_accepts_json_string():
     )
     assert timed == {"permanent": False, "endsAt": "2026-08-22T18:30:00.000Z"}
 
-    empty = ModelUpdateService._normalize_paid_access(
+    # A timed gate whose window end is not recorded yet. CivitAI only returns a
+    # non-null DTO for an ACTIVE gate (tombstones come back as null) and enforces
+    # this shape too - the model page reports canDownload: false for it - so it
+    # must be kept. Dropping it was the #1060 class of bug.
+    pending_end = ModelUpdateService._normalize_paid_access(
         '{"permanent": false, "endsAt": null}'
     )
-    assert empty is None
+    assert pending_end == {"permanent": False, "endsAt": None}
 
     malformed = ModelUpdateService._normalize_paid_access("{not json")
     assert malformed is None
 
+    # No gate keys at all is not a gate signal.
+    assert ModelUpdateService._normalize_paid_access("{}") is None
+
 
 def test_has_update_for_base_hide_paid():
     """hide_paid also suppresses permanent paid versions in the same-base

tests/utils/test_paid_access.py

@@ -0,0 +1,112 @@
+"""Tests for the shared CivitAI paidAccess interpretation.
+
+These rules decide both the Paid/Early Access badges and whether a download is
+warned about, so the interesting cases are the boundary shapes CivitAI actually
+emits.
+"""
+
+from datetime import datetime, timezone
+
+import pytest
+
+from py.utils.paid_access import (
+    is_early_access_deadline_active,
+    is_gate_active,
+    is_permanent_paid,
+    normalize_paid_access,
+    parse_civitai_timestamp,
+)
+
+NOW = datetime(2026, 9, 30, 12, 0, 0, tzinfo=timezone.utc)
+
+
+@pytest.mark.parametrize(
+    "value,expected",
+    [
+        (None, None),
+        ("", None),
+        ("{not json", None),
+        ("[1, 2]", None),
+        ({}, None),
+        ({"unknown": 1}, None),
+        # A non-null DTO from CivitAI's public API is always an active gate, so a
+        # timed gate with no recorded end must survive normalization.
+        ({"permanent": False, "endsAt": None}, {"permanent": False, "endsAt": None}),
+        ({"permanent": True, "endsAt": None}, {"permanent": True, "endsAt": None}),
+        (
+            {"permanent": False, "endsAt": "2026-10-10T13:10:17.404Z"},
+            {"permanent": False, "endsAt": "2026-10-10T13:10:17.404Z"},
+        ),
+    ],
+)
+def test_normalize_paid_access_shapes(value, expected):
+    assert normalize_paid_access(value) == expected
+
+
+def test_normalize_paid_access_accepts_json_text():
+    """The by-hash enrichment path can carry the DTO as a JSON string."""
+
+    assert normalize_paid_access('{"permanent": true, "endsAt": null}') == {
+        "permanent": True,
+        "endsAt": None,
+    }
+
+
+def test_normalize_paid_access_blank_ends_at_is_none():
+    assert normalize_paid_access({"permanent": False, "endsAt": "  "}) == {
+        "permanent": False,
+        "endsAt": None,
+    }
+
+
+def test_is_permanent_paid():
+    assert is_permanent_paid({"permanent": True, "endsAt": None}) is True
+    assert is_permanent_paid({"permanent": False, "endsAt": None}) is False
+    assert is_permanent_paid(None) is False
+
+
+@pytest.mark.parametrize(
+    "paid_access,expected",
+    [
+        (None, False),
+        ({"permanent": False, "endsAt": None}, True),
+        ({"permanent": True, "endsAt": None}, True),
+        # A permanent gate with a stale endsAt stays active.
+        ({"permanent": True, "endsAt": "2020-01-01T00:00:00.000Z"}, True),
+        ({"permanent": False, "endsAt": "2026-10-10T13:10:17.404Z"}, True),
+        ({"permanent": False, "endsAt": "2026-08-27T16:56:57.438Z"}, False),
+        ({"permanent": False, "endsAt": "not-a-date"}, True),
+    ],
+)
+def test_is_gate_active(paid_access, expected):
+    assert is_gate_active(paid_access, now=NOW) is expected
+
+
+@pytest.mark.parametrize(
+    "ends_at,expected",
+    [
+        (None, False),
+        ("", False),
+        ("2026-10-10T13:10:17.404Z", True),
+        ("2026-08-27T16:56:57.438Z", False),
+        ("garbage", True),
+    ],
+)
+def test_is_early_access_deadline_active(ends_at, expected):
+    assert is_early_access_deadline_active(ends_at, now=NOW) is expected
+
+
+@pytest.mark.parametrize(
+    "value,expected",
+    [
+        (None, None),
+        (123, None),
+        ("", None),
+        ("garbage", None),
+        ("2026-10-10T13:10:17.404Z", datetime(2026, 10, 10, 13, 10, 17, 404000, tzinfo=timezone.utc)),
+        # Naive timestamps are assumed UTC.
+        ("2026-10-10T13:10:17", datetime(2026, 10, 10, 13, 10, 17, tzinfo=timezone.utc)),
+    ],
+)
+def test_parse_civitai_timestamp(value, expected):
+    assert parse_civitai_timestamp(value) == expected