state

napt.state.cache

Discovery cache persistence for NAPT.

This module implements the discovery cache: a disposable optimization file (default cache/discovery.json) that tracks discovered versions, ETags, and download metadata between runs. Deleting it costs one full re-download per app and nothing else — the filesystem and deployment state remain the source of truth.

The cache supports two optimization approaches:

• VERSION-FIRST (url_pattern, api_github, api_json): Uses known_version for comparison
• FILE-FIRST (url_download): Uses etag/last_modified for HTTP conditional requests

Key Features:

• JSON-based cache storage (fast parsing, standard library)
• Automatic ETag/Last-Modified tracking for conditional requests
• Version change detection for version-first strategies
• Robust error handling (corrupted files, missing data)
• Auto-creation of cache files and directories

Example

High-level API with DiscoveryCache:

from pathlib import Path
from napt.state import DiscoveryCache

cache = DiscoveryCache(Path("cache/discovery.json"))
cache.load()

# Get entry for conditional requests
entry = cache.get_cache("napt-chrome")

# Update after discovery
cache.update_cache("napt-chrome", version="130.0.0", ...)
cache.save()

Low-level API with functions:

from pathlib import Path
from napt.state import load_cache, save_cache

data = load_cache(Path("cache/discovery.json"))
# ... modify cache dict ...
save_cache(data, Path("cache/discovery.json"))

DiscoveryCache

Manages the discovery cache with automatic persistence.

This class provides a high-level interface for loading, querying, and updating the cache file. It handles file I/O, error recovery, and provides convenience methods for common operations.

Attributes:

Name	Type	Description
cache_file		Path to the JSON cache file.
data	dict[str, Any]	In-memory cache dictionary.

Example

Basic usage:

from pathlib import Path

cache = DiscoveryCache(Path("cache/discovery.json"))
cache.load()
entry = cache.get_cache("napt-chrome")
cache.update_cache(
    "napt-chrome",
    url="https://...",
    sha256="...",
    known_version="130.0.0"
)
cache.save()

Source code in napt/state/cache.py

