clanker-saddle/test/bundles/canon_deploy.py

"""The canon deploy-hardening bundle (goal 27, after the third review
rejection "the canon never gets installed"): the class of miss was NOT
the install rules or the endpoints — a `meson install --destdir` image
boots and answers the canon API — it was that the running instance had
the canon CODE but no canon DIR (a stale cached package: MANIFEST.usm
1.0 pre-dated canon/, so `usm manifest deploy` satisfied the deploy from
an old artifact). This bundle pins the deploy-time truths:

- the packaged archive REALLY carries the canon files (a tar inspection
  of the .usm package -- skipped when usm is absent)
- the DEPLOYED layout resolves: `meson install --destdir` with a /usr
  prefix, then the app booted FROM THAT TREE (binary at
  <tmp>/usr/bin/clanker-saddle) in a cwd with NO ./canon and no
  CLANKER_CANON_DIR answers /api/saddle/canon with the 15 texts and
  serves one document by slug — the exe-relative installed share is the
  resolver's first non-override candidate
- the startup health line names the resolved directory and count (and
  the empty-shelf shape logs every candidate + the override)
- MANIFEST.usm/meson.build versions moved past 1.0 (the stale-package
  class: a re-deploy must build a fresh artifact)

Mind the machine: XDG_DATA_DIRS must include /usr/share (GLib schemas),
web-config.json exists only through `statum-genkeys --out` (a bare
redirect corrupts it), and the deployed app runs OUTSIDE the harness's
WORK dir with its own scratch db and its own port so nothing collides
with the primary shard app.
"""

import os
import shutil
import subprocess
import tarfile
import tempfile
import urllib.error
import urllib.request

from harness import *  # noqa: F401,F403 — the shared wire helpers


def http_raw(base, method, path, token=None):
    """A raw call against an arbitrary base URL (the deployed tree runs
    on its own port, not the harness's WEB)."""
    request = urllib.request.Request(base + path, method=method)
    if token is not None:
        request.add_header("Authorization", "Bearer " + token)
    try:
        with urllib.request.urlopen(request, timeout=15) as response:
            return response.status, json.loads(response.read().decode() or "{}")
    except urllib.error.HTTPError as error:
        body = error.read().decode()
        try:
            return error.code, json.loads(body)
        except ValueError:
            return error.code, {"raw": body}


def deploy_tree(target):
    """`meson install` the existing build with a /usr prefix into
    <target>/usr; returns (ok, log tail). The builddir may carry meson's
    default /usr/local prefix (a fresh `meson setup` does) — pin /usr so
    the deployed layout is exactly an image-style install: binary at
    <target>/usr/bin, canon at <target>/usr/share/clanker-saddle/canon.
    The reconfigure regenerates but never rebuilds, and the prefix never
    affects the test-run binary's location."""
    configure = subprocess.run(
        ["meson", "configure", os.path.join(REPO, "builddir"), "-Dprefix=/usr"],
        capture_output=True, text=True, timeout=120)
    if configure.returncode != 0:
        print(configure.stdout[-1000:])
        print(configure.stderr[-1000:])
        return False, "meson configure -Dprefix=/usr failed"
    command = ["meson", "install", "-C", os.path.join(REPO, "builddir"),
               "--destdir", target]
    done = subprocess.run(command, capture_output=True, text=True, timeout=300)
    if done.returncode != 0:
        print(done.stdout[-2000:])
        print(done.stderr[-2000:])
        return False, "meson install failed"
    return True, done.stdout[-500:]


