state

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 downloads folder, 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.

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, whether the pending version is lower than the published one (pending_is_downgrade), 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, whether the pending
            version is lower than the published one
            (``pending_is_downgrade``), 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"),
                # Derived on every read, never stored: the two versions it
                # compares change independently.
                "pending_is_downgrade": bool(pending)
                and is_downgrade(pending["version"], published.get("version")),
                "rings": {
                    name: entry.get("version") for name, entry in sorted(rings.items())
                },
            }
        )
    return rows

napt.state.stamp

Provenance stamp for NAPT-managed Intune apps.

The stamp is a single machine-parseable line written to the Intune notes field of every app NAPT publishes:

napt/v1 id=<recipe-id> entry=<install|update> sha256=<installer-hash>

It serves two purposes: ownership (presence of the stamp marks an app as NAPT-managed; unstamped apps are never touched) and identity (the recipe id, entry type, and installer hash tie the Intune object to a specific publish instance recorded in deployment state). The notes field is reserved for NAPT and is not recipe-configurable.

build_stamp

build_stamp(recipe_id: str, entry: str, sha256: str) -> str

Builds the provenance stamp for one Intune app entry.

Parameters:

Name	Type	Description	Default
recipe_id	str	Recipe identifier (from recipe's 'id' field).	required
entry	str	Entry type, either "install" or "update".	required
sha256	str	SHA-256 hex digest of the source installer.	required

Returns:

Type	Description
str	The stamp line to write to the Intune notes field.

Raises:

Type	Description
ConfigError	If the stamp would exceed Intune's notes field length limit (only possible with an extremely long recipe id).

Source code in napt/state/stamp.py

def build_stamp(recipe_id: str, entry: str, sha256: str) -> str:
    """Builds the provenance stamp for one Intune app entry.

    Args:
        recipe_id: Recipe identifier (from recipe's 'id' field).
        entry: Entry type, either "install" or "update".
        sha256: SHA-256 hex digest of the source installer.

    Returns:
        The stamp line to write to the Intune notes field.

    Raises:
        ConfigError: If the stamp would exceed Intune's notes field length
            limit (only possible with an extremely long recipe id).

    """
    stamp = f"{STAMP_PREFIX} id={recipe_id} entry={entry} sha256={sha256}"
    if len(stamp) > NOTES_MAX_LENGTH:
        raise ConfigError(
            f"Provenance stamp for '{recipe_id}' is {len(stamp)} characters, "
            f"over Intune's {NOTES_MAX_LENGTH}-character notes field limit. "
            "Shorten the recipe id."
        )
    return stamp

find_stamped_app

find_stamped_app(
    apps: list[dict], recipe_id: str, entry: str, sha256: str
) -> dict | None

Finds the app whose provenance stamp matches a publish instance.

Parameters:

Name	Type	Description	Default
apps	list[dict]	Mobile app dicts (with "notes") from list_mobile_apps.	required
recipe_id	str	Recipe identifier to match.	required
entry	str	Entry type to match ("install" or "update").	required
sha256	str	Installer hash to match.	required

Returns:

Type	Description
dict | None	The matching app dict, or None when no stamped app matches.

Source code in napt/state/stamp.py

def find_stamped_app(
    apps: list[dict],
    recipe_id: str,
    entry: str,
    sha256: str,
) -> dict | None:
    """Finds the app whose provenance stamp matches a publish instance.

    Args:
        apps: Mobile app dicts (with "notes") from list_mobile_apps.
        recipe_id: Recipe identifier to match.
        entry: Entry type to match ("install" or "update").
        sha256: Installer hash to match.

    Returns:
        The matching app dict, or None when no stamped app matches.

    """
    for app in apps:
        stamp = parse_stamp(app.get("notes"))
        if (
            stamp
            and stamp["id"] == recipe_id
            and stamp["entry"] == entry
            and stamp["sha256"] == sha256
        ):
            return app
    return None

parse_stamp

parse_stamp(notes: str | None) -> dict[str, str] | None

Parses a provenance stamp from an Intune notes field value.

Parameters:

Name	Type	Description	Default
notes	str | None	The notes field content, or None.	required

Returns:

Type	Description
dict[str, str] | None	A dict with "id", "entry", and "sha256" keys, or None when the notes do not carry a complete NAPT stamp.

Source code in napt/state/stamp.py

def parse_stamp(notes: str | None) -> dict[str, str] | None:
    """Parses a provenance stamp from an Intune notes field value.

    Args:
        notes: The notes field content, or None.

    Returns:
        A dict with "id", "entry", and "sha256" keys, or None when the
            notes do not carry a complete NAPT stamp.

    """
    if not notes or not notes.startswith(f"{STAMP_PREFIX} "):
        return None

    fields: dict[str, str] = {}
    for token in notes.split()[1:]:
        key, sep, value = token.partition("=")
        if sep and value:
            fields[key] = value

    if any(key not in fields for key in _REQUIRED_KEYS):
        return None
    return {key: fields[key] for key in _REQUIRED_KEYS}