class DiscoveryCache:
    """Manages the discovery cache with automatic persistence.

    This class provides a high-level interface for loading, querying, and
    updating the cache file. It handles file I/O, error recovery, and
    provides convenience methods for common operations.

    Attributes:
        cache_file: Path to the JSON cache file.
        data: In-memory cache dictionary.

    Example:
        Basic usage:
            ```python
            from pathlib import Path

            cache = DiscoveryCache(Path("cache/discovery.json"))
            cache.load()
            entry = cache.get_cache("napt-chrome")
            cache.update_cache(
                "napt-chrome",
                url="https://...",
                sha256="...",
                known_version="130.0.0"
            )
            cache.save()
            ```

    """

    def __init__(self, cache_file: Path):
        """Initialize discovery cache.

        Args:
            cache_file: Path to JSON cache file. Created if doesn't exist.

        """
        self.cache_file = cache_file
        self.data: dict[str, Any] = {}

    def load(self) -> dict[str, Any]:
        """Load cache from file.

        Creates default cache structure if file doesn't exist.
        Handles corrupted files by creating backup and starting fresh.

        Returns:
            Loaded cache dictionary.

        Raises:
            OSError: If file permissions prevent reading.

        """
        try:
            self.data = load_cache(self.cache_file)
        except FileNotFoundError:
            # First run, create default cache
            self.data = create_default_cache()
            self.cache_file.parent.mkdir(parents=True, exist_ok=True)
            self.save()
        except json.JSONDecodeError as err:
            # Corrupted file, backup and create new
            backup = self.cache_file.with_suffix(".json.backup")
            self.cache_file.rename(backup)
            self.data = create_default_cache()
            self.save()
            raise StateError(
                f"Corrupted cache file backed up to {backup}. "
                f"Created fresh cache file."
            ) from err

        return self.data

    def save(self) -> None:
        """Save current cache to file.

        Updates metadata.last_updated timestamp automatically.
        Creates parent directories if needed.

        Raises:
            OSError: If file permissions prevent writing.

        """
        # Update metadata
        self.data.setdefault("metadata", {})
        self.data["metadata"]["last_updated"] = datetime.now(UTC).isoformat()

        # Ensure parent directory exists
        self.cache_file.parent.mkdir(parents=True, exist_ok=True)

        save_cache(self.data, self.cache_file)

    def get_cache(self, recipe_id: str) -> dict[str, Any] | None:
        """Get cached information for a recipe.

        Args:
            recipe_id: Recipe identifier (from recipe's 'id' field).

        Returns:
            Cached data if available, None otherwise.

        Example:
            Retrieve cached information:
                ```python
                entry = cache.get_cache("napt-chrome")
                if entry:
                    etag = entry.get('etag')
                    known_version = entry.get('known_version')
                ```

        """
        return self.data.get("apps", {}).get(recipe_id)

    def update_cache(
        self,
        recipe_id: str,
        url: str,
        sha256: str,
        etag: str | None = None,
        last_modified: str | None = None,
        known_version: str | None = None,
        strategy: str | None = None,
    ) -> None:
        """Update cached information for a recipe.

        Args:
            recipe_id: Recipe identifier.
            url: Download URL for provenance tracking. For version-first strategies
                (url_pattern, api_github, api_json), this is the actual download URL
                from version_info. For file-first (url_download), this is discovery.url.
            sha256: SHA-256 hash of file (for integrity checks).
            etag: ETag header from download response. Used by url_download for HTTP 304
                conditional requests. Saved but unused by version-first strategies.
            last_modified: Last-Modified header from download response.
                Used by url_download as fallback for conditional requests.
                Saved but unused by version-first.
            known_version: Version string. PRIMARY cache key for
                version-first strategies (compared to skip downloads).
                Informational only for url_download.
            strategy: Discovery strategy used (for debugging).

        Example:
            Update cache entry:
                ```python
                cache.update_cache(
                    "napt-chrome",
                    url="https://dl.google.com/chrome.msi",
                    sha256="abc123...",
                    etag='W/"def456"',
                    known_version="130.0.0"
                )
                ```

        Note:
            Schema v2: Removed file_path, last_checked, and renamed
            version -> known_version.

            Field usage differs by strategy type:

            - Version-first: known_version is PRIMARY cache key,
                etag/last_modified unused
            - File-first: etag/last_modified are PRIMARY cache keys,
                known_version informational

            The cache is for optimization only; the filesystem and
            deployment state are the source of truth.

        """
        if "apps" not in self.data:
            self.data["apps"] = {}

        cache_entry = {
            "url": url,
            "etag": etag,
            "last_modified": last_modified,
            "sha256": sha256,
        }

        # Optional fields (only add if provided)
        if known_version is not None:
            cache_entry["known_version"] = known_version
        if strategy is not None:
            cache_entry["strategy"] = strategy

        self.data["apps"][recipe_id] = cache_entry

    def has_version_changed(self, recipe_id: str, new_version: str) -> bool:
        """Check if discovered version differs from cached known_version.

        Args:
            recipe_id: Recipe identifier.
            new_version: Newly discovered version.

        Returns:
            True if version changed or no cached version exists.

        Example:
            Check if version has changed:
                ```python
                if cache.has_version_changed("napt-chrome", "130.0.0"):
                    print("New version available!")
                ```

        Note:
            Uses 'known_version' field which is informational only.
            Real version should be extracted from filesystem during build.

        """
        entry = self.get_cache(recipe_id)
        if not entry:
            return True  # No cache, treat as changed

        return entry.get("known_version") != new_version

__init__

__init__(cache_file: Path)

Initialize discovery cache.

Parameters:

Name	Type	Description	Default
cache_file	Path	Path to JSON cache file. Created if doesn't exist.	required

Source code in napt/state/cache.py

def __init__(self, cache_file: Path):
    """Initialize discovery cache.

    Args:
        cache_file: Path to JSON cache file. Created if doesn't exist.

    """
    self.cache_file = cache_file
    self.data: dict[str, Any] = {}

load

load() -> dict[str, Any]

Load cache from file.

Creates default cache structure if file doesn't exist. Handles corrupted files by creating backup and starting fresh.

Returns:

Type	Description
dict[str, Any]	Loaded cache dictionary.

Raises:

Type	Description
OSError	If file permissions prevent reading.

Source code in napt/state/cache.py