def boot_deployed(root, port, binary=None, extra_env=None):
    """Boots the DEPLOYED binary (default root/usr/bin/clanker-saddle) in
    a cwd with no ./canon; returns the process handle (caller
    terminates). `binary` overrides the executed path (the slot-layout
    cases boot a copy whose exe-relative share is absent) and
    `extra_env` rides the scrubbed environment (USMRAD_APPDIR)."""
    run_dir = os.path.join(root, "run")
    os.makedirs(run_dir, exist_ok=True)
    config = json.load(open(os.path.join(REPO, "web-config.json")))
    with open(os.path.join(run_dir, "web-config.json"), "w") as handle:
        json.dump(config, handle)
    env = dict(os.environ)
    env.update({
        "SPRY_DB_PATH": os.path.join(root, "deploy.sqlite"),
        "SPRY_SEED_ADMIN_USERNAME": "admin",
        "SPRY_SEED_ADMIN_PASSWORD": "adminpass123",
        # A registered saddle's bearer token is needed for the canon
        # routes; the deployed app shares nothing with the primary
        # shard, so point its ingest at the shard's fake daemon — its
        # /register handshake delivers a token we can read from the
        # primary app's /saddles slot.  The SADDLE_* knobs keep the
        # poller quiet while the shard tears down.
        "SADDLE_LLM_BASE_URL": "http://127.0.0.1:%d" % DAEMON_PORT,
        "SADDLE_HTTP_TIMEOUT_S": "2",
        "SADDLE_DISPATCH_TICK_S": "1",
        "LD_LIBRARY_PATH": "",
        "XDG_DATA_DIRS": "/usr/share",
    })
    env.pop("CLANKER_CANON_DIR", None)
    # a developer machine whose /app carries a clanker canon must not
    # leak into the non-slot phases; the slot phases set it explicitly
    env.pop("USMRAD_APPDIR", None)
    if extra_env:
        env.update(extra_env)
    log = open(os.path.join(root, "deploy-app.log"), "w")
    process = subprocess.Popen(
        [binary or os.path.join(root, "usr", "bin", "clanker-saddle"),
         str(port)],
        cwd=run_dir, env=env, stdout=log, stderr=subprocess.STDOUT)
    return process


def wait_boot(base, process, timeout=30):
    def ready():
        if process.poll() is not None:
            return "exited-%d" % process.poll()
        try:
            with urllib.request.urlopen(base + "/login", timeout=3) as response:
                return response.status == 200 and "ready" or False
        except (urllib.error.URLError, OSError):
            return False
    return wait_for(ready, timeout=timeout, what="deployed app ready")


def register_saddle(base, auth, name, host):
    """Registers a saddle THROUGH the deployed app (the canon routes
    need its own bearer token) and returns the token."""
    state = slot_public_on(base, auth, "/saddles", "saddles")
    act_on(base, auth, state["register"], {"name": name, "host": host})
    rows = [s for s in slot_public_on(base, auth, "/saddles", "saddles")["saddles"]
            if s["name"] == name]
    return rows[0]["token"]


def act_on(base, auth, action, fields):
    headers = {"X-Statum-Slot": "key=%s; as_at=%s" % (auth["key"], auth["as_at"])}
    if action.get("private"):
        headers["X-Statum-Private"] = action["private"]
    request = urllib.request.Request(
        base + action.get("uri", ""),
        data=urllib.parse.urlencode(fields).encode(), method="POST")
    request.add_header("Content-Type", "application/x-www-form-urlencoded")
    for key, value in headers.items():
        request.add_header(key, value)
    with urllib.request.urlopen(request, timeout=15) as response:
        return json.loads(response.read().decode() or "{}")


def slot_public_on(base, auth, uri, type_name):
    request = urllib.request.Request(
        base + "/_statum/entrypoint?uri=" + urllib.parse.quote(uri.split("?")[0]))
    request.add_header("X-Statum-Slot",
                       "key=%s; as_at=%s" % (auth["key"], auth["as_at"]))
    with urllib.request.urlopen(request, timeout=15) as response:
        frames = json.loads(response.read().decode())
    for directive in frames:
        if directive["type"] == "set":
            content = directive["snapshot_frame"].get("content")
            if not content:
                continue
            content = json.loads(content)
            if content["slot"]["type"] == type_name:
                return content["public"]
    raise SystemExit("no %s slot in deployed entrypoint state for %s"
                     % (type_name, uri))


