diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index c015d6e5ea435..d7b7160eb3884 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -307,6 +307,21 @@ jobs:
       - name: Stop gradle daemon
         run: ./app/gradlew --stop
 
+  host-tests:
+    name: Host tests (python)
+    runs-on: ubuntu-24.04
+    timeout-minutes: 10
+    steps:
+      - name: Check out
+        uses: actions/checkout@v6
+
+      - name: GrapheneOS contract suites
+        run: python3 -m unittest tests.test_grapheneos_bindings tests.test_grapheneos_zygote_contract -v
+
+      - name: Zygisk exec-spawn suite
+        working-directory: native/src/core/zygisk
+        run: python3 -m unittest discover -p 'test_*.py' -v
+
   avd-test:
     name: Test API ${{ matrix.version }} (x86_64)
     runs-on: ubuntu-24.04
diff --git a/.github/workflows/grapheneos_zygote_contract_monitor.yml b/.github/workflows/grapheneos_zygote_contract_monitor.yml
new file mode 100644
index 0000000000000..dbfb4edf31bc3
--- /dev/null
+++ b/.github/workflows/grapheneos_zygote_contract_monitor.yml
@@ -0,0 +1,90 @@
+name: GrapheneOS Zygote Contract Monitor
+
+on:
+  workflow_dispatch:
+  schedule:
+    # Twice a month (1st and 15th) at 04:17 UTC.
+    - cron: "17 4 1,15 * *"
+
+permissions:
+  contents: read
+  issues: write
+
+concurrency:
+  group: grapheneos-zygote-contract-monitor
+  cancel-in-progress: false
+
+jobs:
+  monitor:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4
+        with:
+          persist-credentials: false
+
+      - name: Check GrapheneOS zygote contract
+        id: monitor
+        run: |
+          set +e
+          python3 scripts/grapheneos_zygote_contract.py \
+            --out-json grapheneos_zygote_contract_current.json \
+            --report grapheneos_zygote_contract_report.md
+          status=$?
+          set -e
+          if [ "$status" = "0" ]; then
+            echo "drift=false" >> "$GITHUB_OUTPUT"
+          elif [ "$status" = "2" ]; then
+            echo "drift=true" >> "$GITHUB_OUTPUT"
+          else
+            exit "$status"
+          fi
+
+      - name: Upload contract report
+        if: always()
+        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4
+        with:
+          name: grapheneos-zygote-contract
+          if-no-files-found: warn
+          path: |
+            grapheneos_zygote_contract_current.json
+            grapheneos_zygote_contract_report.md
+
+      - name: Run contract regression suites
+        if: always()
+        run: python3 -m unittest tests.test_grapheneos_bindings tests.test_grapheneos_zygote_contract -v
+
+      - name: Run zygisk behavior suite
+        if: always()
+        working-directory: native/src/core/zygisk
+        run: python3 -m unittest discover -p 'test_*.py' -v
+
+      - name: Sync drift issue
+        if: github.repository == 'pixincreate/Magisk'
+        env:
+          GH_TOKEN: ${{ github.token }}
+          DRIFT: ${{ steps.monitor.outputs.drift }}
+          MARKER: "<!-- grapheneos-zygote-contract-monitor:v1 -->"
+          RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
+        run: |
+          set -euo pipefail
+          issue="$(gh issue list --state all --limit 1000 --json number,body,title,state,author --jq "map(select(.title == \"GrapheneOS zygote semantic contract drift\" and .author.login == \"github-actions[bot]\" and ((.body // \"\") | contains(\"$MARKER\")))) | sort_by(.number) | last | if . then [.number, .state] | @tsv else \"\" end")"
+          issue_number="${issue%%$'\t'*}"
+          issue_state="${issue#*$'\t'}"
+          if [ "$DRIFT" = "true" ]; then
+            {
+              cat grapheneos_zygote_contract_report.md
+              printf '\nWorkflow run: %s\n' "$RUN_URL"
+            } > issue_body.md
+            if [ -n "$issue_number" ]; then
+              if [ "$issue_state" = "CLOSED" ]; then
+                gh issue reopen "$issue_number"
+              fi
+              gh issue edit "$issue_number" --title "GrapheneOS zygote semantic contract drift" --body-file issue_body.md
+            else
+              gh issue create --title "GrapheneOS zygote semantic contract drift" --body-file issue_body.md
+            fi
+          elif [ -n "$issue_number" ] && [ "$issue_state" = "OPEN" ]; then
+            gh issue comment "$issue_number" --body "GrapheneOS zygote contract matches the committed baseline again. Closing from $RUN_URL."
+            gh issue close "$issue_number" --reason completed
+          fi
diff --git a/app/test/src/main/java/com/topjohnwu/magisk/test/BootloaderLockUiTest.kt b/app/test/src/main/java/com/topjohnwu/magisk/test/BootloaderLockUiTest.kt
new file mode 100644
index 0000000000000..cc7cdf009de74
--- /dev/null
+++ b/app/test/src/main/java/com/topjohnwu/magisk/test/BootloaderLockUiTest.kt
@@ -0,0 +1,132 @@
+package com.topjohnwu.magisk.test
+
+import android.os.ParcelFileDescriptor.AutoCloseInputStream
+import androidx.annotation.Keep
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import androidx.test.platform.app.InstrumentationRegistry
+import androidx.test.uiautomator.By
+import androidx.test.uiautomator.UiDevice
+import org.junit.After
+import org.junit.Assert.assertNotNull
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import org.junit.runner.RunWith
+
+/**
+ * Regression tests for the bootloader-lock UI guard introduced in bf2e846
+ * (hide direct install / uninstall when the bootloader is locked).
+ *
+ * Runs against the real app UI with UiAutomator. The lock state comes from
+ * ro.boot.vbmeta.device_state, toggled with resetprop through Magisk's su.
+ * Because Info.isBootloaderLocked caches the value per process, every state
+ * change is followed by an app restart.
+ */
+@Keep
+@RunWith(AndroidJUnit4::class)
+class BootloaderLockUiTest {
+
+    companion object {
+        private const val APP_PKG = "com.topjohnwu.magisk"
+        private const val LOCK_STATE_PROP = "ro.boot.vbmeta.device_state"
+
+        private const val UNINSTALL = "Uninstall Magisk"
+        private const val DIRECT_INSTALL = "Direct install (Recommended)"
+        private const val PATCH_FILE = "Select and patch a file"
+
+        // The home card action label depends on environment state
+        // (Reinstall/Install/Update), so accept all of them as "home ready".
+        private val ACTION_LABELS = arrayOf("Reinstall", "Install", "Update")
+
+        private const val TIMEOUT_MS = 15_000L
+        private const val GRACE_MS = 3_000L
+    }
+
+    private val device = UiDevice.getInstance(InstrumentationRegistry.getInstrumentation())
+    private val uiAutomation get() = InstrumentationRegistry.getInstrumentation().uiAutomation
+
+    @After
+    fun tearDown() {
+        setLockStateAndRestart("unlocked")
+    }
+
+    private fun shell(cmd: String): String {
+        val pfd = uiAutomation.executeShellCommand(cmd)
+        return AutoCloseInputStream(pfd).reader().use { it.readText() }
+    }
+
+    private fun setLockStateAndRestart(state: String) {
+        shell("su -c 'resetprop $LOCK_STATE_PROP $state'")
+        shell("am force-stop $APP_PKG")
+        shell("monkey -p $APP_PKG -c android.intent.category.LAUNCHER 1")
+        assertNotNull(
+            "Magisk home never became ready after restart (state=$state)",
+            awaitAnyOf(*ACTION_LABELS)
+        )
+    }
+
+    private fun awaitAnyOf(vararg texts: String): String? {
+        val deadline = System.currentTimeMillis() + TIMEOUT_MS
+        while (System.currentTimeMillis() < deadline) {
+            texts.firstOrNull { device.findObject(By.text(it)) != null }?.let { return it }
+            Thread.sleep(250)
+        }
+        return null
+    }
+
+    /** True if [text] shows up within [timeout]; used both for presence and short-grace absence checks. */
+    private fun waitText(text: String, timeout: Long = GRACE_MS): Boolean {
+        val deadline = System.currentTimeMillis() + timeout
+        while (System.currentTimeMillis() < deadline) {
+            if (device.findObject(By.text(text)) != null) return true
+            Thread.sleep(250)
+        }
+        return false
+    }
+
+    private fun openInstallSheet() {
+        val action = awaitAnyOf(*ACTION_LABELS)
+        assertNotNull("No install action button found on home screen", action)
+        val button = action?.let { device.findObject(By.text(it)) }
+        assertNotNull("Action button disappeared before it could be tapped", button)
+        button!!.click()
+        // Positive control: this row exists regardless of lock state,
+        // so reaching it proves the sheet actually opened.
+        assertTrue(
+            "Install sheet did not open ($PATCH_FILE never appeared)",
+            waitText(PATCH_FILE, TIMEOUT_MS)
+        )
+    }
+
+    @Test
+    fun testBootloaderLockTogglesInstallSurfaces() {
+        // --- Locked: destructive surfaces hidden ---
+        setLockStateAndRestart("locked")
+
+        assertTrue(
+            "$UNINSTALL must be hidden when the bootloader is locked",
+            !waitText(UNINSTALL)
+        )
+
+        openInstallSheet()
+        assertTrue(
+            "$DIRECT_INSTALL must be hidden when the bootloader is locked",
+            !waitText(DIRECT_INSTALL)
+        )
+        device.pressBack()
+
+        // --- Unlocked: destructive surfaces restored ---
+        setLockStateAndRestart("unlocked")
+
+        assertTrue(
+            "$UNINSTALL must be visible when the bootloader is unlocked",
+            waitText(UNINSTALL, TIMEOUT_MS)
+        )
+
+        openInstallSheet()
+        assertTrue(
+            "$DIRECT_INSTALL must be visible when the bootloader is unlocked",
+            waitText(DIRECT_INSTALL, TIMEOUT_MS)
+        )
+        device.pressBack()
+    }
+}
diff --git a/scripts/grapheneos_zygote_contract.py b/scripts/grapheneos_zygote_contract.py
new file mode 100644
index 0000000000000..47c747fef7fc3
--- /dev/null
+++ b/scripts/grapheneos_zygote_contract.py
@@ -0,0 +1,341 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import re
+import sys
+from pathlib import Path
+from urllib.error import HTTPError, URLError
+from urllib.request import Request, urlopen
+
+REPOSITORY = "GrapheneOS/platform_frameworks_base"
+BRANCH = "17"
+API_URL = f"https://api.github.com/repos/{REPOSITORY}/commits/{BRANCH}"
+BRANCHES_URL = f"https://api.github.com/repos/{REPOSITORY}/branches"
+RAW_BASE_URL = f"https://raw.githubusercontent.com/{REPOSITORY}"
+USER_AGENT = "Magisk GrapheneOS zygote contract monitor/1"
+MAX_SOURCE_BYTES = 2 * 1024 * 1024
+MAX_BRANCH_PAGES = 3
+SOURCE_PATHS = (
+    "core/java/com/android/internal/os/ZygoteConnection.java",
+    "core/java/com/android/internal/os/ZygoteExtraArgs.java",
+    "core/java/com/android/internal/os/ExecSpawning.java",
+    "core/jni/com_android_internal_os_Zygote.cpp",
+)
+JNI_METHODS = (
+    "nativeForkAndSpecialize",
+    "nativeForkSystemServer",
+    "nativeSpecializeAppProcess",
+)
+
+
+class ContractError(RuntimeError):
+    pass
+
+
+def compact(text: str) -> str:
+    return re.sub(r"\s+", " ", text).strip()
+
+
+def require(pattern: str, text: str, field: str, flags: int = re.S) -> re.Match[str]:
+    match = re.search(pattern, text, flags)
+    if match is None:
+        raise ContractError(f"missing GrapheneOS zygote contract field: {field}")
+    return match
+
+
+def strip_comments(text: str) -> str:
+    text = re.sub(r"/\*.*?\*/", "", text, flags=re.S)
+    return re.sub(r"//.*", "", text)
+
+
+def int_expr(expr: str) -> int:
+    clean = expr.replace(" ", "")
+    match = re.fullmatch(r"(\d+)(?:<<(\d+))?", clean)
+    if match is None:
+        raise ContractError(f"unsupported integer expression: {expr}")
+    value = int(match.group(1))
+    return value << int(match.group(2) or "0")
+
+
+def read_response(url: str, timeout: float, max_bytes: int) -> bytes:
+    request = Request(url, headers={"User-Agent": USER_AGENT})
+    try:
+        with urlopen(request, timeout=timeout) as response:
+            content_length = response.headers.get("Content-Length")
+            if content_length is not None and int(content_length) > max_bytes:
+                raise ContractError(f"response exceeds {max_bytes} bytes: {url}")
+            data = response.read(max_bytes + 1)
+    except (HTTPError, URLError, TimeoutError, OSError, ValueError) as exc:
+        raise ContractError(f"failed to fetch {url}: {exc}") from exc
+    if len(data) > max_bytes:
+        raise ContractError(f"response exceeds {max_bytes} bytes: {url}")
+    return data
+
+
+def fetch_sources(timeout: float) -> tuple[dict[str, str], str]:
+    try:
+        revision = json.loads(read_response(API_URL, timeout, MAX_SOURCE_BYTES))["sha"]
+    except (json.JSONDecodeError, KeyError, TypeError) as exc:
+        raise ContractError(f"failed to resolve GrapheneOS branch {BRANCH}: {exc}") from exc
+    if not isinstance(revision, str) or re.fullmatch(r"[0-9a-f]{40}", revision) is None:
+        raise ContractError(f"invalid GrapheneOS revision: {revision!r}")
+
+    sources: dict[str, str] = {}
+    for path in SOURCE_PATHS:
+        try:
+            sources[path] = read_response(
+                f"{RAW_BASE_URL}/{revision}/{path}", timeout, MAX_SOURCE_BYTES
+            ).decode("utf-8")
+        except UnicodeDecodeError as exc:
+            raise ContractError(f"failed to decode {path}: {exc}") from exc
+    return sources, revision
+
+
+def newer_upstream_branches(names: list[str]) -> list[str]:
+    pinned = int(BRANCH)
+    found = [name for name in names if re.fullmatch(r"\d+", name) and int(name) > pinned]
+    return sorted(found, key=int)
+
+
+def list_remote_branch_names(timeout: float) -> list[str]:
+    names: list[str] = []
+    url: str | None = f"{BRANCHES_URL}?per_page=100"
+    for _ in range(MAX_BRANCH_PAGES):
+        if url is None:
+            break
+        request = Request(url, headers={"User-Agent": USER_AGENT})
+        try:
+            with urlopen(request, timeout=timeout) as response:
+                link = response.headers.get("Link", "")
+                page_names = [entry["name"] for entry in json.load(response)]
+        except (HTTPError, URLError, TimeoutError, OSError, ValueError, json.JSONDecodeError, KeyError, TypeError) as exc:
+            raise ContractError(f"failed to list GrapheneOS branches: {exc}") from exc
+        names.extend(name for name in page_names if isinstance(name, str))
+        match = re.search(r'<([^>]+)>;\s*rel="next"', link)
+        url = match.group(1) if match else None
+    return names
+
+
+def check_newer_branches(timeout: float) -> list[str]:
+    return newer_upstream_branches(list_remote_branch_names(timeout))
+
+
+def read_sources(source_dir: Path) -> dict[str, str]:
+    sources: dict[str, str] = {}
+    for path in SOURCE_PATHS:
+        file_path = source_dir / path
+        try:
+            sources[path] = file_path.read_text(encoding="utf-8")
+        except OSError as exc:
+            raise ContractError(f"failed to read {file_path}: {exc}") from exc
+    return sources
+
+
+def extract_jni_descriptors(native: str) -> dict[str, str]:
+    table = require(r"static\s+const\s+JNINativeMethod\s+gMethods\[\]\s*=\s*\{(?P<body>.*?)\};", native, "JNI table").group("body")
+    descriptors: dict[str, str] = {}
+    for name in JNI_METHODS:
+        pattern = r'\{"' + re.escape(name) + r'"\s*,\s*(?P<parts>(?:"[^"]*"\s*)+)\s*,\s*\(void\*\)'
+        parts = require(pattern, table, f"JNI descriptor {name}").group("parts")
+        descriptors[name] = "".join(re.findall(r'"([^"]*)"', parts))
+    return descriptors
+
+
+def braced_block(pattern: str, text: str, field: str) -> str:
+    match = require(pattern, text, field, 0)
+    start = text.find("{", match.end())
+    if start == -1:
+        raise ContractError(f"missing opening brace for {field}")
+    depth = 0
+    for index in range(start, len(text)):
+        if text[index] == "{":
+            depth += 1
+        elif text[index] == "}":
+            depth -= 1
+            if depth == 0:
+                return text[start + 1 : index]
+    raise ContractError(f"missing closing brace for {field}")
+
+
+def extract_extra_args(extra: str, native: str) -> dict[str, object]:
+    clean_extra = strip_comments(extra)
+    use_expr = require(r"int\s+USE_ZYGOTE_SPAWNING\s*=\s*([^;]+);", clean_extra, "USE_ZYGOTE_SPAWNING").group(1)
+    selinux_idx = require(r"IDX_SELINUX_FLAGS\s*=\s*(\d+)", clean_extra, "IDX_SELINUX_FLAGS").group(1)
+    flags_idx = require(r"IDX_FLAGS\s*=\s*(\d+)", clean_extra, "IDX_FLAGS").group(1)
+    arr_len = require(r"ARR_LEN\s*=\s*(\d+)", clean_extra, "ARR_LEN").group(1)
+    make_body = require(r"long\[\]\s+res\s*=\s*new\s+long\[ARR_LEN\];(?P<body>.*?)return\s+res\s*;", clean_extra, "makeJniLongArray").group("body")
+    native_body = require(r"ExtraArgs\(JNIEnv\* env, jlongArray jlongArgs\).*?\{(?P<body>.*?)\n\s*\}", native, "native ExtraArgs constructor").group("body")
+    native_flag = require(r"FORCIBLY_ENABLE_MEMORY_TAGGING\s*=\s*([^;]+);", native, "native ExtraArgs flag").group(1)
+    java_order = [
+        list(item) for item in re.findall(r"res\[(IDX_[A-Z_]+)\]\s*=\s*([A-Za-z0-9_.$]+)", make_body)
+    ]
+    expected_java_order = [["IDX_SELINUX_FLAGS", "selinuxFlags"], ["IDX_FLAGS", "flags"]]
+    if java_order != expected_java_order:
+        raise ContractError(f"unexpected Java extra-long-args order: {java_order}")
+
+    native_order = [
+        list(item) for item in re.findall(r"(selinux_flags|flags)\s*=.*?jlong_arr\[(\d+)\]", native_body)
+    ]
+    expected_native_order = [["selinux_flags", "0"], ["flags", "1"]]
+    if native_order != expected_native_order:
+        raise ContractError(f"unexpected native extra-long-args order: {native_order}")
+
+    exec_body = compact(braced_block(r"boolean\s+shouldUseExecSpawning\s*\(\s*\)", clean_extra, "shouldUseExecSpawning"))
+    if exec_body != "return !hasFlag(Flag.USE_ZYGOTE_SPAWNING);":
+        raise ContractError(f"unexpected shouldUseExecSpawning contract: {exec_body}")
+
+    return {
+        "use_zygote_spawning": {"expression": compact(use_expr), "value": int_expr(use_expr)},
+        "should_use_exec_spawning": exec_body,
+        "java_indices": {"IDX_SELINUX_FLAGS": int(selinux_idx), "IDX_FLAGS": int(flags_idx), "ARR_LEN": int(arr_len)},
+        "java_make_jni_long_array": java_order,
+        "native_jlong_array_order": native_order,
+        "native_forcibly_enable_memory_tagging": {"expression": compact(native_flag), "value": int_expr(native_flag)},
+    }
+
+
+def extract_replay_contract(connection: str, exec_spawning: str, native: str) -> dict[str, object]:
+    clean_conn = strip_comments(connection)
+    fds = require(r"int\s*\[\]\s*fdsToClose\s*=\s*\{\s*(-?\d+)\s*,\s*(-?\d+)\s*\};", clean_conn, "fdsToClose replay sentinel")
+    guard_body = braced_block(r"if\s*\(\s*!isReplayingZygoteCommands\s*\)", clean_conn[fds.end() :], "fdsToClose replay guard")
+    assignments = sorted(
+        int(index)
+        for index in re.findall(r"fdsToClose\s*\[\s*([01])\s*\]\s*=\s*(?:fd|zygoteFd)\.getInt\$\(\s*\)", guard_body)
+    )
+    if assignments != [0, 1]:
+        raise ContractError(f"unexpected guarded fdsToClose assignments: {assignments}")
+    replay_call = require(r"processCommand\(\s*zygoteServer\s*,\s*false\s*,\s*cmd\s*\)", exec_spawning, "replay processCommand false")
+    detach = require(r"for\s*\(int\s+fd\s*:\s*fds_to_close\).*?if\s*\(\s*fd\s*==\s*-1\s*&&\s*gIsExecSpawning\s*\)\s*\{\s*continue\s*;\s*\}", native, "native fds_to_close -1 exec sentinel")
+    fork = require(r"pid_t\s+pid\s*=\s*gIsExecSpawning\s*\?\s*0\s*:\s*fork\s*\(\s*\)\s*;", native, "gIsExecSpawning fork behavior")
+    return {
+        "fds_to_close_initial_sentinel": [int(fds.group(1)), int(fds.group(2))],
+        "fds_to_close_socket_fill_guard": "!isReplayingZygoteCommands",
+        "fds_to_close_guarded_assignments": assignments,
+        "replay_process_command_multiple_ok": False,
+        "replay_process_command_call": compact(replay_call.group(0)),
+        "native_detach_exec_sentinel": compact(detach.group(0)),
+        "native_fork_expression": compact(fork.group(0)),
+    }
+
+
+def extract_contract(sources: dict[str, str], source_revision: str | None = None) -> dict[str, object]:
+    native = sources["core/jni/com_android_internal_os_Zygote.cpp"]
+    contract = {
+        "upstream": {"repository": REPOSITORY, "branch": BRANCH},
+        "jni_descriptors": extract_jni_descriptors(native),
+        "extra_long_args": extract_extra_args(sources["core/java/com/android/internal/os/ZygoteExtraArgs.java"], native),
+        "replay": extract_replay_contract(
+            sources["core/java/com/android/internal/os/ZygoteConnection.java"],
+            sources["core/java/com/android/internal/os/ExecSpawning.java"],
+            native,
+        ),
+    }
+    if source_revision is not None:
+        contract["source_revision"] = source_revision
+    contract["canonical_sha256"] = canonical_hash(contract)
+    return contract
+
+
+def canonical_json(contract: dict[str, object]) -> str:
+    semantic = {
+        key: value
+        for key, value in contract.items()
+        if key not in {"canonical_sha256", "source_revision"}
+    }
+    return json.dumps(semantic, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+
+
+def canonical_hash(contract: dict[str, object]) -> str:
+    return hashlib.sha256(canonical_json(contract).encode()).hexdigest()
+
+
+def write_json(path: Path, contract: dict[str, object]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(contract, sort_keys=True, indent=2) + "\n", encoding="utf-8")
+
+
+def markdown_report(
+    current: dict[str, object],
+    baseline: dict[str, object] | None,
+    newer_branches: list[str] | None = None,
+) -> str:
+    lines = ["<!-- grapheneos-zygote-contract-monitor:v1 -->", "# GrapheneOS Zygote contract drift", ""]
+    lines.append(f"Current hash: `{current['canonical_sha256']}`")
+    if "source_revision" in current:
+        lines.append(f"Source revision: `{current['source_revision']}`")
+    if baseline is None:
+        lines.append("Baseline: missing")
+    else:
+        lines.append(f"Baseline hash: `{baseline.get('canonical_sha256', '<missing>')}`")
+        if canonical_hash(current) == canonical_hash(baseline):
+            lines.append("\nNo semantic contract drift detected.")
+        else:
+            lines.append("\nSemantic contract drift detected. Review `grapheneos_zygote_contract_current.json`.")
+    lines.append("\nMonitored fields: JNI descriptors, extra-long-args indices/flags, replay fds sentinel, replay `processCommand(..., false, ...)`, and exec-spawn fork behavior.")
+    if newer_branches is not None:
+        if newer_branches:
+            lines.append(
+                f"\n**Newer upstream Android branch detected: {', '.join(newer_branches)}.** "
+                f"The committed baseline targets branch `{BRANCH}`. Re-baseline required: bump "
+                f"`BRANCH` in `scripts/grapheneos_zygote_contract.py`, refresh the fixtures and "
+                f"`baseline.json`, and review the new descriptors for additional JNI hook variants."
+            )
+        else:
+            lines.append(f"\nNo upstream Android branch newer than `{BRANCH}` detected.")
+    return "\n".join(lines) + "\n"
+
+
+def run(source_dir: Path | None, baseline_path: Path, out_json: Path, report: Path, timeout: float) -> int:
+    newer_branches: list[str] | None = None
+    if source_dir:
+        sources = read_sources(source_dir)
+        revision = None
+    else:
+        sources, revision = fetch_sources(timeout)
+        try:
+            newer_branches = check_newer_branches(timeout)
+        except ContractError:
+            # The watchdog must never mask the primary drift signal.
+            newer_branches = None
+    current = extract_contract(sources, revision)
+    baseline = json.loads(baseline_path.read_text(encoding="utf-8")) if baseline_path.exists() else None
+    write_json(out_json, current)
+    report.write_text(markdown_report(current, baseline, newer_branches), encoding="utf-8")
+    drift = baseline is None or canonical_hash(current) != canonical_hash(baseline)
+    if drift or newer_branches:
+        return 2
+    return 0
+
+
+def main_with_args_for_test(source_dir: Path, baseline: Path, out_json: Path, report: Path) -> int:
+    return run(source_dir, baseline, out_json, report, 20.0)
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Check GrapheneOS zygote semantic contract drift.")
+    parser.add_argument("--source-dir", type=Path)
+    parser.add_argument("--baseline", type=Path, default=Path("tests/fixtures/grapheneos_zygote_contract/baseline.json"))
+    parser.add_argument("--out-json", type=Path, default=Path("grapheneos_zygote_contract_current.json"))
+    parser.add_argument("--report", type=Path, default=Path("grapheneos_zygote_contract_report.md"))
+    parser.add_argument("--timeout", type=float, default=20.0)
+    args = parser.parse_args()
+    try:
+        return run(args.source_dir, args.baseline, args.out_json, args.report, args.timeout)
+    except ContractError as exc:
+        message = f"error: {exc}"
+        print(message, file=sys.stderr)
+        args.report.parent.mkdir(parents=True, exist_ok=True)
+        args.report.write_text(
+            "<!-- grapheneos-zygote-contract-monitor:v1 -->\n"
+            "# GrapheneOS Zygote contract extraction failed\n\n"
+            f"`{message}`\n",
+            encoding="utf-8",
+        )
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/test_common.sh b/scripts/test_common.sh
index 6be47d189029c..a760559bc9225 100644
--- a/scripts/test_common.sh
+++ b/scripts/test_common.sh
@@ -76,6 +76,10 @@ run_tests() {
   # Run app tests
   am_instrument '.MagiskAppTest,.AdditionalTest' $app
 
+  # Test the bootloader-lock UI guard (needs su granted by MagiskAppTest,
+  # and must run before app hiding renames the package)
+  am_instrument '.BootloaderLockUiTest' $self
+
   # Test app hiding
   am_instrument '.AppMigrationTest#testAppHide' $self
 
diff --git a/tests/fixtures/grapheneos_zygote_contract/baseline.json b/tests/fixtures/grapheneos_zygote_contract/baseline.json
new file mode 100644
index 0000000000000..2a7966510a086
--- /dev/null
+++ b/tests/fixtures/grapheneos_zygote_contract/baseline.json
@@ -0,0 +1,63 @@
+{
+  "canonical_sha256": "a19bec7b1ff9fb0b8211fdfc777e5e8c15a6e386c9fa84cc2b52d94101f26829",
+  "extra_long_args": {
+    "java_indices": {
+      "ARR_LEN": 2,
+      "IDX_FLAGS": 1,
+      "IDX_SELINUX_FLAGS": 0
+    },
+    "java_make_jni_long_array": [
+      [
+        "IDX_SELINUX_FLAGS",
+        "selinuxFlags"
+      ],
+      [
+        "IDX_FLAGS",
+        "flags"
+      ]
+    ],
+    "native_forcibly_enable_memory_tagging": {
+      "expression": "1 << 2",
+      "value": 4
+    },
+    "native_jlong_array_order": [
+      [
+        "selinux_flags",
+        "0"
+      ],
+      [
+        "flags",
+        "1"
+      ]
+    ],
+    "should_use_exec_spawning": "return !hasFlag(Flag.USE_ZYGOTE_SPAWNING);",
+    "use_zygote_spawning": {
+      "expression": "1 << 3",
+      "value": 8
+    }
+  },
+  "jni_descriptors": {
+    "nativeForkAndSpecialize": "([JII[II[[IILjava/lang/String;Ljava/lang/String;[I[IZLjava/lang/String;Ljava/lang/String;ZZ[Ljava/lang/String;[Ljava/lang/String;ZZZ)I",
+    "nativeForkSystemServer": "(II[II[[IJJ)I",
+    "nativeSpecializeAppProcess": "([JII[II[[IILjava/lang/String;Ljava/lang/String;ZLjava/lang/String;Ljava/lang/String;Z[Ljava/lang/String;[Ljava/lang/String;ZZZ)V"
+  },
+  "replay": {
+    "fds_to_close_guarded_assignments": [
+      0,
+      1
+    ],
+    "fds_to_close_initial_sentinel": [
+      -1,
+      -1
+    ],
+    "fds_to_close_socket_fill_guard": "!isReplayingZygoteCommands",
+    "native_detach_exec_sentinel": "for (int fd : fds_to_close) { if (fd == -1 && gIsExecSpawning) { continue; }",
+    "native_fork_expression": "pid_t pid = gIsExecSpawning ? 0 : fork();",
+    "replay_process_command_call": "processCommand(zygoteServer, false, cmd)",
+    "replay_process_command_multiple_ok": false
+  },
+  "upstream": {
+    "branch": "17",
+    "repository": "GrapheneOS/platform_frameworks_base"
+  }
+}
diff --git a/tests/fixtures/grapheneos_zygote_contract/sources/core/java/com/android/internal/os/ExecSpawning.java b/tests/fixtures/grapheneos_zygote_contract/sources/core/java/com/android/internal/os/ExecSpawning.java
new file mode 100644
index 0000000000000..b59ca19360be0
--- /dev/null
+++ b/tests/fixtures/grapheneos_zygote_contract/sources/core/java/com/android/internal/os/ExecSpawning.java
@@ -0,0 +1,11 @@
+package com.android.internal.os;
+
+class ExecSpawning {
+    static boolean isReplayingZygoteCommands() { return true; }
+
+    static Runnable replay(ZygoteConnection pseudoConnection, ZygoteServer zygoteServer,
+            ZygoteArguments cmd) {
+        Runnable r = pseudoConnection.processCommand(zygoteServer, false, cmd);
+        return r;
+    }
+}
diff --git a/tests/fixtures/grapheneos_zygote_contract/sources/core/java/com/android/internal/os/ZygoteConnection.java b/tests/fixtures/grapheneos_zygote_contract/sources/core/java/com/android/internal/os/ZygoteConnection.java
new file mode 100644
index 0000000000000..4474b84229857
--- /dev/null
+++ b/tests/fixtures/grapheneos_zygote_contract/sources/core/java/com/android/internal/os/ZygoteConnection.java
@@ -0,0 +1,17 @@
+package com.android.internal.os;
+
+class ZygoteConnection {
+    Runnable processCommand(ZygoteServer zygoteServer, boolean multipleOK, ZygoteArguments command) {
+        boolean isReplayingZygoteCommands = ExecSpawning.isReplayingZygoteCommands();
+        int [] fdsToClose = { -1, -1 };
+        if (!isReplayingZygoteCommands) {
+            FileDescriptor fd = mSocket.getFileDescriptor();
+            if (fd != null) { fdsToClose[0] = fd.getInt$(); }
+            FileDescriptor zygoteFd = zygoteServer.getZygoteSocketFileDescriptor();
+            if (zygoteFd != null) { fdsToClose[1] = zygoteFd.getInt$(); }
+        }
+        boolean shouldUseExecSpawning = !isReplayingZygoteCommands
+                && parsedArgs.mExtraArgs.shouldUseExecSpawning();
+        return null;
+    }
+}
diff --git a/tests/fixtures/grapheneos_zygote_contract/sources/core/java/com/android/internal/os/ZygoteExtraArgs.java b/tests/fixtures/grapheneos_zygote_contract/sources/core/java/com/android/internal/os/ZygoteExtraArgs.java
new file mode 100644
index 0000000000000..4c1333ecf5e15
--- /dev/null
+++ b/tests/fixtures/grapheneos_zygote_contract/sources/core/java/com/android/internal/os/ZygoteExtraArgs.java
@@ -0,0 +1,32 @@
+package com.android.internal.os;
+
+public class ZygoteExtraArgs {
+    public interface Flag {
+        int DISABLE_HARDENED_MALLOC = 1;
+        int ENABLE_COMPAT_VA_39_BIT = 1 << 1;
+        int FORCIBLY_ENABLE_MEMORY_TAGGING = 1 << 2;
+        int USE_ZYGOTE_SPAWNING = 1 << 3;
+        int PREFER_COMPAT_ZYGOTE = 1 << 4;
+        int MANUALLY_RUN_ZYGOTE_PRELOAD = 1 << 5;
+    }
+
+    private long selinuxFlags;
+    private int flags;
+
+    public boolean shouldUseExecSpawning() {
+        return !hasFlag(Flag.USE_ZYGOTE_SPAWNING);
+    }
+
+    public boolean hasFlag(int flag) { return (flags & flag) == flag; }
+
+    private static final int IDX_SELINUX_FLAGS = 0;
+    private static final int IDX_FLAGS = 1;
+    private static final int ARR_LEN = 2;
+
+    public long[] makeJniLongArray() {
+        long[] res = new long[ARR_LEN];
+        res[IDX_SELINUX_FLAGS] = selinuxFlags;
+        res[IDX_FLAGS] = flags;
+        return res;
+    }
+}
diff --git a/tests/fixtures/grapheneos_zygote_contract/sources/core/jni/com_android_internal_os_Zygote.cpp b/tests/fixtures/grapheneos_zygote_contract/sources/core/jni/com_android_internal_os_Zygote.cpp
new file mode 100644
index 0000000000000..efd029a68ac9c
--- /dev/null
+++ b/tests/fixtures/grapheneos_zygote_contract/sources/core/jni/com_android_internal_os_Zygote.cpp
@@ -0,0 +1,44 @@
+namespace ExtraArgsFlag {
+    static const int FORCIBLY_ENABLE_MEMORY_TAGGING = 1 << 2;
+}
+
+struct ExtraArgs {
+    uint64_t selinux_flags = 0;
+    int flags = 0;
+
+    ExtraArgs(JNIEnv* env, jlongArray jlongArgs) {
+        const size_t num_jlong_args = 2;
+        jlong jlong_arr[num_jlong_args];
+        env->GetLongArrayRegion(jlongArgs, 0, num_jlong_args, (jlong *) &jlong_arr);
+        selinux_flags = (uint64_t) jlong_arr[0];
+        flags = (int) jlong_arr[1];
+    }
+};
+
+static bool gIsExecSpawning = false;
+
+static void DetachDescriptors(JNIEnv* env, const std::vector<int>& fds_to_close, fail_fn_t fail_fn) {
+    for (int fd : fds_to_close) {
+        if (fd == -1 && gIsExecSpawning) {
+            continue;
+        }
+    }
+}
+
+static pid_t ForkCommon() {
+    pid_t pid = gIsExecSpawning ? 0 : fork();
+    return pid;
+}
+
+static const JNINativeMethod gMethods[] = {
+        {"nativeForkAndSpecialize",
+         "([JII[II[[IILjava/lang/String;Ljava/lang/String;[I[IZLjava/lang/String;Ljava/lang/"
+         "String;ZZ[Ljava/lang/String;[Ljava/lang/String;ZZZ)I",
+         (void*)com_android_internal_os_Zygote_nativeForkAndSpecialize},
+        {"nativeForkSystemServer", "(II[II[[IJJ)I",
+         (void*)com_android_internal_os_Zygote_nativeForkSystemServer},
+        {"nativeSpecializeAppProcess",
+         "([JII[II[[IILjava/lang/String;Ljava/lang/String;ZLjava/lang/String;Ljava/lang/"
+         "String;Z[Ljava/lang/String;[Ljava/lang/String;ZZZ)V",
+         (void*)com_android_internal_os_Zygote_nativeSpecializeAppProcess},
+};
diff --git a/tests/test_grapheneos_bindings.py b/tests/test_grapheneos_bindings.py
new file mode 100644
index 0000000000000..350a968d80f69
--- /dev/null
+++ b/tests/test_grapheneos_bindings.py
@@ -0,0 +1,158 @@
+"""Regression bindings between the GrapheneOS contract baseline and this repo.
+
+The drift monitor (test_grapheneos_zygote_contract.py) proves the *baseline*
+still matches upstream GrapheneOS sources. This suite proves Magisk's *own*
+artifacts still implement that baseline:
+
+  1. generated JNI hook descriptors == baseline jni_descriptors
+  2. exec_spawn_replay.hpp constants == baseline Java indices / flag values
+  3. jni_hooks.hpp is freshly generated (gen_jni_hooks.py regen-diff)
+  4. the Android 17 memfd_file sepolicy rule is still present
+
+If any of these fail, Zygisk will silently stop matching GrapheneOS 17.
+"""
+from __future__ import annotations
+
+import json
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+REPO = Path(__file__).parents[1]
+BASELINE = REPO / "tests/fixtures/grapheneos_zygote_contract/baseline.json"
+ZYGISK = REPO / "native/src/core/zygisk"
+JNI_HOOKS = ZYGISK / "jni_hooks.hpp"
+REPLAY_HEADER = ZYGISK / "exec_spawn_replay.hpp"
+GENERATOR = ZYGISK / "gen_jni_hooks.py"
+SEPOLICY_RULES = REPO / "native/src/sepolicy/rules.rs"
+
+# Generated variant -> baseline jni_descriptors key
+DESCRIPTOR_BINDINGS = {
+    "nativeForkAndSpecialize_grapheneos_c": "nativeForkAndSpecialize",
+    "nativeSpecializeAppProcess_grapheneos_c": "nativeSpecializeAppProcess",
+    "nativeForkSystemServer_grapheneos_u": "nativeForkSystemServer",
+}
+
+
+def baseline() -> dict:
+    return json.loads(BASELINE.read_text(encoding="utf-8"))
+
+
+def descriptor_of_variant(jni_text: str, variant: str) -> str:
+    """Extract the canonical JNI descriptor string of one generated variant."""
+    lines = jni_text.splitlines()
+    try:
+        start = lines.index(f"    // {variant}")
+    except ValueError:
+        raise AssertionError(
+            f"jni_hooks.hpp no longer contains variant '{variant}'; "
+            "the GrapheneOS binding is gone from gen_jni_hooks.py"
+        )
+    for line in lines[start + 1 : start + 5]:
+        stripped = line.strip()
+        if stripped.startswith('"('):
+            return stripped.rstrip(",").strip('"')
+    raise AssertionError(f"variant '{variant}' has no descriptor string near line {start}")
+
+
+class GrapheneOsBindingsTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.baseline = baseline()
+        self.jni_text = JNI_HOOKS.read_text(encoding="utf-8")
+
+    def test_jni_descriptors_match_baseline(self) -> None:
+        for variant, method in DESCRIPTOR_BINDINGS.items():
+            expected = self.baseline["jni_descriptors"][method]
+            actual = descriptor_of_variant(self.jni_text, variant)
+            self.assertEqual(
+                actual,
+                expected,
+                f"{variant} descriptor drifted from GrapheneOS baseline "
+                f"({method}). Regenerate hooks or update the baseline "
+                "deliberately.",
+            )
+
+    def test_native_fork_flags_index_matches_java_flags_index(self) -> None:
+        expected = self.baseline["extra_long_args"]["java_indices"]["IDX_FLAGS"]
+        match = re.search(r"kGrapheneOsNativeForkFlagsIndex = (\d+)", REPLAY_HEADER.read_text(encoding="utf-8"))
+        if match is None:
+            self.fail("kGrapheneOsNativeForkFlagsIndex missing from exec_spawn_replay.hpp")
+        self.assertEqual(
+            int(match.group(1)),
+            expected,
+            "C++ flags index no longer matches Java extra-long-args IDX_FLAGS; "
+            "the replay predicate would read the wrong flag",
+        )
+
+    def test_use_zygote_spawning_flag_matches_java_value(self) -> None:
+        expected = self.baseline["extra_long_args"]["use_zygote_spawning"]["value"]
+        header = REPLAY_HEADER.read_text(encoding="utf-8")
+        match = re.search(r"kGrapheneOsUseZygoteSpawning = 1LL << (\d+)", header)
+        if match is None:
+            self.fail("kGrapheneOsUseZygoteSpawning missing or changed shape")
+        self.assertEqual(
+            1 << int(match.group(1)),
+            expected,
+            "USE_ZYGOTE_SPAWNING bit changed upstream; update "
+            "kGrapheneOsUseZygoteSpawning and the replay predicate together",
+        )
+
+    def test_replay_predicate_keeps_fd_sentinel(self) -> None:
+        sentinel = self.baseline["replay"]["fds_to_close_initial_sentinel"]
+        header = REPLAY_HEADER.read_text(encoding="utf-8")
+        self.assertIn("is_grapheneos_exec_spawn_replay_contract", header)
+        for i, fd in enumerate(sentinel):
+            self.assertIn(
+                f"fds_to_close[{i}] == {fd}",
+                header,
+                f"replay predicate lost the fds_to_close[{i}] == {fd} sentinel",
+            )
+        self.assertIn(
+            "(native_fork_flags & kGrapheneOsUseZygoteSpawning) == 0",
+            header,
+            "replay predicate no longer requires USE_ZYGOTE_SPAWNING to be unset",
+        )
+
+    def test_generated_jni_hooks_are_fresh(self) -> None:
+        committed = JNI_HOOKS.read_bytes()
+        proc = None
+        with tempfile.TemporaryDirectory() as tmp:
+            shutil.copyfile(GENERATOR, Path(tmp) / GENERATOR.name)
+            proc = subprocess.run(
+                [sys.executable, GENERATOR.name],
+                cwd=tmp,
+                capture_output=True,
+            )
+            generated = (Path(tmp) / "jni_hooks.hpp").read_bytes() if proc.returncode == 0 else b""
+        if proc is None or proc.returncode != 0:
+            stderr = proc.stderr.decode(errors="replace") if proc else "generator did not run"
+            self.fail(f"gen_jni_hooks.py failed: {stderr}")
+        self.assertEqual(
+            committed,
+            generated,
+            "jni_hooks.hpp is stale: regenerate it with "
+            "`cd native/src/core/zygisk && python3 gen_jni_hooks.py`",
+        )
+
+    def test_memfd_file_sepolicy_rule_present(self) -> None:
+        rules = SEPOLICY_RULES.read_text(encoding="utf-8")
+        self.assertIn(
+            'allow(["domain"], [proc], ["memfd_file"]',
+            rules,
+            "Android 17 memfd_file sepolicy rule vanished; Zygisk cannot map "
+            "executable memory on GrapheneOS without it",
+        )
+        for perm in ("getattr", "read", "write", "map", "execute"):
+            self.assertRegex(
+                rules,
+                rf'"memfd_file"\], \[[^\]]*"{perm}"',
+                f"memfd_file rule lost permission '{perm}'",
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_grapheneos_zygote_contract.py b/tests/test_grapheneos_zygote_contract.py
new file mode 100644
index 0000000000000..91b767f79154f
--- /dev/null
+++ b/tests/test_grapheneos_zygote_contract.py
@@ -0,0 +1,216 @@
+from __future__ import annotations
+
+import json
+import shutil
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+from scripts import grapheneos_zygote_contract as contract
+
+FIXTURE = Path(__file__).parent / "fixtures" / "grapheneos_zygote_contract"
+SOURCES = FIXTURE / "sources"
+BASELINE = FIXTURE / "baseline.json"
+
+
+class GrapheneOsZygoteContractTest(unittest.TestCase):
+    def test_extract_contract_from_offline_fixture(self) -> None:
+        current = contract.extract_contract(contract.read_sources(SOURCES))
+
+        self.assertEqual(current, json.loads(BASELINE.read_text(encoding="utf-8")))
+
+    def test_formatting_tolerance_for_jni_descriptor_fragments(self) -> None:
+        native = (SOURCES / "core/jni/com_android_internal_os_Zygote.cpp").read_text(encoding="utf-8")
+        native = native.replace('"nativeForkSystemServer", "(II[II[[IJJ)I"', '"nativeForkSystemServer",\n         "(II[II[[IJJ)I"')
+        sources = contract.read_sources(SOURCES)
+        sources["core/jni/com_android_internal_os_Zygote.cpp"] = native
+
+        current = contract.extract_contract(sources)
+
+        self.assertEqual(current["canonical_sha256"], "a19bec7b1ff9fb0b8211fdfc777e5e8c15a6e386c9fa84cc2b52d94101f26829")
+
+    def test_missing_field_fails_loudly(self) -> None:
+        sources = contract.read_sources(SOURCES)
+        sources["core/java/com/android/internal/os/ZygoteExtraArgs.java"] = sources[
+            "core/java/com/android/internal/os/ZygoteExtraArgs.java"
+        ].replace("int USE_ZYGOTE_SPAWNING = 1 << 3;", "")
+
+        with self.assertRaisesRegex(contract.ContractError, "USE_ZYGOTE_SPAWNING"):
+            contract.extract_contract(sources)
+
+    def test_partial_extra_args_order_fails_loudly(self) -> None:
+        sources = contract.read_sources(SOURCES)
+        sources["core/java/com/android/internal/os/ZygoteExtraArgs.java"] = sources[
+            "core/java/com/android/internal/os/ZygoteExtraArgs.java"
+        ].replace("res[IDX_FLAGS] = flags;", "")
+
+        with self.assertRaisesRegex(contract.ContractError, "Java extra-long-args order"):
+            contract.extract_contract(sources)
+
+    def test_exec_spawning_semantic_flip_fails_loudly(self) -> None:
+        sources = contract.read_sources(SOURCES)
+        sources["core/java/com/android/internal/os/ZygoteExtraArgs.java"] = sources[
+            "core/java/com/android/internal/os/ZygoteExtraArgs.java"
+        ].replace("return !hasFlag(Flag.USE_ZYGOTE_SPAWNING);", "return hasFlag(Flag.USE_ZYGOTE_SPAWNING);")
+
+        with self.assertRaisesRegex(contract.ContractError, "shouldUseExecSpawning contract"):
+            contract.extract_contract(sources)
+
+    def test_socket_assignment_outside_replay_guard_fails_loudly(self) -> None:
+        sources = contract.read_sources(SOURCES)
+        connection = sources["core/java/com/android/internal/os/ZygoteConnection.java"]
+        connection = connection.replace(
+            "if (fd != null) { fdsToClose[0] = fd.getInt$(); }",
+            "if (fd != null) { }",
+        ).replace(
+            "boolean shouldUseExecSpawning",
+            "fdsToClose[0] = fd.getInt$();\n        boolean shouldUseExecSpawning",
+        )
+        sources["core/java/com/android/internal/os/ZygoteConnection.java"] = connection
+
+        with self.assertRaisesRegex(contract.ContractError, "guarded fdsToClose assignments"):
+            contract.extract_contract(sources)
+
+    def test_canonical_hash_is_stable_under_key_order(self) -> None:
+        current = contract.extract_contract(contract.read_sources(SOURCES))
+        shuffled = {"upstream": current["upstream"], "replay": current["replay"], "jni_descriptors": current["jni_descriptors"], "extra_long_args": current["extra_long_args"]}
+
+        self.assertEqual(contract.canonical_hash(shuffled), current["canonical_sha256"])
+
+    def test_source_revision_is_not_semantic_drift(self) -> None:
+        sources = contract.read_sources(SOURCES)
+        first = contract.extract_contract(sources, "a" * 40)
+        second = contract.extract_contract(sources, "b" * 40)
+
+        self.assertEqual(contract.canonical_hash(first), contract.canonical_hash(second))
+
+    def test_drift_report_names_current_and_baseline_hashes(self) -> None:
+        current = contract.extract_contract(contract.read_sources(SOURCES))
+        baseline = json.loads(BASELINE.read_text(encoding="utf-8"))
+        baseline["replay"]["fds_to_close_initial_sentinel"] = [0, -1]
+        baseline["canonical_sha256"] = contract.canonical_hash(baseline)
+
+        report = contract.markdown_report(current, baseline)
+
+        self.assertIn("<!-- grapheneos-zygote-contract-monitor:v1 -->", report)
+        self.assertIn("Semantic contract drift detected", report)
+        self.assertIn(current["canonical_sha256"], report)
+        self.assertIn(baseline["canonical_sha256"], report)
+
+    def test_cli_returns_zero_for_match_and_two_for_drift(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_path = Path(tmp)
+            out_json = tmp_path / "current.json"
+            report = tmp_path / "report.md"
+            self.assertEqual(contract.main_with_args_for_test(SOURCES, BASELINE, out_json, report), 0)
+            drift = tmp_path / "drift.json"
+            shutil.copyfile(BASELINE, drift)
+            data = json.loads(drift.read_text(encoding="utf-8"))
+            data["upstream"]["branch"] = "drift"
+            drift.write_text(json.dumps(data), encoding="utf-8")
+            self.assertEqual(contract.main_with_args_for_test(SOURCES, drift, out_json, report), 2)
+
+    def test_cli_writes_report_when_extraction_fails(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_path = Path(tmp)
+            sources = tmp_path / "sources"
+            shutil.copytree(SOURCES, sources)
+            extra_args = sources / "core/java/com/android/internal/os/ZygoteExtraArgs.java"
+            extra_args.write_text(
+                extra_args.read_text(encoding="utf-8").replace(
+                    "int USE_ZYGOTE_SPAWNING = 1 << 3;", ""
+                ),
+                encoding="utf-8",
+            )
+            report = tmp_path / "report.md"
+            argv = [
+                "grapheneos_zygote_contract.py",
+                "--source-dir",
+                str(sources),
+                "--baseline",
+                str(BASELINE),
+                "--out-json",
+                str(tmp_path / "current.json"),
+                "--report",
+                str(report),
+            ]
+
+            with mock.patch.object(sys, "argv", argv):
+                self.assertEqual(contract.main(), 1)
+
+            self.assertIn("contract extraction failed", report.read_text(encoding="utf-8"))
+
+    def test_workflow_handles_empty_issue_bodies_and_uploads_failure_report(self) -> None:
+        workflow = (
+            Path(__file__).parents[1]
+            / ".github/workflows/grapheneos_zygote_contract_monitor.yml"
+        ).read_text(encoding="utf-8")
+
+        self.assertIn('(.body // \\"\\") | contains', workflow)
+        self.assertIn("if: always()", workflow)
+        self.assertIn("if-no-files-found: warn", workflow)
+        self.assertIn(
+            'elif [ -n "$issue_number" ] && [ "$issue_state" = "OPEN" ]; then',
+            workflow,
+        )
+        self.assertIn('cron: "17 4 1,15 * *"', workflow)
+        self.assertIn("Run contract regression suites", workflow)
+
+    def test_newer_branch_detection(self) -> None:
+        self.assertEqual(contract.newer_upstream_branches(["main", "17", "16", "18"]), ["18"])
+        self.assertEqual(contract.newer_upstream_branches(["main", "17"]), [])
+        self.assertEqual(contract.newer_upstream_branches(["18", "19"]), ["18", "19"])
+        self.assertEqual(contract.newer_upstream_branches([]), [])
+
+    def test_report_flags_newer_upstream_branch(self) -> None:
+        current = contract.extract_contract(contract.read_sources(SOURCES))
+        baseline = json.loads(BASELINE.read_text(encoding="utf-8"))
+
+        clean = contract.markdown_report(current, baseline, newer_branches=[])
+        alert = contract.markdown_report(current, baseline, newer_branches=["18"])
+
+        self.assertIn(f"No upstream Android branch newer than `{contract.BRANCH}` detected", clean)
+        self.assertIn("Newer upstream Android branch detected: 18", alert)
+        self.assertIn("Re-baseline required", alert)
+
+    def test_live_run_alerts_when_newer_branch_exists(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_path = Path(tmp)
+            sources = contract.read_sources(SOURCES)
+            out_json = tmp_path / "current.json"
+            report = tmp_path / "report.md"
+            with (
+                mock.patch.object(contract, "fetch_sources", return_value=(sources, "a" * 40)),
+                mock.patch.object(contract, "check_newer_branches", return_value=["18"]),
+            ):
+                status = contract.run(None, BASELINE, out_json, report, 20.0)
+
+            self.assertEqual(status, 2)
+            self.assertIn("Newer upstream Android branch detected: 18", report.read_text(encoding="utf-8"))
+
+    def test_live_run_survives_watchdog_failure_without_masking_drift_signal(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            tmp_path = Path(tmp)
+            sources = contract.read_sources(SOURCES)
+            out_json = tmp_path / "current.json"
+            report = tmp_path / "report.md"
+            with (
+                mock.patch.object(contract, "fetch_sources", return_value=(sources, "a" * 40)),
+                mock.patch.object(
+                    contract,
+                    "check_newer_branches",
+                    side_effect=contract.ContractError("branches api down"),
+                ),
+            ):
+                status = contract.run(None, BASELINE, out_json, report, 20.0)
+
+            self.assertEqual(status, 0)
+            body = report.read_text(encoding="utf-8")
+            self.assertIn("No semantic contract drift detected.", body)
+            self.assertNotIn("Newer upstream Android branch detected", body)
+
+
+if __name__ == "__main__":
+    unittest.main()