def load(self) -> dict[str, Any]:
    """Load cache from file.

    Creates default cache structure if file doesn't exist.
    Handles corrupted files by creating backup and starting fresh.

    Returns:
        Loaded cache dictionary.

    Raises:
        OSError: If file permissions prevent reading.

    """
    try:
        self.data = load_cache(self.cache_file)
    except FileNotFoundError:
        # First run, create default cache
        self.data = create_default_cache()
        self.cache_file.parent.mkdir(parents=True, exist_ok=True)
        self.save()
    except json.JSONDecodeError as err:
        # Corrupted file, backup and create new
        backup = self.cache_file.with_suffix(".json.backup")
        self.cache_file.rename(backup)
        self.data = create_default_cache()
        self.save()
        raise StateError(
            f"Corrupted cache file backed up to {backup}. "
            f"Created fresh cache file."
        ) from err

    return self.data

save

save() -> None

Save current cache to file.

Updates metadata.last_updated timestamp automatically. Creates parent directories if needed.

Raises:

Type	Description
OSError	If file permissions prevent writing.

Source code in napt/state/cache.py

def save(self) -> None:
    """Save current cache to file.

    Updates metadata.last_updated timestamp automatically.
    Creates parent directories if needed.

    Raises:
        OSError: If file permissions prevent writing.

    """
    # Update metadata
    self.data.setdefault("metadata", {})
    self.data["metadata"]["last_updated"] = datetime.now(UTC).isoformat()

    # Ensure parent directory exists
    self.cache_file.parent.mkdir(parents=True, exist_ok=True)

    save_cache(self.data, self.cache_file)

get_cache

get_cache(recipe_id: str) -> dict[str, Any] | None

Get cached information for a recipe.

Parameters:

Name	Type	Description	Default
recipe_id	str	Recipe identifier (from recipe's 'id' field).	required

Returns:

Type	Description
dict[str, Any] | None	Cached data if available, None otherwise.

Example

Retrieve cached information:

entry = cache.get_cache("napt-chrome")
if entry:
    etag = entry.get('etag')
    known_version = entry.get('known_version')

Source code in napt/state/cache.py

def get_cache(self, recipe_id: str) -> dict[str, Any] | None:
    """Get cached information for a recipe.

    Args:
        recipe_id: Recipe identifier (from recipe's 'id' field).

    Returns:
        Cached data if available, None otherwise.

    Example:
        Retrieve cached information:
            ```python
            entry = cache.get_cache("napt-chrome")
            if entry:
                etag = entry.get('etag')
                known_version = entry.get('known_version')
            ```

    """
    return self.data.get("apps", {}).get(recipe_id)

update_cache

update_cache(
    recipe_id: str,
    url: str,
    sha256: str,
    etag: str | None = None,
    last_modified: str | None = None,
    known_version: str | None = None,
    strategy: str | None = None,
) -> None

Update cached information for a recipe.

Parameters:

Name	Type	Description	Default
recipe_id	str	Recipe identifier.	required
url	str	Download URL for provenance tracking. For version-first strategies (url_pattern, api_github, api_json), this is the actual download URL from version_info. For file-first (url_download), this is discovery.url.	required
sha256	str	SHA-256 hash of file (for integrity checks).	required
etag	str | None	ETag header from download response. Used by url_download for HTTP 304 conditional requests. Saved but unused by version-first strategies.	None
last_modified	str | None	Last-Modified header from download response. Used by url_download as fallback for conditional requests. Saved but unused by version-first.	None
known_version	str | None	Version string. PRIMARY cache key for version-first strategies (compared to skip downloads). Informational only for url_download.	None
strategy	str | None	Discovery strategy used (for debugging).	None

Example

Update cache entry:

cache.update_cache(
    "napt-chrome",
    url="https://dl.google.com/chrome.msi",
    sha256="abc123...",
    etag='W/"def456"',
    known_version="130.0.0"
)

Note

Schema v2: Removed file_path, last_checked, and renamed version -> known_version.

Field usage differs by strategy type:

• Version-first: known_version is PRIMARY cache key, etag/last_modified unused
• File-first: etag/last_modified are PRIMARY cache keys, known_version informational

The cache is for optimization only; the filesystem and deployment state are the source of truth.

Source code in napt/state/cache.py

def update_cache(
    self,
    recipe_id: str,
    url: str,
    sha256: str,
    etag: str | None = None,
    last_modified: str | None = None,
    known_version: str | None = None,
    strategy: str | None = None,
) -> None:
    """Update cached information for a recipe.

    Args:
        recipe_id: Recipe identifier.
        url: Download URL for provenance tracking. For version-first strategies
            (url_pattern, api_github, api_json), this is the actual download URL
            from version_info. For file-first (url_download), this is discovery.url.
        sha256: SHA-256 hash of file (for integrity checks).
        etag: ETag header from download response. Used by url_download for HTTP 304
            conditional requests. Saved but unused by version-first strategies.
        last_modified: Last-Modified header from download response.
            Used by url_download as fallback for conditional requests.
            Saved but unused by version-first.
        known_version: Version string. PRIMARY cache key for
            version-first strategies (compared to skip downloads).
            Informational only for url_download.
        strategy: Discovery strategy used (for debugging).

    Example:
        Update cache entry:
            ```python
            cache.update_cache(
                "napt-chrome",
                url="https://dl.google.com/chrome.msi",
                sha256="abc123...",
                etag='W/"def456"',
                known_version="130.0.0"
            )
            ```

    Note:
        Schema v2: Removed file_path, last_checked, and renamed
        version -> known_version.

        Field usage differs by strategy type:

        - Version-first: known_version is PRIMARY cache key,
            etag/last_modified unused
        - File-first: etag/last_modified are PRIMARY cache keys,
            known_version informational

        The cache is for optimization only; the filesystem and
        deployment state are the source of truth.

    """
    if "apps" not in self.data:
        self.data["apps"] = {}

    cache_entry = {
        "url": url,
        "etag": etag,
        "last_modified": last_modified,
        "sha256": sha256,
    }

    # Optional fields (only add if provided)
    if known_version is not None:
        cache_entry["known_version"] = known_version
    if strategy is not None:
        cache_entry["strategy"] = strategy

    self.data["apps"][recipe_id] = cache_entry

has_version_changed

has_version_changed(recipe_id: str, new_version: str) -> bool

Check if discovered version differs from cached known_version.

Parameters:

Name	Type	Description	Default
recipe_id	str	Recipe identifier.	required
new_version	str	Newly discovered version.	required

Returns:

Type	Description
bool	True if version changed or no cached version exists.

Example

Check if version has changed:

if cache.has_version_changed("napt-chrome", "130.0.0"):
    print("New version available!")

Note

Uses 'known_version' field which is informational only. Real version should be extracted from filesystem during build.

Source code in napt/state/cache.py

def has_version_changed(self, recipe_id: str, new_version: str) -> bool:
    """Check if discovered version differs from cached known_version.

    Args:
        recipe_id: Recipe identifier.
        new_version: Newly discovered version.

    Returns:
        True if version changed or no cached version exists.

    Example:
        Check if version has changed:
            ```python
            if cache.has_version_changed("napt-chrome", "130.0.0"):
                print("New version available!")
            ```

    Note:
        Uses 'known_version' field which is informational only.
        Real version should be extracted from filesystem during build.

    """
    entry = self.get_cache(recipe_id)
    if not entry:
        return True  # No cache, treat as changed

    return entry.get("known_version") != new_version

cache_file_path

cache_file_path(config: dict[str, Any]) -> Path

Returns the discovery cache file path from merged configuration.

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	Merged configuration containing directories.cache.	required

Returns:

Type	Description
Path	Path to the discovery cache file (<directories.cache>/discovery.json).

Source code in napt/state/cache.py

def cache_file_path(config: dict[str, Any]) -> Path:
    """Returns the discovery cache file path from merged configuration.

    Args:
        config: Merged configuration containing ``directories.cache``.

    Returns:
        Path to the discovery cache file (``<directories.cache>/discovery.json``).

    """
    return Path(config["directories"]["cache"]) / "discovery.json"

create_default_cache

create_default_cache() -> dict[str, Any]

Create a default empty cache structure.

Returns:

Type	Description
dict[str, Any]	Empty cache with metadata section.

Example

Create default cache structure:

data = create_default_cache()
data["apps"] = {}

Source code in napt/state/cache.py

def create_default_cache() -> dict[str, Any]:
    """Create a default empty cache structure.

    Returns:
        Empty cache with metadata section.

    Example:
        Create default cache structure:
            ```python
            data = create_default_cache()
            data["apps"] = {}
            ```

    """
    return {
        "metadata": {
            "napt_version": __version__,
            "schema_version": "2",
            "last_updated": datetime.now(UTC).isoformat(),
        },
        "apps": {},
    }

load_cache

load_cache(cache_file: Path) -> dict[str, Any]

Load cache from JSON file.

Parameters:

Name	Type	Description	Default
cache_file	Path	Path to JSON cache file.	required

Returns:

Type	Description
dict[str, Any]	Loaded cache dictionary.

Raises:

Type	Description
FileNotFoundError	If cache file doesn't exist.
JSONDecodeError	If file contains invalid JSON.
OSError	If file cannot be read due to permissions.

Example

Load cache from file:

from pathlib import Path

data = load_cache(Path("cache/discovery.json"))
apps = data.get("apps", {})

Source code in napt/state/cache.py

def load_cache(cache_file: Path) -> dict[str, Any]:
    """Load cache from JSON file.

    Args:
        cache_file: Path to JSON cache file.

    Returns:
        Loaded cache dictionary.

    Raises:
        FileNotFoundError: If cache file doesn't exist.
        json.JSONDecodeError: If file contains invalid JSON.
        OSError: If file cannot be read due to permissions.

    Example:
        Load cache from file:
            ```python
            from pathlib import Path

            data = load_cache(Path("cache/discovery.json"))
            apps = data.get("apps", {})
            ```

    """
    with open(cache_file, encoding="utf-8") as f:
        return json.load(f)

save_cache

save_cache(data: dict[str, Any], cache_file: Path) -> None

Save cache to JSON file with pretty-printing.

Creates parent directories if needed. Uses 2-space indentation and sorted keys for consistent diffs in version control.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Cache dictionary to save.	required
cache_file	Path	Path to JSON cache file.	required

Raises:

Type	Description
OSError	If file cannot be written due to permissions.

Example

Save cache to file:

from pathlib import Path

data = {"metadata": {}, "apps": {}}
save_cache(data, Path("cache/discovery.json"))

Note

• Uses 2-space indentation for readability
• Sorts keys alphabetically for consistent diffs
• Adds trailing newline for git compatibility

Source code in napt/state/cache.py

def save_cache(data: dict[str, Any], cache_file: Path) -> None:
    """Save cache to JSON file with pretty-printing.

    Creates parent directories if needed. Uses 2-space indentation
    and sorted keys for consistent diffs in version control.

    Args:
        data: Cache dictionary to save.
        cache_file: Path to JSON cache file.

    Raises:
        OSError: If file cannot be written due to permissions.

    Example:
        Save cache to file:
            ```python
            from pathlib import Path

            data = {"metadata": {}, "apps": {}}
            save_cache(data, Path("cache/discovery.json"))
            ```

    Note:
        - Uses 2-space indentation for readability
        - Sorts keys alphabetically for consistent diffs
        - Adds trailing newline for git compatibility

    """
    cache_file.parent.mkdir(parents=True, exist_ok=True)

    with open(cache_file, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, sort_keys=True)
        f.write("\n")  # Trailing newline for git

napt.state.deployment

Deployment state persistence for NAPT.

This module implements per-app deployment state: authoritative records of what NAPT has published to Intune and what is awaiting publication. Unlike the discovery cache, deployment state is not regenerable.

Each app gets its own file, state/deployment/<recipe-id>.json, so that concurrent changes to different apps never conflict and each file's diff is scoped to one app. A file names its app once at the top (app_id matching the filename, plus the recipe's display name, refreshed on every save) and holds five sections:

• published: The release currently in Intune, with its SHA-256 hash and Intune app IDs. Null until the first upload. Publishing uploads the release without assigning it — napt promote deploys it through the rings afterwards.
• install_assigned: The release the install entry is currently assigned to (the result of a promote plan's assign action).
• pending: The discovered release awaiting publication, with version, download URL, and SHA-256 hash. A single slot — a newer discovery replaces an unpublished candidate (newest wins). Null when nothing is awaiting publication.
• rings: Which version currently holds each deployment ring. Written by napt promote.
• retained: Displaced versions kept in Intune for rollback.

Serialization is deterministic (fixed reading-order keys, fixed indentation, no timestamps), so re-running a command that produces no logical change produces a byte-identical file and a clean git diff. Keys follow reading order — lifecycle order at the top level, version first and hashes last inside blocks — because these files are what a publish PR diff shows its reviewer.

Example

Recording a discovered release:

from pathlib import Path
from napt.state import (
    deployment_state_path,
    load_deployment_state,
    record_pending,
    save_deployment_state,
)

path = deployment_state_path(Path("state/deployment"), "napt-chrome")
state = load_deployment_state(path)
action = record_pending(
    state,
    version="130.0.0",
    sha256="abc123...",
    url="https://dl.google.com/chrome.msi",
)
if action:
    save_deployment_state(state, path)

deployment_state_path

deployment_state_path(state_dir: Path, recipe_id: str) -> Path

Returns the deployment state file path for a recipe.

Parameters:

Name	Type	Description	Default
state_dir	Path	Directory holding per-app deployment state files (typically state/deployment).	required
recipe_id	str	Recipe identifier (from recipe's 'id' field).	required

Returns:

Type	Description
Path	Path to the app's deployment state file.

Source code in napt/state/deployment.py

def deployment_state_path(state_dir: Path, recipe_id: str) -> Path:
    """Returns the deployment state file path for a recipe.

    Args:
        state_dir: Directory holding per-app deployment state files
            (typically ``state/deployment``).
        recipe_id: Recipe identifier (from recipe's 'id' field).

    Returns:
        Path to the app's deployment state file.

    """
    return state_dir / f"{recipe_id}.json"

create_default_deployment_state

create_default_deployment_state() -> dict[str, Any]

Creates an empty deployment state structure.

The identity fields (app_id, name) are stamped at save time — app_id from the filename, name by whichever writer holds the recipe configuration.

Returns:

Type	Description
dict[str, Any]	Deployment state with no published release, no pending release,
dict[str, Any]	no ring assignments, and no retained versions.

Source code in napt/state/deployment.py

def create_default_deployment_state() -> dict[str, Any]:
    """Creates an empty deployment state structure.

    The identity fields (``app_id``, ``name``) are stamped at save time —
    ``app_id`` from the filename, ``name`` by whichever writer holds the
    recipe configuration.

    Returns:
        Deployment state with no published release, no pending release,
        no ring assignments, and no retained versions.

    """
    return {
        "schemaVersion": DEPLOYMENT_STATE_SCHEMA_VERSION,
        "published": None,
        "pending": None,
        "rings": {},
        "retained": [],
    }

load_deployment_state

load_deployment_state(state_path: Path) -> dict[str, Any]

Loads deployment state for one app.

Returns a default empty structure when the file does not exist. Does not create the file — deployment state is only written when there is something to record.

Parameters:

Name	Type	Description	Default
state_path	Path	Path to the app's deployment state file.	required

Returns:

Type	Description
dict[str, Any]	Deployment state dictionary.

Raises:

Type	Description
StateError	If the file exists but contains invalid JSON, its schemaVersion is missing or unsupported, or its declared app_id disagrees with the filename. Deployment state is authoritative, so a corrupted file is never silently replaced.

Source code in napt/state/deployment.py

def load_deployment_state(state_path: Path) -> dict[str, Any]:
    """Loads deployment state for one app.

    Returns a default empty structure when the file does not exist. Does
    not create the file — deployment state is only written when there is
    something to record.

    Args:
        state_path: Path to the app's deployment state file.

    Returns:
        Deployment state dictionary.

    Raises:
        StateError: If the file exists but contains invalid JSON, its
            schemaVersion is missing or unsupported, or its declared
            app_id disagrees with the filename. Deployment state is
            authoritative, so a corrupted file is never silently
            replaced.

    """
    try:
        with open(state_path, encoding="utf-8") as f:
            state = json.load(f)
    except FileNotFoundError:
        return create_default_deployment_state()
    except json.JSONDecodeError as err:
        raise StateError(
            f"Corrupted deployment state file: {state_path}. "
            "Deployment state is authoritative and is not auto-replaced. "
            "Fix the JSON or restore the file from a backup."
        ) from err

    found = state.get("schemaVersion")
    if found != DEPLOYMENT_STATE_SCHEMA_VERSION:
        raise StateError(
            f"Unsupported deployment state schema version {found!r} in "
            f"{state_path} (this NAPT release supports version "
            f"{DEPLOYMENT_STATE_SCHEMA_VERSION})."
        )

    declared = state.get("app_id")
    if declared is not None and declared != state_path.stem:
        raise StateError(
            f"Deployment state file {state_path} declares app_id "
            f"{declared!r}, but its filename says {state_path.stem!r}. "
            "The file was likely copied or renamed. Fix whichever is "
            "wrong before continuing."
        )
    return state

save_deployment_state

save_deployment_state(state: dict[str, Any], state_path: Path) -> None

Saves deployment state for one app deterministically.

Creates parent directories if needed. Stamps the schema version and the app_id (from the filename, which is the identity). Output is byte-identical for logically identical state: keys follow reading order, indentation is fixed at 2 spaces, rings are sorted by name, and no timestamps or run-specific values are written.

Parameters:

Name	Type	Description	Default
state	dict[str, Any]	Deployment state dictionary to save.	required
state_path	Path	Path to the app's deployment state file.	required

Raises:

Type	Description
OSError	If the file cannot be written due to permissions.

Source code in napt/state/deployment.py

def save_deployment_state(state: dict[str, Any], state_path: Path) -> None:
    """Saves deployment state for one app deterministically.

    Creates parent directories if needed. Stamps the schema version and
    the ``app_id`` (from the filename, which is the identity). Output is
    byte-identical for logically identical state: keys follow reading
    order, indentation is fixed at 2 spaces, rings are sorted by name,
    and no timestamps or run-specific values are written.

    Args:
        state: Deployment state dictionary to save.
        state_path: Path to the app's deployment state file.

    Raises:
        OSError: If the file cannot be written due to permissions.

    """
    state["schemaVersion"] = DEPLOYMENT_STATE_SCHEMA_VERSION
    state["app_id"] = state_path.stem
    state_path.parent.mkdir(parents=True, exist_ok=True)

    ordered = _in_reading_order(state, _TOP_LEVEL_ORDER)
    for block, order in _BLOCK_ORDERS.items():
        if isinstance(ordered.get(block), dict):
            ordered[block] = _in_reading_order(ordered[block], order)
    if isinstance(ordered.get("rings"), dict):
        ordered["rings"] = {
            ring: _in_reading_order(entry, _RING_ENTRY_ORDER)
            for ring, entry in sorted(ordered["rings"].items())
        }
    if isinstance(ordered.get("retained"), list):
        ordered["retained"] = [
            _in_reading_order(entry, _RETAINED_ENTRY_ORDER)
            for entry in ordered["retained"]
        ]

    with open(state_path, "w", encoding="utf-8") as f:
        json.dump(ordered, f, indent=2)
        f.write("\n")  # Trailing newline for git

record_pending

record_pending(
    state: dict[str, Any], version: str, sha256: str, url: str
) -> str | None

Records a discovered release as the pending publication candidate.

The pending slot holds exactly one candidate and the newest discovery wins: a release that differs from both the published release and the current pending candidate replaces the pending candidate. Identity is the SHA-256 hash, not the version string, so a vendor re-release of the same version with a different binary is treated as new.

Parameters:

Name	Type	Description	Default
state	dict[str, Any]	Deployment state dictionary to update in place.	required
version	str	Discovered version string.	required
sha256	str	SHA-256 hash of the discovered installer.	required
url	str	Download URL of the discovered installer.	required

Returns:

Type	Description
str | None	A string naming the change made ("recorded" for a first candidate, "replaced" when a candidate was overwritten, "cleared" when the vendor serves the already-published release), or None when the state did not change.

Source code in napt/state/deployment.py

def record_pending(
    state: dict[str, Any],
    version: str,
    sha256: str,
    url: str,
) -> str | None:
    """Records a discovered release as the pending publication candidate.

    The pending slot holds exactly one candidate and the newest discovery
    wins: a release that differs from both the published release and the
    current pending candidate replaces the pending candidate. Identity is
    the SHA-256 hash, not the version string, so a vendor re-release of
    the same version with a different binary is treated as new.

    Args:
        state: Deployment state dictionary to update in place.
        version: Discovered version string.
        sha256: SHA-256 hash of the discovered installer.
        url: Download URL of the discovered installer.

    Returns:
        A string naming the change made ("recorded" for a first candidate,
            "replaced" when a candidate was overwritten, "cleared" when the
            vendor serves the already-published release), or
            None when the state did not change.

    """
    published = state.get("published")
    pending = state.get("pending")

    if published and published.get("sha256") == sha256:
        # Vendor serves exactly what is published; nothing awaits publication.
        if pending is not None:
            state["pending"] = None
            return "cleared"
        return None

    if pending and pending.get("sha256") == sha256:
        return None

    state["pending"] = {
        "version": version,
        "sha256": sha256,
        "url": url,
    }
    return "replaced" if pending else "recorded"

record_published

record_published(
    state: dict[str, Any],
    version: str,
    sha256: str,
    intune_app_id: str | None,
    intune_update_app_id: str | None,
) -> None

Records a successful publication as the published release.

Replaces the published section and clears the pending slot when the pending candidate is the release that was just published. A pending candidate with a different hash (a newer discovery) is left in place.

Parameters:

Name	Type	Description	Default
state	dict[str, Any]	Deployment state dictionary to update in place.	required
version	str	Published version string.	required
sha256	str	SHA-256 hash of the published release's installer.	required
intune_app_id	str | None	Graph API object ID of the install entry, or None when build_types is "update_only".	required
intune_update_app_id	str | None	Graph API object ID of the update entry, or None when build_types is "app_only".	required

Source code in napt/state/deployment.py

def record_published(
    state: dict[str, Any],
    version: str,
    sha256: str,
    intune_app_id: str | None,
    intune_update_app_id: str | None,
) -> None:
    """Records a successful publication as the published release.

    Replaces the ``published`` section and clears the pending slot when
    the pending candidate is the release that was just published. A
    pending candidate with a different hash (a newer discovery) is left
    in place.

    Args:
        state: Deployment state dictionary to update in place.
        version: Published version string.
        sha256: SHA-256 hash of the published release's installer.
        intune_app_id: Graph API object ID of the install entry, or None
            when build_types is "update_only".
        intune_update_app_id: Graph API object ID of the update entry, or
            None when build_types is "app_only".

    """
    state["published"] = {
        "version": version,
        "sha256": sha256,
        "intune_app_id": intune_app_id,
        "intune_update_app_id": intune_update_app_id,
    }

    pending = state.get("pending")
    if pending and pending.get("sha256") == sha256:
        state["pending"] = None

summarize_deployment_states

summarize_deployment_states(deployment_dir: Path) -> list[dict[str, Any]]

Summarizes all per-app deployment state files in a directory.

Parameters:

Name	Type	Description	Default
deployment_dir	Path	Directory holding per-app deployment state files.	required

Returns:

Type	Description
list[dict[str, Any]]	One summary dict per app, sorted by app id, each with the app id, published version, pending version, and a ring-to-version map. Empty when the directory does not exist or holds no state.

Raises:

Type	Description
StateError	On a corrupted deployment state file.

Source code in napt/state/deployment.py

def summarize_deployment_states(deployment_dir: Path) -> list[dict[str, Any]]:
    """Summarizes all per-app deployment state files in a directory.

    Args:
        deployment_dir: Directory holding per-app deployment state files.

    Returns:
        One summary dict per app, sorted by app id, each with the app id,
            published version, pending version, and a ring-to-version
            map. Empty when the directory does not exist or holds no
            state.

    Raises:
        StateError: On a corrupted deployment state file.

    """
    if not deployment_dir.is_dir():
        return []

    rows: list[dict[str, Any]] = []
    for path in sorted(deployment_dir.glob("*.json")):
        state = load_deployment_state(path)
        published = state.get("published") or {}
        pending = state.get("pending") or {}
        rings = state.get("rings") or {}
        rows.append(
            {
                "app_id": path.stem,
                "published": published.get("version"),
                "pending": pending.get("version"),
                "rings": {
                    name: entry.get("version") for name, entry in sorted(rings.items())
                },
            }
        )
    return rows