def login_on(base, username, password, timeout=20):
    """Login against an arbitrary base URL. The deployed app seeds its
    admin from a GLib Idle callback, so /login answering 200 does NOT
    mean the seed landed — poll until the auth slot arrives."""
    deadline = time.monotonic() + timeout
    last_body = ""
    while time.monotonic() < deadline:
        request = urllib.request.Request(base + "/_statum/entrypoint?uri=%2Flogin")
        with urllib.request.urlopen(request, timeout=15) as response:
            frames = json.loads(response.read().decode())
        content = json.loads(frames[0]["snapshot_frame"]["content"])
        action = content["public"]["login"]
        request = urllib.request.Request(
            base + action.get("uri", ""),
            data=urllib.parse.urlencode(
                {"username": username, "password": password}).encode(),
            method="POST")
        request.add_header("Content-Type", "application/x-www-form-urlencoded")
        try:
            with urllib.request.urlopen(request, timeout=15) as response:
                body = response.read().decode()
        except urllib.error.HTTPError as error:
            last_body = error.read().decode()
            time.sleep(0.2)
            continue
        for directive in json.loads(body):
            if directive["type"] == "set":
                frame = json.loads(directive["snapshot_frame"]["content"])
                if frame["slot"]["type"] == "auth":
                    return {"key": frame["slot"]["key"], "as_at": frame["as_at"]}
    raise SystemExit("deployed login response carried no auth slot (last: %s)"
                     % last_body[:300])


