"""S16D signed DM invite bootstrap regressions.

Tests:
- exported DM invites no longer expose the stable DM alias in the invite blob
- imported invites resolve the stable DM alias through the invite lookup handle and pin contacts as invite_pinned
- invite-pinned contacts can still be upgraded to sas_verified
- invite-pinned mismatches escalate to continuity_broken and reject acknowledgment
- compose/bootstrap flows fail closed when a pinned invite disagrees with relay identity material
- bootstrap decrypt rejects sender static keys that disagree with a pinned invite
"""

from __future__ import annotations

import base64
import json
import time

from cryptography.hazmat.primitives.asymmetric import x25519
from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
from services.config import get_settings
from services.mesh.mesh_protocol import PROTOCOL_VERSION


def _b64_pub(pub: x25519.X25519PublicKey) -> str:
    return base64.b64encode(pub.public_bytes(Encoding.Raw, PublicFormat.Raw)).decode("ascii")


def _fresh_wormhole_state(tmp_path, monkeypatch):
    from services.mesh import (
        mesh_dm_relay,
        mesh_secure_storage,
        mesh_wormhole_contacts,
        mesh_wormhole_identity,
        mesh_wormhole_persona,
        mesh_wormhole_prekey,
        mesh_wormhole_root_manifest,
        mesh_wormhole_root_transparency,
    )
    from services.config import get_settings

    monkeypatch.setattr(mesh_dm_relay, "DATA_DIR", tmp_path)
    monkeypatch.setattr(mesh_dm_relay, "RELAY_FILE", tmp_path / "dm_relay.json")
    monkeypatch.setattr(mesh_secure_storage, "DATA_DIR", tmp_path)
    monkeypatch.setattr(mesh_secure_storage, "MASTER_KEY_FILE", tmp_path / "wormhole_secure_store.key")
    monkeypatch.setattr(mesh_wormhole_persona, "DATA_DIR", tmp_path)
    monkeypatch.setattr(mesh_wormhole_persona, "PERSONA_FILE", tmp_path / "wormhole_persona.json")
    monkeypatch.setattr(
        mesh_wormhole_persona,
        "LEGACY_DM_IDENTITY_FILE",
        tmp_path / "wormhole_identity.json",
    )
    monkeypatch.setattr(mesh_wormhole_root_manifest, "DATA_DIR", tmp_path)
    monkeypatch.setattr(mesh_wormhole_root_transparency, "DATA_DIR", tmp_path)
    monkeypatch.setattr(mesh_wormhole_contacts, "DATA_DIR", tmp_path)
    monkeypatch.setattr(mesh_wormhole_contacts, "CONTACTS_FILE", tmp_path / "wormhole_dm_contacts.json")
    monkeypatch.setattr(mesh_secure_storage, "_MASTER_KEY_CACHE", None)
    monkeypatch.setattr(mesh_secure_storage, "_DOMAIN_KEY_CACHE", {})
    for key in (
        "MESH_DM_ROOT_EXTERNAL_WITNESS_IMPORT_PATH",
        "MESH_DM_ROOT_EXTERNAL_WITNESS_IMPORT_URI",
        "MESH_DM_ROOT_TRANSPARENCY_LEDGER_EXPORT_PATH",
        "MESH_DM_ROOT_TRANSPARENCY_LEDGER_READBACK_URI",
    ):
        monkeypatch.setenv(key, "")
    get_settings.cache_clear()

    relay = mesh_dm_relay.DMRelay()
    monkeypatch.setattr(mesh_dm_relay, "dm_relay", relay)

    mesh_wormhole_identity.bootstrap_wormhole_identity(force=True)
    return relay, mesh_wormhole_identity, mesh_wormhole_contacts, mesh_wormhole_prekey


def test_register_wormhole_dm_key_repairs_missing_local_dh_material(tmp_path, monkeypatch):
    relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    identity = identity_mod.read_wormhole_identity()
    original_node_id = identity["node_id"]
    original_public_key = identity["public_key"]
    original_private_key = identity["private_key"]

    identity_mod.write_dm_identity(
        {
            **identity,
            "dh_pub_key": "",
            "dh_private_key": "",
            "bundle_fingerprint": "",
            "bundle_sequence": 0,
            "bundle_registered_at": 0,
        }
    )

    registered = identity_mod.register_wormhole_dm_key()
    repaired = identity_mod.read_wormhole_identity()

    assert registered["ok"] is True
    assert registered["dh_pub_key"]
    assert registered["dh_algo"] == "X25519"
    assert repaired["dh_pub_key"] == registered["dh_pub_key"]
    assert repaired["dh_private_key"]
    assert repaired["node_id"] == original_node_id
    assert repaired["public_key"] == original_public_key
    assert repaired["private_key"] == original_private_key
    assert relay.get_dh_key(original_node_id)["dh_pub_key"] == registered["dh_pub_key"]


def _export_verified_invite(identity_mod):
    exported = identity_mod.export_wormhole_dm_invite()
    assert exported["ok"] is True
    verified = identity_mod.verify_wormhole_dm_invite(exported["invite"])
    assert verified["ok"] is True
    return exported, verified


def _import_invite(identity_mod, *, alias: str = ""):
    exported, verified = _export_verified_invite(identity_mod)
    imported = identity_mod.import_wormhole_dm_invite(exported["invite"], alias=alias)
    assert imported["ok"] is True
    return exported, verified, imported


def _export_compat_invite(identity_mod):
    exported, _verified = _export_verified_invite(identity_mod)
    payload = {
        **dict(exported["invite"]["payload"] or {}),
        "invite_version": identity_mod.DM_INVITE_VERSION_COMPAT,
        "attestations": [],
    }
    invite_node_id, invite_public_key, invite_private_key = identity_mod._generate_invite_signing_identity()
    signed = identity_mod._sign_dm_invite_payload(
        node_id=invite_node_id,
        public_key=invite_public_key,
        private_key=invite_private_key,
        payload=payload,
    )
    invite = {
        "event_type": identity_mod.DM_INVITE_EVENT_TYPE,
        "payload": payload,
        "node_id": str(signed.get("node_id", "") or ""),
        "public_key": str(signed.get("public_key", "") or ""),
        "public_key_algo": str(signed.get("public_key_algo", "") or ""),
        "protocol_version": str(signed.get("protocol_version", PROTOCOL_VERSION) or PROTOCOL_VERSION),
        "sequence": int(signed.get("sequence", 0) or 0),
        "signature": str(signed.get("signature", "") or ""),
        "identity_scope": str(signed.get("identity_scope", "dm_alias") or "dm_alias"),
    }
    verified = identity_mod.verify_wormhole_dm_invite(invite)
    assert verified["ok"] is True
    return invite, verified


def _export_legacy_invite(identity_mod):
    data = identity_mod.read_wormhole_identity()
    payload = {
        "invite_version": identity_mod.DM_INVITE_VERSION_LEGACY,
        "protocol_version": PROTOCOL_VERSION,
        "issued_at": int(time.time()),
        "expires_at": 0,
        "label": "legacy",
        "agent_id": str(data.get("node_id", "") or ""),
        "public_key": str(data.get("public_key", "") or ""),
        "public_key_algo": str(data.get("public_key_algo", "Ed25519") or "Ed25519"),
        "identity_dh_pub_key": str(data.get("dh_pub_key", "") or ""),
        "dh_algo": str(data.get("dh_algo", "X25519") or "X25519"),
    }
    payload["trust_fingerprint"] = identity_mod.trust_fingerprint_for_identity_material(
        agent_id=payload["agent_id"],
        identity_dh_pub_key=payload["identity_dh_pub_key"],
        dh_algo=payload["dh_algo"],
        public_key=payload["public_key"],
        public_key_algo=payload["public_key_algo"],
        protocol_version=payload["protocol_version"],
    )
    signed = identity_mod._sign_dm_invite_payload(
        node_id=str(data.get("node_id", "") or ""),
        public_key=str(data.get("public_key", "") or ""),
        private_key=str(data.get("private_key", "") or ""),
        payload=payload,
    )
    invite = {
        "event_type": identity_mod.DM_INVITE_EVENT_TYPE,
        "payload": payload,
        "node_id": str(signed.get("node_id", "") or ""),
        "public_key": str(signed.get("public_key", "") or ""),
        "public_key_algo": str(signed.get("public_key_algo", "") or ""),
        "protocol_version": str(signed.get("protocol_version", PROTOCOL_VERSION) or PROTOCOL_VERSION),
        "sequence": int(signed.get("sequence", 0) or 0),
        "signature": str(signed.get("signature", "") or ""),
        "identity_scope": str(signed.get("identity_scope", "dm_alias") or "dm_alias"),
    }
    verified = identity_mod.verify_wormhole_dm_invite(invite)
    assert verified["ok"] is True
    return invite, verified


