diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index f35b52c..d577182 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -159,8 +159,9 @@ jobs:
       - name: Verify fresh Python-Go credential interoperability
         if: ${{ matrix.go_tests == true }}
         env:
+          ARBITER_GO_CLIENT_SMOKE_OUTDIR: .ci/go-client-smoke
           GOCACHE: ${{ runner.temp }}/go-build
-        run: nox -s unit -- server/tests/integration/test_credential_peer.py server/tests/integration/test_provisional_peer.py server/tests/integration/test_pairing_peer.py server/tests/integration/test_enrollment_cli_peer.py server/tests/integration/test_control_tls_peer.py server/tests/integration/test_enrollment_receipt_peer.py server/tests/integration/test_activation_restart_peer.py
+        run: nox -s unit -- server/tests/integration/test_credential_peer.py server/tests/integration/test_provisional_peer.py server/tests/integration/test_pairing_peer.py server/tests/integration/test_enrollment_cli_peer.py server/tests/integration/test_control_tls_peer.py server/tests/integration/test_enrollment_receipt_peer.py server/tests/integration/test_activation_restart_peer.py server/tests/integration/test_installed_enrollment.py
 
       - name: Verify native local-control rejection across OS users
         if: ${{ startsWith(matrix.platform_id, 'linux-') || startsWith(matrix.platform_id, 'macos-') }}