def run(ctx):
    auth, fake, fake_token = ctx.auth, ctx.fake, ctx.fake_token

    print("== deploy: the manifest versions moved past 1.0 ==")
    manifest = json.load(open(os.path.join(REPO, "MANIFEST.usm")))
    check(manifest.get("version") != "1.0"
          and manifest.get("version") == "1.1",
          "MANIFEST.usm carries the bumped version 1.1 (%s)" % manifest.get("version"))

    print("== deploy: the packaged archive carries the canon ==")
    # `usm manifest package` writes ../<name>-<version>.usmc relative to
    # the repo. Building it here keeps the check honest (a stale cached
    # package is exactly the miss class this bundle pins).
    usm = shutil.which("usm")
    archive = None
    if usm is None:
        print("  (usm not on PATH — asserting the install rule instead)")
    else:
        done = subprocess.run([usm, "manifest", "package"],
                              cwd=REPO, capture_output=True, text=True, timeout=600)
        if done.returncode != 0:
            print(done.stdout[-1000:])
            print(done.stderr[-1000:])
        check(done.returncode == 0, "usm manifest package succeeds")
        archive = os.path.join(os.path.dirname(REPO),
                               "clanker-saddle-%s.usmc" % manifest["version"])
        check(os.path.exists(archive),
              "the .usmc lands beside the repo (%s)" % archive)
        if os.path.exists(archive):
            with tarfile.open(archive) as bundle:
                names = bundle.getnames()
                canon_files = [n for n in names
                               if n.startswith("./canon/") and n.endswith(".md")]
                check(len(canon_files) >= 15,
                      "the archive carries the 15 canon .md files (%d)"
                      % len(canon_files))
                check(any(n.endswith("gnu-manifesto.md") for n in canon_files)
                      and any(n.endswith("magnifica-humanitas-summary.md")
                              for n in canon_files),
                      "the archive's canon includes the anchor texts")
                # The version INSIDE the archive must be the bumped one
                # (the stale-artifact class: an old 1.0 archive would
                # satisfy a deploy with no canon at all).
                member = bundle.extractfile("./MANIFEST.usm")
                inside = json.loads(member.read().decode())
                check(inside.get("version") == manifest["version"],
                      "the archive's MANIFEST.usm carries version %s (%s)"
                      % (manifest["version"], inside.get("version")))

    print("== deploy: the deployed /usr layout serves the canon ==")
    root = tempfile.mkdtemp(prefix="canon-deploy-")
    deployed_port = free_port()
    deployed_base = "http://127.0.0.1:%d" % deployed_port
    ok, _ = deploy_tree(root)
    check(ok, "meson install --destdir <tmp>/usr with a /usr prefix")
    binary = os.path.join(root, "usr", "bin", "clanker-saddle")
    share = os.path.join(root, "usr", "share", "clanker-saddle", "canon")
    check(os.path.exists(binary), "the installed binary is at <tmp>/usr/bin")
    installed_md = [n for n in (os.listdir(share) if os.path.isdir(share) else [])
                    if n.endswith(".md")]
    check(len(installed_md) >= 15,
          "the installed share/clanker-saddle/canon carries 15 texts (%d)"
          % len(installed_md))
    if not ok or not os.path.exists(binary):
        shutil.rmtree(root, ignore_errors=True)
        return

    process = boot_deployed(root, deployed_port)
    try:
        outcome = wait_boot(deployed_base, process)
        check(outcome == "ready",
              "the deployed tree boots from a canon-less cwd (%s)" % outcome)
        if outcome != "ready":
            return
        deployed_auth = login_on(deployed_base, "admin", "adminpass123")
        token = register_saddle(deployed_base, deployed_auth,
                                "deploy-probe", "127.0.0.1:%d" % fake.port)
        check(bool(token), "a saddle registered through the deployed app")

        status, body = http_raw(deployed_base, "GET", "/api/saddle/canon", token)
        check(status == 200 and body.get("ok") is True,
              "GET /api/saddle/canon answers ok from the deployed tree")
        texts = body.get("texts") or []
        check(body.get("count") == 15 and len(texts) == 15,
              "the deployed canon lists all 15 texts (%s)" % body.get("count"))
        check(os.path.abspath(body.get("source_dir") or "") == share,
              "the deployed canon resolved the installed share dir (%s)"
              % body.get("source_dir"))
        status, doc = http_raw(deployed_base, "GET",
                               "/api/saddle/canon/free-software-definition", token)
        check(status == 200 and doc.get("ok") is True
              and doc.get("slug") == "free-software-definition"
              and "freedom" in (doc.get("body") or "").lower(),
              "the deployed canon serves one document by slug")

        print("== deploy: the startup health line ==")
        log_text = open(os.path.join(root, "deploy-app.log")).read()
        health = [line for line in log_text.splitlines()
                  if "Canon: 15 texts from" in line]
        logged = (health[0].split(" from ", 1)[1].strip()
                  if health else "")
        check(bool(health) and os.path.normpath(logged) == share,
              "the startup log names the resolved canon dir and count (%s)"
              % (health[:1] or ["absent"]))

        print("== deploy: the hero carries the canon count ==")
        dash = slot_public_on(deployed_base, deployed_auth, "/", "dashboard")
        check(int(dash.get("canon_texts", -1)) == 15,
              "the dashboard's public data carries canon_texts = 15 (%s)"
              % dash.get("canon_texts"))
    finally:
        process.terminate()
        try:
            process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            process.kill()

    print("== deploy: the empty shelf is loud ==")
    # Boot the deployed binary in a cwd with no ./canon, no override,
    # and the installed share moved AWAY — the loud warning naming every
    # candidate is the deploying admin's only symptom, so it must be in
    # the log. (The absolute /usr/share and /usr/local/share candidates
    # do not exist on this machine — no install outside the tmpdir.)
    empty_root = tempfile.mkdtemp(prefix="canon-empty-")
    shutil.move(share, os.path.join(empty_root, "canon-parked"))
    empty_port = free_port()
    process = boot_deployed(root, empty_port)
    try:
        empty_base = "http://127.0.0.1:%d" % empty_port
        outcome = wait_boot(empty_base, process)
        check(outcome == "ready", "the canon-less deployed tree still boots (%s)"
              % outcome)
        log_text = open(os.path.join(root, "deploy-app.log")).read()
        check("no canon directory resolved" in log_text
              and "CLANKER_CANON_DIR" in log_text,
              "the empty shelf logs the loud warning naming the override")
        check("/usr/share/clanker-saddle/canon" in log_text
              and "/usr/local/share/clanker-saddle/canon" in log_text,
              "the warning names the absolute system share candidates")
        empty_auth = login_on(empty_base, "admin", "adminpass123")
        dash = slot_public_on(empty_base, empty_auth, "/", "dashboard")
        check(int(dash.get("canon_texts", -1)) == 0,
              "the canon-less dashboard reports canon_texts = 0 (%s)"
              % dash.get("canon_texts"))
        token = register_saddle(empty_base, empty_auth,
                                "empty-probe", "127.0.0.1:%d" % fake.port)
        status, body = http_raw(empty_base, "GET", "/api/saddle/canon", token)
        check(status == 200 and body.get("count") == 0,
              "the canon-less API answers an empty library, not an error")
    finally:
        process.terminate()
        try:
            process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            process.kill()
        shutil.move(os.path.join(empty_root, "canon-parked"), share)
        shutil.rmtree(empty_root, ignore_errors=True)
        shutil.rmtree(root, ignore_errors=True)

    print("== deploy: the manifest provides the canon as resources ==")
    # The slot-fallback miss class: usm's manifest install copies ONLY
    # the declared provides resources into the install root, so a
    # manifest providing just bin:clanker-saddle deploys a slot with
    # the binary and no canon at all — the running instance answers an
    # empty library no matter what the resolver tries. Every canon
    # file must be declared as a res: resource (the grammar usm's own
    # autoprovides emits for share data).
    manifest = json.load(open(os.path.join(REPO, "MANIFEST.usm")))
    canon_files = sorted(
        name for name in os.listdir(os.path.join(REPO, "canon"))
        if name.endswith(".md"))
    missing = [name for name in canon_files
               if "res:clanker-saddle/canon/%s" % name
               not in manifest.get("provides", {})]
    check(len(canon_files) >= 15,
          "the source canon carries the texts (%d)" % len(canon_files))
    check(not missing,
          "MANIFEST.usm provides every canon .md as a res: resource "
          "(missing: %s)" % (missing or "none"))

    print("== deploy: the usmrad slot layout serves the canon ==")
    # The deployed usmrad slot: <appdir>/<slot>/usr/bin + usr/share,
    # with a `current` symlink flipped onto the fresh slot. usmrad
    # launches the binary THROUGH that link, so /proc/self/exe sits in
    # <appdir>/current/usr/bin and the exe-relative climb finds the
    # share. This phase pins that whole layout end-to-end.
    slot_root = tempfile.mkdtemp(prefix="canon-slot-")
    appdir = os.path.join(slot_root, "app")
    ok, _ = deploy_tree(slot_root)
    check(ok, "meson install --destdir for the slot layout")
    # arrange the usmrad shape: app/a/usr/… + current -> a
    os.makedirs(os.path.join(appdir, "a"), exist_ok=True)
    shutil.move(os.path.join(slot_root, "usr"), os.path.join(appdir, "a", "usr"))
    os.symlink("a", os.path.join(appdir, "current"))
    slot_share = os.path.join(appdir, "a", "usr", "share",
                              "clanker-saddle", "canon")
    check(os.path.isdir(slot_share),
          "the slot carries usr/share/clanker-saddle/canon")
    process = boot_deployed(
        slot_root, deployed_port,
        binary=os.path.join(appdir, "current", "usr", "bin", "clanker-saddle"),
        extra_env={"USMRAD_APPDIR": appdir})
    try:
        slot_base = "http://127.0.0.1:%d" % deployed_port
        outcome = wait_boot(slot_base, process)
        check(outcome == "ready", "the slot binary boots through `current` (%s)"
              % outcome)
        if outcome != "ready":
            return
        slot_auth = login_on(slot_base, "admin", "adminpass123")
        token = register_saddle(slot_base, slot_auth,
                                "slot-probe", "127.0.0.1:%d" % fake.port)
        status, body = http_raw(slot_base, "GET", "/api/saddle/canon", token)
        check(status == 200 and body.get("count") == 15,
              "the slot layout lists the 15 texts (%s)" % body.get("count"))
        status, doc = http_raw(slot_base, "GET",
                               "/api/saddle/canon/gnu-manifesto", token)
        check(status == 200 and doc.get("ok") is True
              and "GNU" in (doc.get("title") or ""),
              "the slot layout serves a document by slug")
    finally:
        process.terminate()
        try:
            process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            process.kill()
        shutil.rmtree(slot_root, ignore_errors=True)

    print("== deploy: the slot fallback when the exe share is absent ==")
    # The mandate case: the canon exists ONLY in the deployment slot's
    # data directory and the system path misses (no /usr/share canon,
    # no exe-relative share). The resolver must fall back to
    # <appdir>/<slot>/usr/share/… resolved through $USMRAD_APPDIR —
    # the exe cannot anchor it once its share neighbours are gone.
    bare_root = tempfile.mkdtemp(prefix="canon-bare-")
    appdir = os.path.join(bare_root, "app")
    ok, _ = deploy_tree(bare_root)
    check(ok, "meson install --destdir for the bare-exe layout")
    os.makedirs(os.path.join(appdir, "b"), exist_ok=True)
    # the slot: share under app/b/usr, the binary copied WITHOUT its
    # share neighbours (only the file, no usr tree beside it)
    shutil.move(os.path.join(bare_root, "usr", "share"),
                os.path.join(appdir, "b", "usr-tmp"))
    os.makedirs(os.path.join(appdir, "b", "usr"), exist_ok=True)
    shutil.move(os.path.join(appdir, "b", "usr-tmp"),
                os.path.join(appdir, "b", "usr", "share"))
    os.makedirs(os.path.join(appdir, "bin"), exist_ok=True)
    shutil.copy2(os.path.join(bare_root, "usr", "bin", "clanker-saddle"),
                 os.path.join(appdir, "bin", "clanker-saddle"))
    shutil.rmtree(os.path.join(bare_root, "usr"))
    os.symlink("b", os.path.join(appdir, "current"))
    slot_share = os.path.join(appdir, "b", "usr", "share",
                              "clanker-saddle", "canon")
    check(os.path.isdir(slot_share), "the slot carries the canon data dir")
    process = boot_deployed(
        bare_root, deployed_port,
        binary=os.path.join(appdir, "bin", "clanker-saddle"),
        extra_env={"USMRAD_APPDIR": appdir})
    try:
        bare_base = "http://127.0.0.1:%d" % deployed_port
        outcome = wait_boot(bare_base, process)
        check(outcome == "ready", "the bare-exe slot binary boots (%s)"
              % outcome)
        if outcome != "ready":
            return
        log_text = open(os.path.join(bare_root, "deploy-app.log")).read()
        health = [line for line in log_text.splitlines()
                  if "Canon: 15 texts from" in line]
        logged = (health[0].split(" from ", 1)[1].strip()
                  if health else "")
        check(bool(health) and os.path.realpath(logged)
              == os.path.realpath(slot_share),
              "the bare exe resolves the canon from the slot data dir (%s)"
              % (health[:1] or ["absent"]))
        bare_auth = login_on(bare_base, "admin", "adminpass123")
        token = register_saddle(bare_base, bare_auth,
                                "bare-probe", "127.0.0.1:%d" % fake.port)
        status, body = http_raw(bare_base, "GET", "/api/saddle/canon", token)
        check(status == 200 and body.get("count") == 15,
              "the slot-only canon lists the 15 texts (%s)" % body.get("count"))
        check(os.path.realpath(body.get("source_dir") or "")
              == os.path.realpath(slot_share),
              "the API reports the slot data dir as the source (%s)"
              % body.get("source_dir"))
        status, doc = http_raw(bare_base, "GET",
                               "/api/saddle/canon/free-software-definition",
                               token)
        check(status == 200 and doc.get("ok") is True
              and "freedom" in (doc.get("body") or "").lower(),
              "the slot-only canon serves one document by slug")
        # the warning must NOT fire: the fallback resolved a shelf
        check("no canon directory resolved" not in log_text,
              "the slot fallback resolves without the empty-shelf warning")
    finally:
        process.terminate()
        try:
            process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            process.kill()
        shutil.rmtree(bare_root, ignore_errors=True)

    print("== deploy: the higher-precedence share wins over the slot ==")
    # The precedence half of the mandate: when the canon exists in a
    # HIGHER-precedence location (the system path — represented here
    # by the exe-relative share, which outranks the slot in exactly
    # the same way the absolute /usr|/usr/local shares do) AND in the
    # slot, the higher-precedence shelf is served — the slot fallback
    # must never shadow it. A one-text marker shelf distinguishes them.
    prec_root = tempfile.mkdtemp(prefix="canon-prec-")
    appdir = os.path.join(prec_root, "app")
    ok, _ = deploy_tree(prec_root)
    check(ok, "meson install --destdir for the precedence layout")
    os.makedirs(os.path.join(appdir, "a"), exist_ok=True)
    shutil.move(os.path.join(prec_root, "usr", "share"),
                os.path.join(appdir, "a", "usr-tmp"))
    os.makedirs(os.path.join(appdir, "a", "usr"), exist_ok=True)
    shutil.move(os.path.join(appdir, "a", "usr-tmp"),
                os.path.join(appdir, "a", "usr", "share"))
    os.makedirs(os.path.join(appdir, "bin"), exist_ok=True)
    shutil.copy2(os.path.join(prec_root, "usr", "bin", "clanker-saddle"),
                 os.path.join(appdir, "bin", "clanker-saddle"))
    # the higher-precedence shelf: the exe-relative share beside the
    # binary, holding ONE marker text
    exe_share = os.path.join(appdir, "bin", "..", "share",
                             "clanker-saddle", "canon")
    os.makedirs(exe_share, exist_ok=True)
    with open(os.path.join(exe_share, "system-marker.md"), "w") as handle:
        handle.write(
            "---\nslug: system-marker\ntitle: The System Path\n"
            "author: test\nsource: test\nlicence: test\n"
            "summary: marks the higher-precedence shelf\n---\n\nsystem\n")
    os.symlink("a", os.path.join(appdir, "current"))
    process = boot_deployed(
        prec_root, deployed_port,
        binary=os.path.join(appdir, "bin", "clanker-saddle"),
        extra_env={"USMRAD_APPDIR": appdir})
    try:
        prec_base = "http://127.0.0.1:%d" % deployed_port
        outcome = wait_boot(prec_base, process)
        check(outcome == "ready", "the precedence binary boots (%s)" % outcome)
        if outcome != "ready":
            return
        prec_auth = login_on(prec_base, "admin", "adminpass123")
        token = register_saddle(prec_base, prec_auth,
                                "prec-probe", "127.0.0.1:%d" % fake.port)
        status, body = http_raw(prec_base, "GET", "/api/saddle/canon", token)
        check(status == 200
              and os.path.abspath(body.get("source_dir") or "")
              == os.path.normpath(exe_share),
              "the higher-precedence share is served, not the slot (%s)"
              % body.get("source_dir"))
        check(body.get("count") == 1
              and [t.get("slug") for t in (body.get("texts") or [])]
              == ["system-marker"],
              "the served shelf is the higher-precedence one (%s texts)"
              % body.get("count"))
    finally:
        process.terminate()
        try:
            process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            process.kill()
        shutil.rmtree(prec_root, ignore_errors=True)