def test_exported_dm_invite_verifies_and_tamper_fails(tmp_path, monkeypatch):
    _relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    from services.mesh import mesh_wormhole_persona

    exported, verified = _export_verified_invite(identity_mod)
    local_identity = identity_mod.read_wormhole_identity()
    persona_state = mesh_wormhole_persona.read_wormhole_persona_state()
    root_identity = persona_state["root_identity"]
    attestations = list(exported["invite"]["payload"].get("attestations") or [])
    stable_attestation = next(
        (
            item
            for item in attestations
            if isinstance(item, dict) and str(item.get("type", "")).strip().lower() == "stable_dm_identity"
        ),
        None,
    )

    assert verified["peer_id"] == exported["peer_id"]
    assert verified["trust_fingerprint"] == exported["trust_fingerprint"]
    assert exported["peer_id"] != local_identity["node_id"]
    assert exported["invite"]["payload"]["invite_version"] == identity_mod.DM_INVITE_VERSION
    assert stable_attestation is not None
    assert stable_attestation["event_type"] == identity_mod.DM_INVITE_ATTESTATION_EVENT_TYPE
    assert stable_attestation["signer_scope"] == "root"
    assert stable_attestation["root_node_id"] == root_identity["node_id"]
    assert stable_attestation["root_public_key"] == root_identity["public_key"]
    assert stable_attestation["root_manifest_fingerprint"]
    assert stable_attestation["root_node_id"] != local_identity["node_id"]
    assert stable_attestation["root_public_key"] != local_identity["public_key"]
    assert exported["invite"]["payload"]["root_manifest"]["payload"]["root_fingerprint"]
    assert exported["invite"]["payload"]["root_manifest_witness"]["payload"]["manifest_fingerprint"]
    assert len(exported["invite"]["payload"]["root_manifest_witnesses"]) == 3
    assert "agent_id" not in exported["invite"]["payload"]
    assert "public_key" not in exported["invite"]["payload"]
    assert "identity_dh_pub_key" not in exported["invite"]["payload"]
    assert local_identity["node_id"] not in json.dumps(exported["invite"]["payload"], sort_keys=True)
    assert local_identity["public_key"] not in json.dumps(exported["invite"]["payload"], sort_keys=True)

    tampered = dict(exported["invite"])
    tampered["payload"] = {
        **dict(exported["invite"]["payload"]),
        "identity_commitment": "ff" * 32,
    }
    rejected = identity_mod.verify_wormhole_dm_invite(tampered)

    assert rejected["ok"] is False
    assert rejected["detail"] == "invite signature invalid"


def test_exported_dm_invite_requires_stable_identity_attestation(tmp_path, monkeypatch):
    _relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)

    exported, _verified = _export_verified_invite(identity_mod)
    tampered = dict(exported["invite"])
    tampered["payload"] = {
        **dict(exported["invite"]["payload"]),
        "attestations": [],
    }

    rejected = identity_mod.verify_wormhole_dm_invite(tampered)

    assert rejected["ok"] is False
    assert rejected["detail"] == "invite stable identity attestation required"


def test_exported_dm_invite_requires_root_manifest_distribution(tmp_path, monkeypatch):
    _relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)

    exported, _verified = _export_verified_invite(identity_mod)
    tampered = dict(exported["invite"])
    tampered["payload"] = {
        **dict(exported["invite"]["payload"]),
        "root_manifest": {},
    }

    rejected = identity_mod.verify_wormhole_dm_invite(tampered)

    assert rejected["ok"] is False
    assert rejected["detail"] == "invite root manifest required"


def test_exported_dm_invite_carries_staged_external_witness_receipts(tmp_path, monkeypatch):
    _relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    from services.mesh import mesh_wormhole_root_manifest as manifest_mod

    external_identity = manifest_mod._witness_identity_record(index=9)
    external_identity["management_scope"] = "external"
    external_identity["independence_group"] = "independent_a"
    manifest_mod.configure_external_root_witness_descriptors(
        [manifest_mod._public_witness_descriptor(external_identity)]
    )
    published = manifest_mod.publish_current_root_manifest(expires_in_s=3600, policy_version=1)
    external_receipt = manifest_mod._sign_with_witness_identity(
        identity=external_identity,
        event_type=manifest_mod.STABLE_DM_ROOT_MANIFEST_WITNESS_EVENT_TYPE,
        payload=manifest_mod._witness_payload(published["manifest"]),
    )
    staged = manifest_mod.stage_external_root_manifest_witnesses(
        [external_receipt],
        manifest=published["manifest"],
    )

    exported, verified = _export_verified_invite(identity_mod)
    manifest = dict(exported["invite"]["payload"].get("root_manifest") or {})
    witness_set = list(exported["invite"]["payload"].get("root_manifest_witnesses") or [])
    external_receipts = [
        item
        for item in witness_set
        if str(item.get("node_id", "") or "").strip() == str(external_identity.get("node_id", "") or "").strip()
        and str(item.get("public_key", "") or "").strip() == str(external_identity.get("public_key", "") or "").strip()
    ]
    witness_verified = manifest_mod.verify_root_manifest_witness_set(manifest, witness_set)

    assert staged["ok"] is True
    assert staged["external_witness_count"] == 1
    assert staged["witness_independent_quorum_met"] is True
    assert verified["ok"] is True
    assert len(witness_set) == 4
    assert len(external_receipts) == 1
    assert witness_verified["ok"] is True
    assert witness_verified["witness_domain_count"] == 2
    assert witness_verified["witness_independent_quorum_met"] is True


def test_exported_dm_invite_requires_proven_witnessed_root_rotation(tmp_path, monkeypatch):
    _relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    from services.mesh import mesh_wormhole_persona
    from services.mesh import mesh_wormhole_root_manifest
    from services.mesh import mesh_wormhole_root_transparency

    _export_verified_invite(identity_mod)
    mesh_wormhole_persona.bootstrap_wormhole_persona_state(force=True)
    exported, _verified = _export_verified_invite(identity_mod)

    tampered_manifest_payload = {
        **dict(exported["invite"]["payload"]["root_manifest"]["payload"] or {}),
        "previous_root_cross_sequence": 0,
        "previous_root_cross_signature": "",
    }
    resigned_manifest = mesh_wormhole_persona.sign_root_wormhole_event(
        event_type=mesh_wormhole_root_manifest.STABLE_DM_ROOT_MANIFEST_EVENT_TYPE,
        payload=tampered_manifest_payload,
    )
    tampered_manifest = {
        "type": mesh_wormhole_root_manifest.STABLE_DM_ROOT_MANIFEST_TYPE,
        "event_type": mesh_wormhole_root_manifest.STABLE_DM_ROOT_MANIFEST_EVENT_TYPE,
        "node_id": str(resigned_manifest.get("node_id", "") or ""),
        "public_key": str(resigned_manifest.get("public_key", "") or ""),
        "public_key_algo": str(resigned_manifest.get("public_key_algo", "Ed25519") or "Ed25519"),
        "protocol_version": str(resigned_manifest.get("protocol_version", PROTOCOL_VERSION) or PROTOCOL_VERSION),
        "sequence": int(resigned_manifest.get("sequence", 0) or 0),
        "payload": dict(resigned_manifest.get("payload") or {}),
        "signature": str(resigned_manifest.get("signature", "") or ""),
        "identity_scope": "root",
    }
    witness_state = mesh_wormhole_root_manifest.read_root_distribution_state()
    witness_identities = list(witness_state.get("witness_identities") or [])
    tampered_witnesses = [
        mesh_wormhole_root_manifest._sign_with_witness_identity(
            identity=dict(identity or {}),
            event_type=mesh_wormhole_root_manifest.STABLE_DM_ROOT_MANIFEST_WITNESS_EVENT_TYPE,
            payload=mesh_wormhole_root_manifest._witness_payload(tampered_manifest),
        )
        for identity in witness_identities
    ]
    tampered_transparency = mesh_wormhole_root_transparency.publish_root_transparency_record(
        distribution={"manifest": tampered_manifest, "witnesses": tampered_witnesses}
    )

    rejected = identity_mod._verify_dm_invite_root_distribution(
        {
            **dict(exported["invite"]["payload"] or {}),
            "root_manifest": tampered_manifest,
            "root_manifest_witness": dict(tampered_witnesses[0] or {}),
            "root_manifest_witnesses": tampered_witnesses,
            "root_transparency_record": dict(tampered_transparency.get("record") or {}),
        }
    )

    assert rejected["ok"] is False
    assert rejected["detail"] == "invite root rotation proof required"