diff --git a/server/news/74.misc.md b/server/news/74.misc.md
new file mode 100644
index 0000000..859c635
--- /dev/null
+++ b/server/news/74.misc.md
@@ -0,0 +1,2 @@
+Add installed native-client pairing and receipt-recovery acceptance coverage,
+including fragmented proxy writes and concurrent expiry/shutdown ownership.
diff --git a/server/tests/integration/enrollment_terminal_driver.py b/server/tests/integration/enrollment_terminal_driver.py
new file mode 100644
index 0000000..a75521e
--- /dev/null
+++ b/server/tests/integration/enrollment_terminal_driver.py
@@ -0,0 +1,163 @@
+"""Test-only human terminal driver for an unmodified installed executable."""
+
+import json
+import os
+import re
+import select
+import socket
+import struct
+import subprocess
+import sys
+import time
+from typing import Any
+
+
+def run():
+    config = json.loads(sys.stdin.readline())
+    master = slave = input_fd = output_fd = -1
+    terminal_options: dict[str, Any]
+    if sys.platform == "win32":
+        import ctypes
+        from ctypes import wintypes
+        import msvcrt
+
+        kernel = ctypes.WinDLL("kernel32", use_last_error=True)
+        read = kernel.ReadConsoleOutputCharacterW
+        read.argtypes = [
+            wintypes.HANDLE,
+            wintypes.LPWSTR,
+            wintypes.DWORD,
+            wintypes.DWORD,
+            ctypes.POINTER(wintypes.DWORD),
+        ]
+        write = kernel.WriteConsoleInputW
+        write.argtypes = [
+            wintypes.HANDLE,
+            wintypes.LPVOID,
+            wintypes.DWORD,
+            ctypes.POINTER(wintypes.DWORD),
+        ]
+        input_fd = os.open("CONIN$", os.O_RDWR)
+        output_fd = os.open("CONOUT$", os.O_RDWR)
+
+        def terminal_read():
+            buffer = ctypes.create_unicode_buffer(8192)
+            size = wintypes.DWORD()
+            if not read(
+                msvcrt.get_osfhandle(output_fd), buffer, 8192, 0, ctypes.byref(size)
+            ):
+                raise OSError("could not inspect test console")
+            return buffer.value.encode()
+
+        def enter():
+            # INPUT_RECORD(KEY_EVENT): down/up, one RETURN, no secret input.
+            records = b"".join(
+                struct.pack("<H2xiHHHHI", 1, down, 1, 13, 0, 13, 0) for down in (1, 0)
+            )
+            buffer = ctypes.create_string_buffer(records)
+            size = wintypes.DWORD()
+            if not write(msvcrt.get_osfhandle(input_fd), buffer, 2, ctypes.byref(size)):
+                raise OSError("could not confirm test console")
+
+        terminal_options = {}
+    else:
+        import fcntl
+        import pty
+        import termios
+
+        master, slave = pty.openpty()
+
+        def terminal_read():
+            if select.select([master], [], [], 0)[0]:
+                return os.read(master, 4096)
+            return b""
+
+        def enter():
+            os.write(master, b"\n")
+
+        terminal_options = dict(
+            start_new_session=True,
+            pass_fds=(slave,),
+            preexec_fn=lambda: fcntl.ioctl(slave, termios.TIOCSCTTY, 0),
+        )
+    command = [
+        config["command"],
+        "--config-dir",
+        config["directory"],
+        "arbiter.connect_timeout=15s",
+        "enrol",
+        "url=" + config["url"],
+        "name=Installed CLI",
+    ]
+    process: subprocess.Popen[bytes] | None = None
+    try:
+        process = subprocess.Popen(
+            command,
+            stdin=subprocess.DEVNULL,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=False,
+            **terminal_options,
+        )
+        terminal, handed_off = b"", False
+        selector = secret = ""
+        deadline = time.monotonic() + 45
+        with socket.create_connection(
+            ("127.0.0.1", config["control"]), timeout=10
+        ) as control:
+            control.settimeout(15)
+            with control.makefile("rwb") as stream:
+                while process.poll() is None and time.monotonic() < deadline:
+                    data = terminal_read()
+                    terminal = data if sys.platform == "win32" else terminal + data
+                    if len(terminal) > 32768:
+                        raise RuntimeError("unexpected terminal output volume")
+                    match = re.search(
+                        rb"Request: ([A-Za-z0-9_-]{22})\s+Secret[^:]*: ([A-Za-z0-9_-]{43})",
+                        terminal,
+                    )
+                    if (
+                        not handed_off
+                        and match
+                        and b"press Enter to continue: " in terminal
+                    ):
+                        selector, secret = (part.decode() for part in match.groups())
+                        stream.write(
+                            json.dumps(dict(selector=selector, secret=secret)).encode()
+                            + b"\n"
+                        )
+                        stream.flush()
+                        if stream.readline(16) != b"true\n":
+                            raise RuntimeError("operator did not approve")
+                        enter()
+                        handed_off = True
+                    time.sleep(0.01)
+        if process.poll() is None:
+            raise TimeoutError("installed enrollment did not settle")
+        stdout, stderr = process.communicate(timeout=5)
+        if not handed_off or secret.encode() in stdout + stderr:
+            raise RuntimeError("invalid protected terminal handoff")
+        confirmed = (
+            process.returncode == 0 and b"Enrollment activated and saved" in stdout
+        )
+        if not confirmed and not (
+            config["recover"]
+            and process.returncode == 1
+            and ("resume=" + selector).encode() in stderr
+        ):
+            raise RuntimeError(
+                "installed enrollment did not confirm or retain recovery"
+            )
+        # Only public completion metadata leaves the synthetic human driver.
+        print(json.dumps(dict(selector=selector, activated=confirmed)))
+    finally:
+        if process is not None and process.poll() is None:
+            process.kill()
+            process.wait(timeout=5)
+        for fd in (input_fd, output_fd, master, slave):
+            if fd >= 0:
+                os.close(fd)
+
+
+if __name__ == "__main__":
+    run()
diff --git a/server/tests/integration/test_installed_enrollment.py b/server/tests/integration/test_installed_enrollment.py
new file mode 100644
index 0000000..0e2b5a7
--- /dev/null
+++ b/server/tests/integration/test_installed_enrollment.py
@@ -0,0 +1,225 @@
+"""Installed platform wheel, real terminal handoff and opaque proxy writes."""
+
+import base64
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import closing, contextmanager
+import json
+from pathlib import Path
+import select
+import socket
+import socketserver
+import sqlite3
+import ssl
+import subprocess
+import sys
+import threading
+import urllib.request
+from typing import cast
+
+import pytest
+from starlette.applications import Starlette
+
+from arbiter_server.authority import initialize_authority
+from arbiter_server.main import _generate_self_signed_tls_certificate
+from server.tests.integration.test_cli_entrypoint import (
+    ClientCommand,
+    current_go_client_binary,
+    packaged_arbiter_client_command,
+)
+from server.tests.unit.conftest import private_windows_test_parent, tmp_path
+from server.tests.unit.test_artifact_http_admission import live_server
+from server.tests.unit.test_enrollment_activation import database
+from server.tests.unit.test_native_http_protocol import _client
+
+
+@contextmanager
+def fragmented_passthrough(port):
+    stopped = threading.Event()
+    writes = []
+
+    class Relay(socketserver.BaseRequestHandler):
+        def handle(self):
+            self.request.settimeout(5)
+            with socket.create_connection(("127.0.0.1", port), timeout=5) as upstream:
+                peers = (self.request, upstream)
+                try:
+                    while not stopped.is_set():
+                        ready, _, _ = select.select(peers, [], [], 0.1)
+                        for source in ready:
+                            data = source.recv(2048)
+                            if not data:
+                                return
+                            destination = peers[1] if source is peers[0] else peers[0]
+                            # The relay sees outer TLS ciphertext only. These
+                            # split writes do not claim fixed TCP packet sizes.
+                            for offset in range(0, len(data), 97):
+                                piece = data[offset : offset + 97]
+                                destination.sendall(piece)
+                                writes.append((source is peers[0], len(piece)))
+                except (ConnectionError, OSError):
+                    return  # Peer cancellation is normal during lost replies.
+
+    with socketserver.ThreadingTCPServer(("127.0.0.1", 0), Relay) as proxy:
+        proxy.daemon_threads = False
+        thread = threading.Thread(target=proxy.serve_forever)
+        thread.start()
+        try:
+            yield proxy.server_address[1], writes
+        finally:
+            stopped.set()
+            proxy.shutdown()
+            thread.join(timeout=5)
+            assert not thread.is_alive()
+
+
+@pytest.mark.parametrize("mode", ["ok", "lost-reply", "promotion-failure"])
+def test_installed_enrollment_and_resume(
+    tmp_path,
+    private_windows_test_parent,
+    packaged_arbiter_client_command: ClientCommand,
+    monkeypatch,
+    mode,
+):
+    root = tmp_path / "server" / "authority"
+    initialize_authority(root, recovery_file=tmp_path / "operator.json")
+    app = cast(Starlette, _client(tmp_path)._app)
+    responder, requests = app.state.pairing_responder, app.state.enrollment_requests
+    original_activate = responder.activate
+    committed = []
+
+    async def activate(*args):
+        receipt = await original_activate(*args)
+        committed.append(True)
+        if mode == "lost-reply":
+            raise ValueError("synthetic loss after actual activation commit")
+        return receipt
+
+    monkeypatch.setattr(responder, "activate", activate)
+    certificate, key = tmp_path / "outer.crt", tmp_path / "outer.key"
+    _generate_self_signed_tls_certificate(
+        cert_path=certificate,
+        key_path=key,
+        names=("127.0.0.1",),
+    )
+    directory = tmp_path / "installed-custody"
+    directory.mkdir(mode=0o700)
+    marker = directory / "enrolled"
+    if mode == "promotion-failure":
+        marker.write_text("preserve unrelated existing file")
+    with (
+        live_server(
+            app,
+            ws=app.state.native_websocket_protocol,
+            http=app.state.native_http_protocol,
+            ssl_certfile=str(certificate),
+            ssl_keyfile=str(key),
+        ) as port,
+        fragmented_passthrough(port) as (proxy_port, writes),
+    ):
+        requests.open_window()
+        with socket.socket() as listener:
+            listener.bind(("127.0.0.1", 0))
+            listener.listen(1)
+            listener.settimeout(30)
+
+            def approve():
+                connection, _ = listener.accept()
+                with connection, connection.makefile("rwb") as stream:
+                    connection.settimeout(20)
+                    handoff = json.loads(stream.readline(1025))
+                    pending = next(
+                        p
+                        for p in requests.pending()
+                        if p.advertisement.selector == handoff["selector"]
+                    )
+                    requests.control(
+                        "approve",
+                        dict(
+                            selector=pending.advertisement.selector,
+                            revision=pending.revision,
+                            request_digest=pending.advertisement.digest,
+                            psk=handoff["secret"],
+                            grants=[],
+                        ),
+                        set(),
+                    )
+                    stream.write(b"true\n")
+                    stream.flush()
+
+            config = dict(
+                command=str(packaged_arbiter_client_command.command),
+                directory=str(directory),
+                url=f"https://127.0.0.1:{proxy_port}",
+                control=listener.getsockname()[1],
+                recover=mode == "promotion-failure",
+            )
+            with ThreadPoolExecutor(max_workers=1) as pool:
+                owner = pool.submit(approve)
+                creationflags = 0
+                if sys.platform == "win32":
+                    creationflags = subprocess.CREATE_NEW_CONSOLE
+                result = subprocess.run(
+                    [
+                        sys.executable,
+                        str(Path(__file__).with_name("enrollment_terminal_driver.py")),
+                    ],
+                    input=json.dumps(config) + "\n",
+                    text=True,
+                    capture_output=True,
+                    timeout=65,
+                    creationflags=creationflags,
+                )
+                assert (
+                    result.returncode == 0
+                ), "installed native terminal journey failed"
+                owner.result(timeout=5)
+            completion = json.loads(result.stdout)
+        selector = completion["selector"]
+        assert completion["activated"] == (mode != "promotion-failure")
+        assert committed == [True]
+        if mode == "promotion-failure":
+            assert marker.read_text() == "preserve unrelated existing file"
+            before = (directory / "prepared" / (selector + ".json")).read_bytes()
+            marker.unlink()  # Explicit operator repair of this test-owned collision.
+        else:
+            before = (directory / "enrolled" / (selector + ".json")).read_bytes()
+        for _ in range(2):
+            resumed = subprocess.run(
+                [
+                    config["command"],
+                    "--config-dir",
+                    str(directory),
+                    "arbiter.connect_timeout=15s",
+                    "enrol",
+                    "resume=" + selector,
+                ],
+                stdin=subprocess.DEVNULL,
+                capture_output=True,
+                text=True,
+                timeout=20,
+            )
+            assert resumed.returncode == 0, "installed same-selector recovery failed"
+            assert "Enrollment activated and saved" in resumed.stdout
+        enrolled = directory / "enrolled" / (selector + ".json")
+        assert enrolled.read_bytes() == before
+        assert list((directory / "enrolled").glob("*.json")) == [enrolled]
+        assert not list((directory / "prepared").glob("*.json"))
+        saved = json.loads(before)
+        assert set(saved["provisional"]) >= {
+            "access_key_pkcs8",
+            "portable_key_pkcs8",
+            "recipient_key_pkcs8",
+        }
+        with closing(sqlite3.connect(database(root))) as db:
+            rows = db.execute("SELECT prepared FROM enrollments").fetchall()
+            assert len(rows) == 1
+            assert bytes(rows[0][0]) == base64.b64decode(saved["result"], validate=True)
+        assert not requests.pending() and not responder._attempts
+        assert writes and {direction for direction, _ in writes} == {False, True}
+        assert all(0 < size <= 97 for _, size in writes)
+        with urllib.request.urlopen(
+            f"https://127.0.0.1:{port}/_health_",
+            context=ssl._create_unverified_context(),
+            timeout=5,
+        ) as health:
+            assert health.status == 200 and json.load(health) == {"status": "ok"}
diff --git a/server/tests/unit/test_pairing_carrier.py b/server/tests/unit/test_pairing_carrier.py
index 77d0fa1..9727645 100644
--- a/server/tests/unit/test_pairing_carrier.py
+++ b/server/tests/unit/test_pairing_carrier.py
@@ -473,6 +473,89 @@ def test_real_server_shutdown_closes_idle_native_carrier(server):
         peer.close()
 
 