def test_exported_dm_invite_requires_witness_threshold(tmp_path, monkeypatch):
    _relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)

    exported, _verified = _export_verified_invite(identity_mod)
    rejected = identity_mod._verify_dm_invite_root_distribution(
        {
            **dict(exported["invite"]["payload"] or {}),
            "root_manifest_witnesses": [dict(exported["invite"]["payload"]["root_manifest_witnesses"][0] or {})],
        }
    )

    assert rejected["ok"] is False
    assert rejected["detail"] == "stable root manifest witness threshold not met"


def test_exported_dm_invite_requires_root_transparency_record(tmp_path, monkeypatch):
    _relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)

    exported, _verified = _export_verified_invite(identity_mod)
    rejected = identity_mod._verify_dm_invite_root_distribution(
        {
            **dict(exported["invite"]["payload"] or {}),
            "root_transparency_record": {},
        }
    )

    assert rejected["ok"] is False
    assert rejected["detail"] == "invite root transparency record required"


def test_exported_dm_invite_requires_witness_policy_change_proof(tmp_path, monkeypatch):
    _relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    from services.mesh import mesh_wormhole_persona
    from services.mesh import mesh_wormhole_root_manifest
    from services.mesh import mesh_wormhole_root_transparency

    _export_verified_invite(identity_mod)
    republished = mesh_wormhole_root_manifest.publish_current_root_manifest(expires_in_s=3600, policy_version=2)

    tampered_manifest_payload = {
        **dict(republished["manifest"]["payload"] or {}),
        "previous_witness_policy_sequence": 0,
        "previous_witness_policy_signature": "",
    }
    resigned_manifest = mesh_wormhole_persona.sign_root_wormhole_event(
        event_type=mesh_wormhole_root_manifest.STABLE_DM_ROOT_MANIFEST_EVENT_TYPE,
        payload=tampered_manifest_payload,
    )
    tampered_manifest = {
        "type": mesh_wormhole_root_manifest.STABLE_DM_ROOT_MANIFEST_TYPE,
        "event_type": mesh_wormhole_root_manifest.STABLE_DM_ROOT_MANIFEST_EVENT_TYPE,
        "node_id": str(resigned_manifest.get("node_id", "") or ""),
        "public_key": str(resigned_manifest.get("public_key", "") or ""),
        "public_key_algo": str(resigned_manifest.get("public_key_algo", "Ed25519") or "Ed25519"),
        "protocol_version": str(resigned_manifest.get("protocol_version", PROTOCOL_VERSION) or PROTOCOL_VERSION),
        "sequence": int(resigned_manifest.get("sequence", 0) or 0),
        "payload": dict(resigned_manifest.get("payload") or {}),
        "signature": str(resigned_manifest.get("signature", "") or ""),
        "identity_scope": "root",
    }
    witness_state = mesh_wormhole_root_manifest.read_root_distribution_state()
    witness_identities = list(witness_state.get("witness_identities") or [])
    tampered_witnesses = [
        mesh_wormhole_root_manifest._sign_with_witness_identity(
            identity=dict(identity or {}),
            event_type=mesh_wormhole_root_manifest.STABLE_DM_ROOT_MANIFEST_WITNESS_EVENT_TYPE,
            payload=mesh_wormhole_root_manifest._witness_payload(tampered_manifest),
        )
        for identity in witness_identities
    ]
    tampered_transparency = mesh_wormhole_root_transparency.publish_root_transparency_record(
        distribution={"manifest": tampered_manifest, "witnesses": tampered_witnesses}
    )

    rejected = identity_mod._verify_dm_invite_root_distribution(
        {
            **dict(republished["manifest"]["payload"] or {}),
            "root_manifest": tampered_manifest,
            "root_manifest_witness": dict(tampered_witnesses[0] or {}),
            "root_manifest_witnesses": tampered_witnesses,
            "root_transparency_record": dict(tampered_transparency.get("record") or {}),
        }
    )

    assert rejected["ok"] is False
    assert rejected["detail"] == "invite root witness policy change proof required"


def test_imported_dm_invite_pins_contact_as_invite_pinned(tmp_path, monkeypatch):
    _relay, identity_mod, contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    exported, verified, imported = _import_invite(identity_mod, alias="alice")
    contact = imported["contact"]
    local_identity = identity_mod.read_wormhole_identity()

    assert contact["alias"] == "alice"
    assert contact["trust_level"] == "invite_pinned"
    assert contact["trustSummary"]["state"] == "invite_pinned"
    assert contact["trustSummary"]["verifiedFirstContact"] is True
    assert contact["trustSummary"]["rootWitnessed"] is True
    assert contact["trustSummary"]["rootDistributionState"] == "quorum_witnessed"
    assert contact["trustSummary"]["rootWitnessQuorumMet"] is True
    assert contact["trustSummary"]["rootWitnessProvenanceState"] == "local_quorum"
    assert contact["trustSummary"]["rootWitnessFinalityMet"] is False
    assert contact["trustSummary"]["rootWitnessThreshold"] == 2
    assert contact["trustSummary"]["rootWitnessCount"] == 3
    assert imported["peer_id"] == local_identity["node_id"]
    assert imported["invite_peer_id"] == verified["peer_id"]
    assert contact["invitePinnedTrustFingerprint"] == imported["trust_fingerprint"]
    assert contact["invitePinnedRootFingerprint"]
    assert contact["remotePrekeyRootFingerprint"] == contact["invitePinnedRootFingerprint"]
    assert contact["remotePrekeyFingerprint"] == imported["trust_fingerprint"]
    assert contact["invitePinnedDhPubKey"] == local_identity["dh_pub_key"]
    assert contact["invitePinnedPrekeyLookupHandle"] == exported["invite"]["payload"]["prekey_lookup_handle"]
    assert contacts_mod.list_wormhole_dm_contacts()[imported["peer_id"]]["trust_level"] == "invite_pinned"


def test_imported_dm_invite_saves_pending_contact_when_prekey_not_visible(tmp_path, monkeypatch):
    _relay, identity_mod, contacts_mod, prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    exported, verified = _export_verified_invite(identity_mod)
    monkeypatch.setattr(
        prekey_mod,
        "fetch_dm_prekey_bundle",
        lambda **_kw: {"ok": False, "detail": "Prekey bundle not found"},
    )

    imported = identity_mod.import_wormhole_dm_invite(exported["invite"], alias="alice")
    contact = imported["contact"]

    assert imported["ok"] is True
    assert imported["pending_prekey"] is True
    assert imported["peer_id"] == verified["peer_id"]
    assert contact["alias"] == "alice"
    assert contact["trust_level"] == "invite_pinned"
    assert contact["invitePinnedPrekeyLookupHandle"] == exported["invite"]["payload"]["prekey_lookup_handle"]
    assert contact["remotePrekeyLookupMode"] == "invite_lookup_handle"
    assert contact["remotePrekeyFingerprint"] == verified["trust_fingerprint"]
    assert contact["dhPubKey"] == ""
    assert contacts_mod.list_wormhole_dm_contacts()[verified["peer_id"]]["trust_level"] == "invite_pinned"


def test_imported_dm_invite_requires_root_attested_prekey_bundle(tmp_path, monkeypatch):
    relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)

    exported, _verified = _export_verified_invite(identity_mod)
    local_identity = identity_mod.read_wormhole_identity()
    agent_id = str(local_identity.get("node_id", "") or "")
    relay._prekey_bundles[agent_id]["bundle"] = {
        **dict(relay._prekey_bundles[agent_id]["bundle"] or {}),
        "root_attestation": {},
    }

    imported = identity_mod.import_wormhole_dm_invite(exported["invite"], alias="alice")

    assert imported["ok"] is False
    assert imported["detail"] == "prekey bundle root attestation required"


def test_imported_dm_invite_requires_same_witnessed_root_manifest(tmp_path, monkeypatch):
    relay, identity_mod, _contacts_mod, prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    from services.mesh import mesh_wormhole_root_manifest
    from services.mesh import mesh_wormhole_root_transparency

    exported, _verified = _export_verified_invite(identity_mod)
    local_identity = identity_mod.read_wormhole_identity()
    agent_id = str(local_identity.get("node_id", "") or "")

    republished = mesh_wormhole_root_manifest.publish_current_root_manifest(expires_in_s=3600, policy_version=2)
    republished_transparency = mesh_wormhole_root_transparency.publish_root_transparency_record(
        distribution={
            "manifest": dict(republished.get("manifest") or {}),
            "witnesses": list(republished.get("witnesses") or []),
        }
    )
    stored = dict(relay._prekey_bundles[agent_id] or {})
    tampered_bundle = dict(stored.get("bundle") or {})
    tampered_bundle["root_manifest"] = dict(republished.get("manifest") or {})
    tampered_bundle["root_manifest_witness"] = dict(republished.get("witness") or {})
    tampered_bundle["root_manifest_witnesses"] = list(republished.get("witnesses") or [])
    tampered_bundle["root_transparency_record"] = dict(republished_transparency.get("record") or {})
    tampered_bundle = prekey_mod._attach_bundle_root_attestation(
        agent_id=agent_id,
        public_key=str(stored.get("public_key", "") or ""),
        public_key_algo=str(stored.get("public_key_algo", "Ed25519") or "Ed25519"),
        protocol_version=str(stored.get("protocol_version", PROTOCOL_VERSION) or PROTOCOL_VERSION),
        bundle=tampered_bundle,
    )
    tampered_bundle = prekey_mod._attach_bundle_signature(
        tampered_bundle,
        signed_at=int(tampered_bundle.get("signed_at", 0) or time.time()),
    )
    relay._prekey_bundles[agent_id]["bundle"] = tampered_bundle

    imported = identity_mod.import_wormhole_dm_invite(exported["invite"], alias="alice")

    assert imported["ok"] is False
    assert imported["detail"] == "invite root manifest mismatch"


def test_imported_dm_invite_requires_matching_root_transparency_binding(tmp_path, monkeypatch):
    relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    from services.mesh import mesh_wormhole_root_transparency

    exported, _verified = _export_verified_invite(identity_mod)
    local_identity = identity_mod.read_wormhole_identity()
    agent_id = str(local_identity.get("node_id", "") or "")

    stored = dict(relay._prekey_bundles[agent_id] or {})
    tampered_bundle = dict(stored.get("bundle") or {})
    tampered_bundle["root_manifest_witnesses"] = [
        dict(item or {}) for item in list(tampered_bundle.get("root_manifest_witnesses") or [])[:2]
    ]
    tampered_bundle["root_manifest_witness"] = dict(tampered_bundle["root_manifest_witnesses"][0] or {})
    tampered_transparency = mesh_wormhole_root_transparency.publish_root_transparency_record(
        distribution={
            "manifest": dict(tampered_bundle.get("root_manifest") or {}),
            "witnesses": list(tampered_bundle.get("root_manifest_witnesses") or []),
        }
    )
    tampered_bundle["root_transparency_record"] = dict(tampered_transparency.get("record") or {})
    relay._prekey_bundles[agent_id]["bundle"] = tampered_bundle

    imported = identity_mod.import_wormhole_dm_invite(exported["invite"], alias="alice")

    assert imported["ok"] is False
    assert imported["detail"] == "invite root transparency mismatch"


def test_imported_dm_invite_accepts_configured_external_witness_and_transparency_sources(tmp_path, monkeypatch):
    _relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    from services.mesh import mesh_wormhole_root_manifest as manifest_mod
    from services.mesh import mesh_wormhole_root_transparency as transparency_mod

    external_identity = manifest_mod._witness_identity_record(index=9)
    external_identity["management_scope"] = "external"
    external_identity["independence_group"] = "independent_a"
    manifest_mod.configure_external_root_witness_descriptors(
        [manifest_mod._public_witness_descriptor(external_identity)]
    )
    published = manifest_mod.publish_current_root_manifest(expires_in_s=3600, policy_version=1)
    external_receipt = manifest_mod._sign_with_witness_identity(
        identity=external_identity,
        event_type=manifest_mod.STABLE_DM_ROOT_MANIFEST_WITNESS_EVENT_TYPE,
        payload=manifest_mod._witness_payload(published["manifest"]),
    )
    manifest_mod.stage_external_root_manifest_witnesses(
        [external_receipt],
        manifest=published["manifest"],
    )

    exported, _verified = _export_verified_invite(identity_mod)
    ledger_path = tmp_path / "external_readback_ledger.json"
    package_path = tmp_path / "external_witness_source.json"
    transparency_mod.publish_root_transparency_ledger_to_file(path=str(ledger_path), max_records=8)
    package_path.write_text(
        manifest_mod._stable_json(
            {
                "type": manifest_mod.STABLE_DM_ROOT_MANIFEST_EXTERNAL_WITNESS_IMPORT_TYPE,
                "schema_version": 1,
                "source_scope": "https_fetch",
                "source_label": "witness-a",
                "exported_at": int(time.time()),
                "descriptors": [manifest_mod._public_witness_descriptor(external_identity)],
                "manifest_fingerprint": str(
                    exported["invite"]["payload"]["attestations"][0]["root_manifest_fingerprint"]
                ),
                "witnesses": [external_receipt],
            }
        ),
        encoding="utf-8",
    )
    monkeypatch.setenv("MESH_DM_ROOT_EXTERNAL_WITNESS_IMPORT_URI", package_path.as_uri())
    monkeypatch.setenv("MESH_DM_ROOT_TRANSPARENCY_LEDGER_READBACK_URI", ledger_path.as_uri())
    get_settings.cache_clear()

    imported = identity_mod.import_wormhole_dm_invite(exported["invite"], alias="alice")

    assert imported["ok"] is True
    assert imported["contact"]["trustSummary"]["state"] == "invite_pinned"
    assert imported["contact"]["trustSummary"]["rootWitnessProvenanceState"] == "independent_quorum"
    assert imported["contact"]["trustSummary"]["rootWitnessFinalityMet"] is True
    assert imported["contact"]["trustSummary"]["verifiedFirstContact"] is True


def test_imported_dm_invite_downgrades_verified_first_contact_when_finality_enforcement_is_enabled(tmp_path, monkeypatch):
    _relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    monkeypatch.setenv("WORMHOLE_ROOT_WITNESS_FINALITY_ENFORCE", "true")
    get_settings.cache_clear()
    try:
        _exported, _verified, imported = _import_invite(identity_mod, alias="alice")
        contact = imported["contact"]

        assert imported["ok"] is True
        assert contact["trust_level"] == "invite_pinned"
        assert contact["trustSummary"]["rootDistributionState"] == "quorum_witnessed"
        assert contact["trustSummary"]["rootWitnessProvenanceState"] == "local_quorum"
        assert contact["trustSummary"]["rootWitnessFinalityMet"] is False
        assert contact["trustSummary"]["verifiedFirstContact"] is False
        assert contact["trustSummary"]["recommendedAction"] == "import_invite"
    finally:
        get_settings.cache_clear()