+def test_simultaneous_expiry_shutdown_retains_two_real_workers(server, monkeypatch):
+    from noise.connection import NoiseConnection
+
+    entered = threading.Barrier(3)
+    release, stopped = threading.Event(), threading.Event()
+    original_read = NoiseConnection.read_message
+    original_connection = PairingProtocol.connection_made
+    carriers = []
+
+    def blocked(noise, data):
+        if threading.current_thread().name == "arbiter-operation":
+            entered.wait(timeout=10)
+            assert release.wait(20)
+        return original_read(noise, data)
+
+    def connected(protocol, transport):
+        original_connection(protocol, transport)
+        if not protocol.closed:
+            carriers.append(protocol)
+
+    monkeypatch.setattr(NoiseConnection, "read_message", blocked)
+    monkeypatch.setattr(PairingProtocol, "connection_made", connected)
+    peers = []
+    with server(max_connections=2, max_handshakes=2) as (app, port, first, path):
+        requests, responder = app.state.enrollment_requests, app.state.pairing_responder
+        pending = requests.register(json.dumps(advertisement()).encode())
+        requests.control(
+            "approve",
+            dict(
+                selector=pending.advertisement.selector,
+                revision=pending.revision,
+                request_digest=pending.advertisement.digest,
+                psk=base64.urlsafe_b64encode(os.urandom(32)).decode().rstrip("="),
+                grants=[],
+            ),
+            set(),
+        )
+        second = requests.approved(pending.advertisement.selector)
+        try:
+            for context in (first, second):
+                peer = Peer(
+                    port,
+                    PAIRING_PATH + context.request.advertisement.selector + "/pair",
+                )
+                peers.append(peer)
+                assert isinstance(peer.next()[0], events.AcceptConnection)
+                _, flight = initiator(context)
+                peer.send(events.BytesMessage(data=flight))
+            entered.wait(timeout=10)
+            with Peer(port, path) as excess:
+                rejected = excess.next()[0]
+                assert isinstance(rejected, events.RejectConnection)
+                assert rejected.status_code == 503
+            assert wire_request(port, "GET", "/_health_")[0] == 200
+
+            def expire_and_stop():
+                requests._clock = lambda: pending.deadline + 1
+                assert not requests.pending()
+                for carrier in carriers:
+                    carrier.shutdown()
+                responder.close()
+                stopped.set()
+
+            carriers[0].loop.call_soon_threadsafe(expire_and_stop)
+            assert stopped.wait(5)
+            assert len(responder._leases) == len(responder._handshakes) == 2
+            assert len(responder._pool._pending) == 2
+            assert all(lease._busy for lease in responder._leases)
+            assert wire_request(port, "GET", "/_health_")[0] == 200
+            release.set()
+            deadline = time.monotonic() + 5
+            while responder._leases or responder._handshakes:
+                assert time.monotonic() < deadline
+                time.sleep(0.001)
+            assert not responder._attempts and not requests.pending()
+            assert not responder._pool._pending
+            assert wire_request(port, "GET", "/_health_")[0] == 200
+        finally:
+            release.set()
+            for peer in peers:
+                peer.close()
+
+
 def test_negative_native_framing_settings_are_rejected():
     for name in (
         "max_headers_bytes",
diff --git a/website/docs/operate/credentials.md b/website/docs/operate/credentials.md
index 09a71f6..9a7ed79 100644
--- a/website/docs/operate/credentials.md
+++ b/website/docs/operate/credentials.md
@@ -273,6 +273,12 @@ commit prevents confirmation. If the server never committed, ask the operator an
 explicitly start a new pairing. Do not restart pairing merely to recover a lost
 receipt: new pairing discards abandoned prepared records.
 
+Use the installed `arbiter` command for both enrollment and recovery. New enrollment
+opens the native terminal even when ordinary output is redirected. Run it from a
+terminal you control; `resume=SELECTOR` can run without a terminal. A network proxy
+must preserve the binary WebSocket stream and permit the separate receipt-control
+connection. Interrupted connections do not justify deleting saved private state.
+
 ### Public enrollment request boundary
 
 `POST /api/v1/enrollment/requests` accepts version-one provisional public
@@ -415,15 +421,17 @@ fresh HTTP connection; an upgrade that bypasses this guard is rejected.
 Ordinary HTTP body, operation and download budgets remain separate.
 
 This endpoint completes the encrypted handshake, mutual signing-key proofs,
-prepared-result delivery and ready verification before closing. Fresh Python/Go
-peers exercise the public CLI/carrier and private prepared persistence.
-Activation remains a subsequent delivery step. Local approval, prepared
-certificates and ready verification still grant no ordinary API access.
+prepared-result delivery and ready verification. After the client durably saves
+the prepared result, ready verification permits the server to recheck approval
+and commit activation. The endpoint returns the recorded receipt before closing.
+Fresh Python/Go peers exercise the installed CLI/carrier and private prepared
+persistence. Local approval or prepared certificates alone do not activate an
+enrollment, and this milestone provides no ordinary authenticated API access.
 
 Registration uses the existing bounded worker admission before reading body
 bytes. The body limit is the smaller of the core request-body limit and 64 KiB,
 with a ten-second total body-read budget. Full capacity returns a retryable
-`resource_exhausted` error. None of these public requests activate an identity.
+`resource_exhausted` error. Registration alone does not activate an identity.
 
 ### Pairing possession proof profile