def test_exported_dm_invite_rejects_configured_external_transparency_readback_mismatch(tmp_path, monkeypatch):
    _relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    from services.config import get_settings
    from services.mesh import mesh_wormhole_root_manifest as manifest_mod
    from services.mesh import mesh_wormhole_root_transparency as transparency_mod

    external_identity = manifest_mod._witness_identity_record(index=9)
    external_identity["management_scope"] = "external"
    external_identity["independence_group"] = "independent_a"
    manifest_mod.configure_external_root_witness_descriptors(
        [manifest_mod._public_witness_descriptor(external_identity)]
    )
    published = manifest_mod.publish_current_root_manifest(expires_in_s=3600, policy_version=1)
    external_receipt = manifest_mod._sign_with_witness_identity(
        identity=external_identity,
        event_type=manifest_mod.STABLE_DM_ROOT_MANIFEST_WITNESS_EVENT_TYPE,
        payload=manifest_mod._witness_payload(published["manifest"]),
    )

    package_path = tmp_path / "external_witness_source.json"
    bad_ledger_path = tmp_path / "external_readback_bad_ledger.json"
    package_path.write_text(
        manifest_mod._stable_json(
            {
                "type": manifest_mod.STABLE_DM_ROOT_MANIFEST_EXTERNAL_WITNESS_IMPORT_TYPE,
                "schema_version": 1,
                "source_scope": "https_fetch",
                "source_label": "witness-a",
                "exported_at": int(time.time()),
                "descriptors": [manifest_mod._public_witness_descriptor(external_identity)],
                "manifest_fingerprint": published["manifest_fingerprint"],
                "witnesses": [external_receipt],
            }
        ),
        encoding="utf-8",
    )
    bad_ledger_path.write_text(
        json.dumps(
            {
                "type": transparency_mod.STABLE_DM_ROOT_TRANSPARENCY_LEDGER_TYPE,
                "schema_version": 1,
                "transparency_scope": transparency_mod.ROOT_TRANSPARENCY_SCOPE,
                "exported_at": int(time.time()),
                "record_count": 0,
                "current_record_fingerprint": "",
                "head_binding_fingerprint": "",
                "chain_fingerprint": transparency_mod.transparency_record_chain_fingerprint([]),
                "records": [],
            }
        ),
        encoding="utf-8",
    )

    monkeypatch.setenv("MESH_DM_ROOT_EXTERNAL_WITNESS_IMPORT_URI", package_path.as_uri())
    monkeypatch.setenv("MESH_DM_ROOT_TRANSPARENCY_LEDGER_READBACK_URI", bad_ledger_path.as_uri())
    get_settings.cache_clear()

    exported = identity_mod.export_wormhole_dm_invite()

    assert exported["ok"] is False
    assert exported["detail"] == "root transparency external ledger head mismatch"


def test_exported_dm_invite_rejects_stale_configured_external_witness_source(tmp_path, monkeypatch):
    _relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    from services.config import get_settings
    from services.mesh import mesh_wormhole_root_manifest as manifest_mod

    external_identity = manifest_mod._witness_identity_record(index=9)
    external_identity["management_scope"] = "external"
    external_identity["independence_group"] = "independent_a"
    manifest_mod.configure_external_root_witness_descriptors(
        [manifest_mod._public_witness_descriptor(external_identity)]
    )
    published = manifest_mod.publish_current_root_manifest(expires_in_s=3600, policy_version=1)
    external_receipt = manifest_mod._sign_with_witness_identity(
        identity=external_identity,
        event_type=manifest_mod.STABLE_DM_ROOT_MANIFEST_WITNESS_EVENT_TYPE,
        payload=manifest_mod._witness_payload(published["manifest"]),
    )
    package_path = tmp_path / "stale_external_witness_source.json"
    package_path.write_text(
        manifest_mod._stable_json(
            {
                "type": manifest_mod.STABLE_DM_ROOT_MANIFEST_EXTERNAL_WITNESS_IMPORT_TYPE,
                "schema_version": 1,
                "source_scope": "https_fetch",
                "source_label": "witness-a",
                "exported_at": int(time.time()) - 120,
                "descriptors": [manifest_mod._public_witness_descriptor(external_identity)],
                "manifest_fingerprint": published["manifest_fingerprint"],
                "witnesses": [external_receipt],
            }
        ),
        encoding="utf-8",
    )
    monkeypatch.setenv("MESH_DM_ROOT_EXTERNAL_WITNESS_IMPORT_URI", package_path.as_uri())
    monkeypatch.setenv("MESH_DM_ROOT_EXTERNAL_WITNESS_MAX_AGE_S", "60")
    get_settings.cache_clear()

    exported = identity_mod.export_wormhole_dm_invite()

    assert exported["ok"] is False
    assert exported["detail"] == "external root witness source stale"


def test_exported_dm_invite_rejects_stale_configured_external_transparency_readback(tmp_path, monkeypatch):
    _relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    from services.config import get_settings
    from services.mesh import mesh_wormhole_root_manifest as manifest_mod
    from services.mesh import mesh_wormhole_root_transparency as transparency_mod

    external_identity = manifest_mod._witness_identity_record(index=9)
    external_identity["management_scope"] = "external"
    external_identity["independence_group"] = "independent_a"
    manifest_mod.configure_external_root_witness_descriptors(
        [manifest_mod._public_witness_descriptor(external_identity)]
    )
    published = manifest_mod.publish_current_root_manifest(expires_in_s=3600, policy_version=1)
    external_receipt = manifest_mod._sign_with_witness_identity(
        identity=external_identity,
        event_type=manifest_mod.STABLE_DM_ROOT_MANIFEST_WITNESS_EVENT_TYPE,
        payload=manifest_mod._witness_payload(published["manifest"]),
    )
    package_path = tmp_path / "fresh_external_witness_source.json"
    readback_path = tmp_path / "stale_external_readback_ledger.json"
    package_path.write_text(
        manifest_mod._stable_json(
            {
                "type": manifest_mod.STABLE_DM_ROOT_MANIFEST_EXTERNAL_WITNESS_IMPORT_TYPE,
                "schema_version": 1,
                "source_scope": "https_fetch",
                "source_label": "witness-a",
                "exported_at": int(time.time()),
                "descriptors": [manifest_mod._public_witness_descriptor(external_identity)],
                "manifest_fingerprint": published["manifest_fingerprint"],
                "witnesses": [external_receipt],
            }
        ),
        encoding="utf-8",
    )
    current_distribution = manifest_mod.get_current_root_manifest()
    current_transparency = transparency_mod.get_current_root_transparency_record(distribution=current_distribution)
    stale_ledger = transparency_mod.export_root_transparency_ledger()["ledger"]
    stale_ledger["exported_at"] = int(time.time()) - 120
    readback_path.write_text(json.dumps(stale_ledger), encoding="utf-8")

    assert current_transparency["ok"] is True

    monkeypatch.setenv("MESH_DM_ROOT_EXTERNAL_WITNESS_IMPORT_URI", package_path.as_uri())
    monkeypatch.setenv("MESH_DM_ROOT_TRANSPARENCY_LEDGER_READBACK_URI", readback_path.as_uri())
    monkeypatch.setenv("MESH_DM_ROOT_TRANSPARENCY_LEDGER_MAX_AGE_S", "60")
    get_settings.cache_clear()

    exported = identity_mod.export_wormhole_dm_invite()

    assert exported["ok"] is False
    assert exported["detail"] == "root transparency external ledger stale"


def test_external_witness_source_loss_downgrades_operator_state_and_blocks_strong_export(tmp_path, monkeypatch):
    _relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    from services.config import get_settings
    from services.mesh import mesh_wormhole_root_manifest as manifest_mod

    external_identity = manifest_mod._witness_identity_record(index=9)
    external_identity["management_scope"] = "external"
    external_identity["independence_group"] = "independent_a"
    package_path = tmp_path / "external_witness_source_loss.json"

    monkeypatch.setenv("MESH_DM_ROOT_EXTERNAL_WITNESS_IMPORT_URI", package_path.as_uri())
    get_settings.cache_clear()

    package_path.write_text(
        manifest_mod._stable_json(
            {
                "type": manifest_mod.STABLE_DM_ROOT_MANIFEST_EXTERNAL_WITNESS_IMPORT_TYPE,
                "schema_version": 1,
                "source_scope": "https_fetch",
                "source_label": "witness-a",
                "exported_at": int(time.time()),
                "descriptors": [manifest_mod._public_witness_descriptor(external_identity)],
            }
        ),
        encoding="utf-8",
    )

    first_distribution = manifest_mod.get_current_root_manifest()
    current_receipt = manifest_mod._sign_with_witness_identity(
        identity=external_identity,
        event_type=manifest_mod.STABLE_DM_ROOT_MANIFEST_WITNESS_EVENT_TYPE,
        payload=manifest_mod._witness_payload(first_distribution["manifest"]),
    )
    package_path.write_text(
        manifest_mod._stable_json(
            {
                "type": manifest_mod.STABLE_DM_ROOT_MANIFEST_EXTERNAL_WITNESS_IMPORT_TYPE,
                "schema_version": 1,
                "source_scope": "https_fetch",
                "source_label": "witness-a",
                "exported_at": int(time.time()),
                "descriptors": [manifest_mod._public_witness_descriptor(external_identity)],
                "manifest_fingerprint": first_distribution["manifest_fingerprint"],
                "witnesses": [current_receipt],
            }
        ),
        encoding="utf-8",
    )

    current_distribution = manifest_mod.get_current_root_manifest()
    package_path.unlink()

    source_lost_distribution = manifest_mod.get_current_root_manifest()
    failed_export = identity_mod.export_wormhole_dm_invite()

    assert current_distribution["external_witness_operator_state"] == "current"
    assert current_distribution["external_witness_reacquire_required"] is False
    assert source_lost_distribution["external_witness_refresh_ok"] is False
    assert "source unreadable" in str(source_lost_distribution["external_witness_refresh_detail"] or "")
    assert source_lost_distribution["external_witness_receipts_current"] is True
    assert source_lost_distribution["external_witness_operator_state"] == "error"
    assert source_lost_distribution["external_witness_reacquire_required"] is True
    assert failed_export["ok"] is False
    assert "external root witness import source unreadable" in str(failed_export.get("detail", "") or "")


def test_imported_dm_invite_ignores_local_external_source_and_readback_mismatch(tmp_path, monkeypatch):
    _relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    from services.mesh import mesh_wormhole_root_manifest as manifest_mod
    from services.mesh import mesh_wormhole_root_transparency as transparency_mod

    external_identity = manifest_mod._witness_identity_record(index=9)
    external_identity["management_scope"] = "external"
    external_identity["independence_group"] = "independent_a"
    manifest_mod.configure_external_root_witness_descriptors(
        [manifest_mod._public_witness_descriptor(external_identity)]
    )
    published = manifest_mod.publish_current_root_manifest(expires_in_s=3600, policy_version=1)
    external_receipt = manifest_mod._sign_with_witness_identity(
        identity=external_identity,
        event_type=manifest_mod.STABLE_DM_ROOT_MANIFEST_WITNESS_EVENT_TYPE,
        payload=manifest_mod._witness_payload(published["manifest"]),
    )
    manifest_mod.stage_external_root_manifest_witnesses(
        [external_receipt],
        manifest=published["manifest"],
    )

    exported, _verified = _export_verified_invite(identity_mod)
    bad_ledger_path = tmp_path / "external_readback_bad_ledger.json"
    package_path = tmp_path / "external_witness_source.json"
    bad_ledger_path.write_text(
        json.dumps(
            {
                "type": transparency_mod.STABLE_DM_ROOT_TRANSPARENCY_LEDGER_TYPE,
                "schema_version": 1,
                "transparency_scope": transparency_mod.ROOT_TRANSPARENCY_SCOPE,
                "exported_at": int(time.time()),
                "record_count": 0,
                "current_record_fingerprint": "",
                "head_binding_fingerprint": "",
                "chain_fingerprint": transparency_mod.transparency_record_chain_fingerprint([]),
                "records": [],
            }
        ),
        encoding="utf-8",
    )
    package_path.write_text(
        manifest_mod._stable_json(
            {
                "type": manifest_mod.STABLE_DM_ROOT_MANIFEST_EXTERNAL_WITNESS_IMPORT_TYPE,
                "schema_version": 1,
                "source_scope": "https_fetch",
                "source_label": "witness-a",
                "exported_at": int(time.time()),
                "descriptors": [manifest_mod._public_witness_descriptor(external_identity)],
                "manifest_fingerprint": "00" * 32,
                "witnesses": [external_receipt],
            }
        ),
        encoding="utf-8",
    )
    monkeypatch.setenv("MESH_DM_ROOT_EXTERNAL_WITNESS_IMPORT_URI", package_path.as_uri())
    monkeypatch.setenv("MESH_DM_ROOT_TRANSPARENCY_LEDGER_READBACK_URI", bad_ledger_path.as_uri())
    get_settings.cache_clear()

    verified = identity_mod.verify_wormhole_dm_invite(exported["invite"])
    imported = identity_mod.import_wormhole_dm_invite(exported["invite"], alias="alice")

    assert verified["ok"] is True
    assert imported["ok"] is True


def test_deployment_style_external_reacquisition_restores_strong_invite_bootstrap(tmp_path, monkeypatch):
    _relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    from services.config import get_settings
    from services.mesh import mesh_wormhole_root_manifest as manifest_mod
    from services.mesh import mesh_wormhole_root_transparency as transparency_mod

    external_identity = manifest_mod._witness_identity_record(index=9)
    external_identity["management_scope"] = "external"
    external_identity["independence_group"] = "independent_a"
    package_path = tmp_path / "deployment_external_witness.json"
    ledger_path = tmp_path / "deployment_external_ledger.json"

    monkeypatch.setenv("MESH_DM_ROOT_EXTERNAL_WITNESS_IMPORT_URI", package_path.as_uri())
    monkeypatch.setenv("MESH_DM_ROOT_TRANSPARENCY_LEDGER_EXPORT_PATH", str(ledger_path))
    monkeypatch.setenv("MESH_DM_ROOT_TRANSPARENCY_LEDGER_READBACK_URI", ledger_path.as_uri())
    get_settings.cache_clear()

    package_path.write_text(
        manifest_mod._stable_json(
            {
                "type": manifest_mod.STABLE_DM_ROOT_MANIFEST_EXTERNAL_WITNESS_IMPORT_TYPE,
                "schema_version": 1,
                "source_scope": "https_fetch",
                "source_label": "witness-a",
                "exported_at": int(time.time()),
                "descriptors": [manifest_mod._public_witness_descriptor(external_identity)],
            }
        ),
        encoding="utf-8",
    )

    first_distribution = manifest_mod.get_current_root_manifest()
    assert first_distribution["external_witness_operator_state"] == "descriptors_only"

    first_receipt = manifest_mod._sign_with_witness_identity(
        identity=external_identity,
        event_type=manifest_mod.STABLE_DM_ROOT_MANIFEST_WITNESS_EVENT_TYPE,
        payload=manifest_mod._witness_payload(first_distribution["manifest"]),
    )
    package_path.write_text(
        manifest_mod._stable_json(
            {
                "type": manifest_mod.STABLE_DM_ROOT_MANIFEST_EXTERNAL_WITNESS_IMPORT_TYPE,
                "schema_version": 1,
                "source_scope": "https_fetch",
                "source_label": "witness-a",
                "exported_at": int(time.time()),
                "descriptors": [manifest_mod._public_witness_descriptor(external_identity)],
                "manifest_fingerprint": first_distribution["manifest_fingerprint"],
                "witnesses": [first_receipt],
            }
        ),
        encoding="utf-8",
    )

    current_distribution = manifest_mod.get_current_root_manifest()
    current_transparency = transparency_mod.get_current_root_transparency_record(distribution=current_distribution)
    initial_import = _import_invite(identity_mod, alias="alice")

    assert current_distribution["external_witness_operator_state"] == "current"
    assert current_transparency["ledger_operator_state"] == "current"
    assert initial_import[2]["ok"] is True

    republished = manifest_mod.publish_current_root_manifest(expires_in_s=3600, policy_version=2)
    stale_distribution = manifest_mod.get_current_root_manifest()
    stale_transparency = transparency_mod.get_current_root_transparency_record(distribution=republished)
    stale_export = identity_mod.export_wormhole_dm_invite()

    assert stale_distribution["external_witness_operator_state"] == "stale"
    assert stale_distribution["external_witness_reacquire_required"] is True
    assert stale_transparency["ledger_operator_state"] == "current"
    assert stale_export["ok"] is False
    assert "external root witness source manifest_fingerprint mismatch" in str(
        stale_export.get("detail", "")
    )

    refreshed_receipt = manifest_mod._sign_with_witness_identity(
        identity=external_identity,
        event_type=manifest_mod.STABLE_DM_ROOT_MANIFEST_WITNESS_EVENT_TYPE,
        payload=manifest_mod._witness_payload(stale_distribution["manifest"]),
    )
    package_path.write_text(
        manifest_mod._stable_json(
            {
                "type": manifest_mod.STABLE_DM_ROOT_MANIFEST_EXTERNAL_WITNESS_IMPORT_TYPE,
                "schema_version": 1,
                "source_scope": "https_fetch",
                "source_label": "witness-a",
                "exported_at": int(time.time()),
                "descriptors": [manifest_mod._public_witness_descriptor(external_identity)],
                "manifest_fingerprint": stale_distribution["manifest_fingerprint"],
                "witnesses": [refreshed_receipt],
            }
        ),
        encoding="utf-8",
    )

    refreshed_distribution = manifest_mod.get_current_root_manifest()
    refreshed_transparency = transparency_mod.get_current_root_transparency_record(distribution=refreshed_distribution)
    recovered_export = identity_mod.export_wormhole_dm_invite()
    recovered_import = identity_mod.import_wormhole_dm_invite(recovered_export["invite"], alias="alice-recovered")

    assert refreshed_distribution["external_witness_operator_state"] == "current"
    assert refreshed_distribution["external_witness_reacquire_required"] is False
    assert refreshed_transparency["ledger_operator_state"] == "current"
    assert recovered_import["ok"] is True


def test_compat_dm_invite_import_is_blocked_by_default(tmp_path, monkeypatch):
    _relay, identity_mod, contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    invite, verified = _export_compat_invite(identity_mod)

    imported = identity_mod.import_wormhole_dm_invite(invite, alias="compat")

    assert imported["ok"] is False
    assert imported["detail"] == "compat dm invite import disabled; ask the sender to re-export a current signed invite"
    assert verified["ok"] is True
    assert contacts_mod.list_wormhole_dm_contacts() == {}


def test_compat_dm_invite_import_downgrades_to_tofu_pinned_when_compat_enabled(tmp_path, monkeypatch):
    _relay, identity_mod, contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    invite, verified = _export_compat_invite(identity_mod)
    monkeypatch.setenv("MESH_DEV_ALLOW_LEGACY_COMPAT", "true")
    monkeypatch.setenv("MESH_ALLOW_COMPAT_DM_INVITE_IMPORT_UNTIL", "2099-01-01")
    get_settings.cache_clear()

    try:
        imported = identity_mod.import_wormhole_dm_invite(invite, alias="compat")
    finally:
        get_settings.cache_clear()

    contact = imported["contact"]
    local_identity = identity_mod.read_wormhole_identity()

    assert imported["ok"] is True
    assert imported["invite_attested"] is False
    assert imported["detail"] == "legacy invite imported as tofu_pinned; SAS verification required before first contact"
    assert imported["peer_id"] == local_identity["node_id"]
    assert imported["invite_peer_id"] == verified["peer_id"]
    assert imported["trust_level"] == "tofu_pinned"
    assert contact["trust_level"] == "tofu_pinned"
    assert contact["trustSummary"]["state"] == "tofu_pinned"
    assert contact["alias"] == "compat"
    assert contact["remotePrekeyFingerprint"] == imported["trust_fingerprint"]
    assert contact["invitePinnedTrustFingerprint"] == ""
    assert contact["invitePinnedAt"] == 0
    assert contact["invitePinnedPrekeyLookupHandle"] == invite["payload"]["prekey_lookup_handle"]
    assert contacts_mod.list_wormhole_dm_contacts()[imported["peer_id"]]["trust_level"] == "tofu_pinned"


def test_legacy_dm_invite_import_is_blocked_by_default(tmp_path, monkeypatch):
    _relay, identity_mod, contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    invite, _verified = _export_legacy_invite(identity_mod)

    imported = identity_mod.import_wormhole_dm_invite(invite, alias="legacy")

    assert imported["ok"] is False
    assert imported["detail"] == "legacy dm invite import disabled; ask the sender to re-export a current signed invite"
    assert contacts_mod.list_wormhole_dm_contacts() == {}


def test_legacy_dm_invite_import_downgrades_to_tofu_pinned_when_compat_enabled(tmp_path, monkeypatch):
    _relay, identity_mod, contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    invite, _verified = _export_legacy_invite(identity_mod)
    monkeypatch.setenv("MESH_DEV_ALLOW_LEGACY_COMPAT", "true")
    monkeypatch.setenv("MESH_ALLOW_COMPAT_DM_INVITE_IMPORT_UNTIL", "2099-01-01")
    get_settings.cache_clear()

    try:
        imported = identity_mod.import_wormhole_dm_invite(invite, alias="legacy")
    finally:
        get_settings.cache_clear()

    contact = imported["contact"]
    local_identity = identity_mod.read_wormhole_identity()

    assert imported["ok"] is True
    assert imported["invite_attested"] is False
    assert imported["detail"] == "legacy invite imported as tofu_pinned; SAS verification required before first contact"
    assert imported["peer_id"] == local_identity["node_id"]
    assert imported["trust_level"] == "tofu_pinned"
    assert contact["trust_level"] == "tofu_pinned"
    assert contact["trustSummary"]["state"] == "tofu_pinned"
    assert contact["alias"] == "legacy"
    assert contact["remotePrekeyFingerprint"] == imported["trust_fingerprint"]
    assert contact["invitePinnedTrustFingerprint"] == ""
    assert contact["invitePinnedAt"] == 0
    assert contact["invitePinnedPrekeyLookupHandle"] == ""
    assert contacts_mod.list_wormhole_dm_contacts()[imported["peer_id"]]["trust_level"] == "tofu_pinned"


def test_invite_pinned_contact_can_upgrade_to_sas_verified(tmp_path, monkeypatch):
    _relay, identity_mod, contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    _exported, _verified, imported = _import_invite(identity_mod)
    monkeypatch.setattr(
        contacts_mod,
        "_derive_expected_contact_sas_phrase",
        lambda *_args, **_kwargs: {"ok": True, "phrase": "able acid", "peer_ref": imported["peer_id"], "words": 2},
    )

    result = contacts_mod.confirm_sas_verification(imported["peer_id"], "able acid")

    assert result["ok"] is True
    assert result["trust_level"] == "sas_verified"


def test_invite_pinned_mismatch_becomes_continuity_broken_and_ack_rejects(tmp_path, monkeypatch):
    _relay, identity_mod, contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    _exported, _verified, imported = _import_invite(identity_mod)

    mismatch = contacts_mod.observe_remote_prekey_identity(
        imported["peer_id"],
        fingerprint="ff" * 32,
        sequence=2,
        signed_at=int(time.time()),
    )
    ack = contacts_mod.acknowledge_changed_fingerprint(imported["peer_id"])

    assert mismatch["trust_level"] == "continuity_broken"
    assert ack["ok"] is False
    assert "invite-pinned" in ack["detail"]


def test_reimport_with_changed_root_fails_closed_and_marks_continuity_broken(tmp_path, monkeypatch):
    _relay, identity_mod, contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    _exported, _verified, imported = _import_invite(identity_mod)

    from services.mesh import mesh_wormhole_persona

    persona_state = mesh_wormhole_persona.read_wormhole_persona_state()
    persona_state["previous_root_identity"] = {
        **dict(persona_state.get("root_identity") or {}),
        "scope": "previous_root",
    }
    persona_state["root_identity"] = mesh_wormhole_persona._identity_record(scope="root", label="root")
    mesh_wormhole_persona._write_wormhole_persona_state(persona_state)

    rotated = identity_mod.export_wormhole_dm_invite()
    assert rotated["ok"] is True

    result = identity_mod.import_wormhole_dm_invite(rotated["invite"], alias="alice-reimport")

    assert result["ok"] is False
    assert result["trust_level"] == "continuity_broken"
    assert "root continuity mismatch" in result["detail"]
    assert result["contact"]["trustSummary"]["state"] == "continuity_broken"
    assert result["contact"]["trustSummary"]["rootMismatch"] is True
    refreshed = contacts_mod.list_wormhole_dm_contacts()[imported["peer_id"]]
    assert refreshed["trust_level"] == "continuity_broken"


def test_compose_wormhole_dm_fails_closed_on_pinned_invite_mismatch(tmp_path, monkeypatch):
    _relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    _exported, _verified, imported = _import_invite(identity_mod)
    local_identity = identity_mod.read_wormhole_identity()

    import main

    monkeypatch.setattr(main, "_resolve_dm_aliases", lambda **_kw: ("local", "remote"))
    monkeypatch.setattr(main, "has_mls_dm_session", lambda *_a, **_kw: {"ok": True, "exists": False})
    monkeypatch.setattr(
        main,
        "fetch_dm_prekey_bundle",
        lambda _peer_id: {
            "ok": True,
            "agent_id": imported["peer_id"],
            "identity_dh_pub_key": local_identity["dh_pub_key"],
            "public_key": local_identity["public_key"],
            "public_key_algo": local_identity["public_key_algo"],
            "protocol_version": PROTOCOL_VERSION,
            "sequence": 2,
            "signed_at": int(time.time()),
            "mls_key_package": "ZmFrZQ==",
            "trust_fingerprint": "ff" * 32,
        },
    )

    result = main.compose_wormhole_dm(
        peer_id=imported["peer_id"],
        peer_dh_pub=local_identity["dh_pub_key"],
        plaintext="hello",
    )

    assert result["ok"] is False
    assert result["trust_level"] == "continuity_broken"


def test_compose_wormhole_dm_blocks_unverified_first_contact(tmp_path, monkeypatch):
    _relay, _identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)

    import main

    initiated = {"called": False}
    monkeypatch.setattr(main, "_resolve_dm_aliases", lambda **_kw: ("local", "remote"))
    monkeypatch.setattr(main, "has_mls_dm_session", lambda *_a, **_kw: {"ok": True, "exists": False})
    monkeypatch.setattr(
        main,
        "initiate_mls_dm_session",
        lambda *_a, **_kw: initiated.__setitem__("called", True) or {"ok": True, "welcome": "welcome"},
    )
    monkeypatch.setattr(
        main,
        "fetch_dm_prekey_bundle",
        lambda _peer_id: {
            "ok": True,
            "agent_id": "peer-unverified",
            "identity_dh_pub_key": "peer-dh-pub",
            "public_key": "peer-signing-pub",
            "public_key_algo": "Ed25519",
            "protocol_version": "infonet/2",
            "sequence": 2,
            "signed_at": int(time.time()),
            "mls_key_package": "ZmFrZQ==",
            "trust_fingerprint": "11" * 32,
        },
    )

    result = main.compose_wormhole_dm(
        peer_id="peer-unverified",
        peer_dh_pub="peer-dh-pub",
        plaintext="hello",
    )

    assert result["ok"] is False
    assert result["detail"] == "signed invite or SAS verification required before secure first contact"
    assert result["trust_level"] == "tofu_pinned"
    assert initiated["called"] is False


def test_compose_wormhole_dm_blocks_legacy_fallback_for_invite_scoped_contact(tmp_path, monkeypatch):
    _relay, identity_mod, _contacts_mod, _prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    _exported, _verified, imported = _import_invite(identity_mod)

    import main

    legacy_called = {"value": False}
    monkeypatch.setattr(main, "_resolve_dm_aliases", lambda **_kw: ("local", "remote"))
    monkeypatch.setattr(main, "has_mls_dm_session", lambda *_a, **_kw: {"ok": True, "exists": False})
    monkeypatch.setattr(main, "fetch_dm_prekey_bundle", lambda _peer_id: {"ok": False, "detail": "Prekey bundle not found"})
    monkeypatch.setattr(
        main,
        "encrypt_wormhole_dm",
        lambda **_kwargs: legacy_called.__setitem__("value", True) or {"ok": True, "result": "legacy"},
    )

    result = main.compose_wormhole_dm(
        peer_id=imported["peer_id"],
        peer_dh_pub="fallback-dh",
        plaintext="hello",
    )

    assert result["ok"] is False
    assert result["detail"] == "invite-scoped bootstrap required; legacy DM fallback disabled"
    assert result["trust_level"] == "invite_pinned"
    assert legacy_called["value"] is False


def test_bootstrap_encrypt_fails_closed_without_claiming_otk_on_pinned_invite_mismatch(tmp_path, monkeypatch):
    relay, identity_mod, _contacts_mod, prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    registered = prekey_mod.register_wormhole_prekey_bundle(force_signed_prekey=True)
    assert registered["ok"] is True

    _exported, _verified, imported = _import_invite(identity_mod)

    agent_id = registered["agent_id"]
    before = len(list((relay.get_prekey_bundle(agent_id) or {}).get("bundle", {}).get("one_time_prekeys") or []))
    tampered_bundle = dict((relay.get_prekey_bundle(agent_id) or {}).get("bundle") or {})
    tampered_bundle["identity_dh_pub_key"] = _b64_pub(x25519.X25519PrivateKey.generate().public_key())
    tampered_bundle = prekey_mod._attach_bundle_signature(tampered_bundle, signed_at=int(time.time()))
    relay._prekey_bundles[agent_id]["bundle"] = tampered_bundle

    result = prekey_mod.bootstrap_encrypt_for_peer(agent_id, "hello")
    after = len(list((relay.get_prekey_bundle(agent_id) or {}).get("bundle", {}).get("one_time_prekeys") or []))

    assert result["ok"] is False
    assert result["trust_level"] == "continuity_broken"
    assert before == after


def test_bootstrap_encrypt_blocks_unverified_first_contact_without_claiming_otk(tmp_path, monkeypatch):
    from services.config import get_settings

    relay, _identity_mod, _contacts_mod, prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    registered = prekey_mod.register_wormhole_prekey_bundle(force_signed_prekey=True)
    assert registered["ok"] is True

    agent_id = registered["agent_id"]
    before = len(list((relay.get_prekey_bundle(agent_id) or {}).get("bundle", {}).get("one_time_prekeys") or []))
    monkeypatch.setenv("MESH_BLOCK_LEGACY_AGENT_ID_LOOKUP", "false")
    get_settings.cache_clear()

    try:
        result = prekey_mod.bootstrap_encrypt_for_peer(agent_id, "hello")
        after = len(list((relay.get_prekey_bundle(agent_id) or {}).get("bundle", {}).get("one_time_prekeys") or []))

        assert result["ok"] is False
        assert result["detail"] == "legacy agent_id lookup disabled; use invite lookup handle"
        assert before == after
    finally:
        get_settings.cache_clear()


def test_bootstrap_encrypt_requires_independent_quorum_finality_when_enforced(tmp_path, monkeypatch):
    relay, identity_mod, _contacts_mod, prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    registered = prekey_mod.register_wormhole_prekey_bundle(force_signed_prekey=True)
    assert registered["ok"] is True
    _exported, _verified, imported = _import_invite(identity_mod)

    agent_id = registered["agent_id"]
    before = len(list((relay.get_prekey_bundle(agent_id) or {}).get("bundle", {}).get("one_time_prekeys") or []))
    monkeypatch.setenv("WORMHOLE_ROOT_WITNESS_FINALITY_ENFORCE", "true")
    get_settings.cache_clear()

    try:
        result = prekey_mod.bootstrap_encrypt_for_peer(agent_id, "hello")
        after = len(list((relay.get_prekey_bundle(agent_id) or {}).get("bundle", {}).get("one_time_prekeys") or []))

        assert result["ok"] is False
        assert result["detail"] == "independent quorum root witness finality required before secure first contact"
        assert result["trust_level"] == "invite_pinned"
        assert imported["contact"]["trustSummary"]["rootWitnessProvenanceState"] == "local_quorum"
        assert before == after
    finally:
        get_settings.cache_clear()


def test_bootstrap_decrypt_rejects_sender_static_key_that_mismatches_pinned_invite(tmp_path, monkeypatch):
    _relay, identity_mod, _contacts_mod, prekey_mod = _fresh_wormhole_state(tmp_path, monkeypatch)
    _exported, _verified, imported = _import_invite(identity_mod)

    fake_envelope = {
        "h": {
            "ik_pub": _b64_pub(x25519.X25519PrivateKey.generate().public_key()),
            "ek_pub": "ZmFrZQ==",
            "spk_id": 1,
            "otk_id": 0,
        },
        "ct": base64.b64encode(b"0" * 16).decode("ascii"),
    }
    ciphertext = "x3dh1:" + base64.b64encode(
        json.dumps(fake_envelope, sort_keys=True, separators=(",", ":")).encode("utf-8")
    ).decode("ascii")

    result = prekey_mod.bootstrap_decrypt_from_sender(imported["peer_id"], ciphertext)

    assert result["ok"] is False
    assert result["detail"] == "sender bootstrap key mismatches pinned invite"
