//! Side-by-side proof for peat-mesh#205:
//! `connect_and_authenticate(peer_id)` *requires* iroh's `address_lookup`
//! chain to resolve the peer's address (because we hand iroh just the id
//! and `impl From<EndpointId> for EndpointAddr` produces an `EndpointAddr`
//! with empty `addrs`); `connect_and_authenticate_with_addr(peer_addr)`
//! bypasses the chain entirely because we hand iroh the address up front.
//!
//! # What this proves
//!
//! On a single dialer endpoint whose `address_lookup` chain is empty (no
//! DNS, no MemoryLookup, no anything — equivalent to "the chain has
//! nothing to resolve with"):
//!
//! - `connect_and_authenticate(target_id)` → **must fail**. iroh wraps
//!   `target_id` as `EndpointAddr { id, addrs: {} }` and calls
//!   `resolve_remote`; with an empty chain, iroh yields
//!   `AddressLookupFailed::NoServiceConfigured` and the dial errors out.
//!   This is the exact same problem peat-cli's `connect_peer` hit in
//!   peat-node PR #114's airgapped docker-compose env — `MemoryLookup`
//!   was populated there, not empty, but iroh's dispatcher still didn't
//!   surface its item before the dial failed.
//! - `connect_and_authenticate_with_addr(target_addr)` → **must succeed**.
//!   iroh receives a populated `EndpointAddr`, `path_state.rs:150-152`
//!   short-circuits resolve to "addresses already known," and the dial
//!   proceeds directly. No chain involvement.
//!
//! Same endpoint, same chain state, opposite outcomes — the chain is
//! the variable, and the new method takes the chain out of the picture.
//!
//! # Why this matters for peat-node
//!
//! The peat-node QUICKSTART failure is the production-network shape of
//! exactly this contrast: the dialer's chain (DNS + populated
//! MemoryLookup) appears to not deliver MemoryLookup's item to iroh's
//! `resolve_remote` before the dial gives up. Whether that's an iroh
//! dispatcher bug or something else, the fix is the same: don't ask
//! iroh to resolve what we already know. This test pins the mechanism.

#![cfg(feature = "automerge-backend")]

use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::Arc;
use std::time::{Duration, Instant};

use iroh::{Endpoint, RelayMode};

use peat_mesh::security::FormationKey;
use peat_mesh::storage::MeshSyncTransport;
use peat_mesh::sync::{
    AutomergeBackend, AutomergeBackendConfig, BackendConfig, DataSyncBackend, SyncEngine,
};
use tempfile::TempDir;

/// Build a normal `AutomergeBackend` for the dial *target* using the
/// standard `with_iroh()` path. Gives a working ALPN handler that can
/// accept the inbound `CAP_AUTOMERGE_ALPN` connection.
async fn build_target(formation_id: &str, shared_key: &str) -> (Arc<AutomergeBackend>, TempDir) {
    let dir = TempDir::new().expect("tempdir");
    let mut cfg = AutomergeBackendConfig::default();
    cfg.data_dir = dir.path().to_path_buf();
    cfg.formation_id = formation_id.to_string();
    cfg.base64_shared_key = shared_key.to_string();
    cfg.iroh_bind_addr = Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0));
    let backend = AutomergeBackend::with_iroh(cfg)
        .await
        .expect("with_iroh target");
    backend
        .initialize(BackendConfig {
            app_id: "airgap-target".to_string(),
            persistence_dir: dir.path().to_path_buf(),
            shared_key: None,
            transport: peat_mesh::sync::TransportConfig::default(),
            extra: std::collections::HashMap::new(),
        })
        .await
        .expect("initialize target");
    backend.start_sync().await.expect("start_sync target");
    (backend, dir)
}

/// Build the *dialer*'s iroh `Endpoint` with a deliberately empty
/// `address_lookup` chain — no DNS, no Pkarr, no MemoryLookup. This
/// is the cleanest possible "chain has nothing to resolve with" state.
///
/// `iroh::endpoint::presets::Empty` adds nothing of its own; we only
/// add the FIPS-aligned crypto provider, disable relay (no internet
/// dependency), and bind to localhost.
async fn build_dialer_endpoint_with_empty_chain() -> Endpoint {
    Endpoint::builder(iroh::endpoint::presets::Empty)
        .crypto_provider(peat_mesh::security::tls_provider::iroh_quic_provider())
        .relay_mode(RelayMode::Disabled)
        .bind_addr(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0))
        .expect("bind_addr")
        .bind()
        .await
        .expect("bind dialer endpoint")
}

/// The side-by-side proof:
///
/// 1. **Old method on empty chain → must fail.** With nothing in the
///    `address_lookup` chain, iroh has no way to resolve `target_id`
///    to a socket address. `resolve_remote` yields
///    `AddressLookupFailed::NoServiceConfigured`. The dial errors out.
/// 2. **New method on the same empty chain → must succeed.** The
///    populated `EndpointAddr` goes straight to
///    `endpoint.connect(addr, ALPN)`. iroh's
///    `socket/remote_map/remote_state/path_state.rs:150-152` empty→
///    non-empty short-circuit emits the pending-resolve completion
///    immediately. The dial proceeds.
///
/// Same `MeshSyncTransport`, same iroh `Endpoint`, same empty chain.
/// The only variable is which method is called. That is the proof.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn old_method_requires_chain_new_method_bypasses_it() {
    let shared_key = FormationKey::generate_secret();
    let formation_id = "peat-mesh-205-bypass-proof";

    // Target: real backend, full ALPN handler, normal `with_iroh()` setup.
    let (target_backend, _target_dir) = build_target(formation_id, &shared_key).await;
    let target_addr = target_backend.blob_store().endpoint().addr();
    let target_id = target_backend.blob_store().endpoint().id();
    assert!(
        !target_addr.addrs.is_empty(),
        "target backend must have bound at least one socket — got empty EndpointAddr"
    );

    // Dialer: iroh `Endpoint` with an empty `address_lookup` chain.
    // Wrapped in `MeshSyncTransport` with the matching `FormationKey`.
    let dialer_endpoint = build_dialer_endpoint_with_empty_chain().await;
    let dialer_formation_key =
        FormationKey::from_base64(formation_id, &shared_key).expect("from_base64");
    let dialer_transport = Arc::new(MeshSyncTransport::new(
        dialer_endpoint,
        dialer_formation_key,
    ));

    // ---- Property 1: old method (peer_id-only) fails on the empty chain ----
    //
    // 30 s is generous — iroh's `NoServiceConfigured` yields fast in
    // practice, but we don't want a slow CI to false-fail. The
    // important property is that the call returns `Err`, not that it
    // returns quickly.
    let old_start = Instant::now();
    let old_result = tokio::time::timeout(
        Duration::from_secs(30),
        dialer_transport.connect_and_authenticate(target_id),
    )
    .await;
    let old_elapsed = old_start.elapsed();

    match old_result {
        // Timeout — old method hung indefinitely waiting for chain
        // resolution that will never come. Acceptable failure shape.
        Err(_) => {
            eprintln!(
                "✓ Proof half 1: connect_and_authenticate(peer_id) hung past timeout \
                 with an empty chain (no resolution path). elapsed={:?}",
                old_elapsed
            );
        }
        Ok(Err(err)) => {
            // iroh yielded an error from the chain (most likely
            // `AddressLookupFailed::NoServiceConfigured`). The exact
            // error string is iroh-version-dependent so we don't pin
            // it — the important property is that the call returned
            // `Err`, which is the "old method requires the chain"
            // signal we wanted.
            eprintln!(
                "✓ Proof half 1: connect_and_authenticate(peer_id) returned Err \
                 with an empty chain. error={err:#}, elapsed={:?}",
                old_elapsed
            );
        }
        Ok(Ok(conn)) => {
            panic!(
                "old method (peer_id-only) UNEXPECTEDLY SUCCEEDED with an empty \
                 address_lookup chain in {:?}. This shouldn't happen — iroh has \
                 no way to resolve {} without a chain entry. Either the dialer \
                 endpoint has more in its chain than intended, or iroh has a \
                 cache we're hitting. Connection: remote_id={}",
                old_elapsed,
                target_id.fmt_short(),
                conn.remote_id().fmt_short()
            );
        }
    }

    // ---- Property 2: new method (full addr) succeeds on the SAME chain ----
    //
    // The same dialer_transport — same iroh Endpoint, same empty
    // chain — must succeed when given the populated `EndpointAddr`
    // directly. 15 s is comfortable for localhost.
    let new_start = Instant::now();
    let new_result = tokio::time::timeout(
        Duration::from_secs(15),
        dialer_transport.connect_and_authenticate_with_addr(target_addr.clone()),
    )
    .await;
    let new_elapsed = new_start.elapsed();

    let conn = new_result
        .expect(
            "connect_and_authenticate_with_addr timed out on the SAME empty chain \
             where the old method also failed. The new method must NOT depend on \
             the chain — if it does, the bypass property is broken (peat-mesh#205).",
        )
        .expect("connect_and_authenticate_with_addr returned Err on the empty chain");

    eprintln!(
        "✓ Proof half 2: connect_and_authenticate_with_addr(peer_addr) succeeded \
         on the SAME empty chain in {:?}. Bypass property holds.",
        new_elapsed
    );

    assert_eq!(conn.remote_id(), target_id, "connected to wrong peer");
    assert!(
        conn.close_reason().is_none(),
        "connection closed immediately after handshake: {:?}",
        conn.close_reason()
    );

    eprintln!(
        "Timing summary: old={:?} (failed, chain required) vs new={:?} (succeeded, chain bypassed)",
        old_elapsed, new_elapsed
    );
}

/// Pin the empty-`addrs` guard added to `connect_and_authenticate_with_addr`.
///
/// The whole point of the new method is "the caller already knows the
/// socket; bypass the chain." If a caller hands in an `EndpointAddr`
/// with an empty `addrs` set, iroh would silently fall back through
/// the `address_lookup` chain — exactly the peat-mesh#205 failure
/// mode the method exists to prevent. The guard converts that latent
/// regression into an immediate, loud error at the misuse site.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn connect_with_addr_rejects_empty_addrs() {
    use iroh::EndpointAddr;

    let shared_key = FormationKey::generate_secret();
    let formation_id = "peat-mesh-205-empty-addrs-guard";

    // Build a target just to get a valid `EndpointId` to address with.
    let (target_backend, _target_dir) = build_target(formation_id, &shared_key).await;
    let target_id = target_backend.blob_store().endpoint().id();

    let dialer_endpoint = build_dialer_endpoint_with_empty_chain().await;
    let dialer_formation_key =
        FormationKey::from_base64(formation_id, &shared_key).expect("from_base64");
    let dialer_transport = Arc::new(MeshSyncTransport::new(
        dialer_endpoint,
        dialer_formation_key,
    ));

    // `EndpointAddr::new(id)` builds an addr with an empty `addrs` set —
    // exactly the misuse the guard exists to catch.
    let empty_addr = EndpointAddr::new(target_id);
    assert!(
        empty_addr.addrs.is_empty(),
        "this test only makes sense if EndpointAddr::new produces empty addrs; \
         iroh changed something — re-check the guard's semantics"
    );

    let err = dialer_transport
        .connect_and_authenticate_with_addr(empty_addr)
        .await
        .expect_err("empty addrs must fail the guard, not silently fall back to the chain");

    let msg = format!("{err:#}");
    let short_id = target_id.fmt_short().to_string();
    assert!(
        msg.contains("empty addrs"),
        "guard error must reference the empty-addrs misuse so the call site is \
         immediately diagnosable. got: {msg}"
    );
    assert!(
        msg.contains(&short_id),
        "guard error must reference the offending peer for triage. got: {msg}"
    );
}

/// Pin the **timing** of the empty-`addrs` guard: it must fail before any
/// I/O happens.
///
/// The guard's value is "fails loud at the misuse site **before** iroh
/// dials anything." That property is implicit in the source ordering (no
/// `.await` between the `if peer_addr.addrs.is_empty()` check and the
/// `bail!`), but isn't pinned by the rejection-test above — a future
/// refactor that moves the guard after `endpoint.connect(...).await`
/// would still cause `connect_with_addr_rejects_empty_addrs` to pass,
/// just for a different reason (the dial would fail downstream).
///
/// Pinning the timing as a measurable contract makes the "no I/O before
/// guard" property resilient against refactor. 50 ms is generous
/// (the guard is two `BTreeSet::is_empty()` reads and a format-string
/// allocation) but well below any plausible iroh dial cost — if the
/// guard ever started dialing, the assertion would fail.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn connect_with_addr_empty_guard_fails_before_any_io() {
    use iroh::EndpointAddr;

    let shared_key = FormationKey::generate_secret();
    let formation_id = "peat-mesh-205-guard-timing-pin";

    let (target_backend, _target_dir) = build_target(formation_id, &shared_key).await;
    let target_id = target_backend.blob_store().endpoint().id();

    let dialer_endpoint = build_dialer_endpoint_with_empty_chain().await;
    let dialer_formation_key =
        FormationKey::from_base64(formation_id, &shared_key).expect("from_base64");
    let dialer_transport = Arc::new(MeshSyncTransport::new(
        dialer_endpoint,
        dialer_formation_key,
    ));

    let empty_addr = EndpointAddr::new(target_id);

    let start = Instant::now();
    let result = dialer_transport
        .connect_and_authenticate_with_addr(empty_addr)
        .await;
    let elapsed = start.elapsed();

    assert!(
        result.is_err(),
        "guard must still reject empty addrs (sanity check)"
    );
    assert!(
        elapsed < Duration::from_millis(50),
        "guard must fail before any I/O (target: <50ms); actual: {:?}. \
         A regression here suggests the guard moved after endpoint.connect — \
         re-check src/storage/mesh_sync_transport.rs to ensure the empty-addrs \
         check is before any .await calls (peat-mesh#205 QA suggestion).",
        elapsed
    );
    eprintln!(
        "✓ Guard fails before I/O: connect_and_authenticate_with_addr(empty) returned \
         Err in {:?} (well under the 50ms I/O floor).",
        elapsed
    );
}

/// End-to-end sync round-trip over a connection established via the
/// adapter's airgap-safe path: `DataSyncBackend::connect_to_peer` with
/// non-empty `addresses` exercises
/// `connect_and_authenticate_with_addr(EndpointAddr)` under the hood
/// (per the consumer-surface half of the peat-mesh#205 fix).
///
/// This converts the test surface from "connect+auth handshake works"
/// to "a real Automerge sync workload survives the new dial path." The
/// downstream sync machinery is shared with the existing `EndpointId`
/// path that `two_backend_persistent_sync_e2e` already covers — but
/// before this test, no in-CI test exercised the with-addr adapter path
/// end-to-end.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn sync_round_trip_via_with_addr_adapter_path() {
    use peat_mesh::sync::Document;
    use std::collections::HashMap;

    let shared_key = FormationKey::generate_secret();
    let formation_id = "peat-mesh-205-round-trip";

    // Two real backends bound to localhost on OS-chosen ports. Both go
    // through `with_iroh()` so they have working ALPN handlers and the
    // full sync stack.
    let (backend_a, _dir_a) = build_target(formation_id, &shared_key).await;
    let (backend_b, _dir_b) = build_target(formation_id, &shared_key).await;

    let addr_b = backend_b.blob_store().endpoint().addr();
    let id_b = backend_b.blob_store().endpoint().id();
    let id_b_hex = id_b.to_string();

    // The address list that an operator (peat-cli or peat-node) would
    // hand the adapter. Format mirrors `EndpointAddr.addrs` round-tripped
    // through the iroh blob_store's helper at iroh_blob_store.rs:1076.
    let address_hints: Vec<String> = addr_b
        .addrs
        .iter()
        .filter_map(|t| match t {
            iroh::TransportAddr::Ip(sock) => Some(sock.to_string()),
            _ => None,
        })
        .collect();
    assert!(
        !address_hints.is_empty(),
        "backend B must have bound at least one IP — got empty TransportAddr set"
    );

    // CRITICAL: do NOT populate MemoryLookup on A. This forces
    // `connect_to_peer` to take the with_addr branch — the consumer-
    // surface half of the airgap-safe fix. If `connect_to_peer`
    // regresses to ignoring `addresses` (the pre-#205 behavior), the
    // chain has nothing to resolve B's id with, and the connect fails.

    // Call the adapter surface peat-cli / peat-node would use.
    let connected = backend_a
        .connect_to_peer(&id_b_hex, &address_hints)
        .await
        .expect("connect_to_peer (with_addr) must succeed");
    assert!(
        connected,
        "connect_to_peer returned Ok(false) — expected Ok(true) on the with_addr path"
    );

    // Push a doc on A; subscribe on B; assert it arrives within a
    // deadline. The deadline is generous (15 s) because localhost
    // sync is sub-second when wired correctly — a timeout here means
    // the dial succeeded but downstream sync isn't picking up the
    // connection.
    let doc_id: String = "round-trip-205".to_string();
    let mut payload = HashMap::new();
    payload.insert(
        "field".to_string(),
        serde_json::Value::String("peat-mesh-205-with-addr".to_string()),
    );
    let doc = Document::with_id(&doc_id, payload);

    backend_a
        .document_store()
        .upsert("e2e", doc.clone())
        .await
        .expect("upsert on A");

    let deadline = Instant::now() + Duration::from_secs(15);
    let observed = loop {
        if let Ok(Some(seen)) = backend_b.document_store().get("e2e", &doc_id).await {
            break seen;
        }
        if Instant::now() >= deadline {
            panic!(
                "doc {doc_id} did not arrive on backend B within 15 s. \
                 Either the with_addr adapter path didn't actually establish \
                 the sync-capable connection, or downstream sync isn't \
                 observing the new connection."
            );
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    };

    assert_eq!(
        observed.id.as_deref(),
        Some(doc_id.as_str()),
        "observed doc id mismatch on B (expected {doc_id})"
    );
    eprintln!(
        "✓ Sync round-trip via with_addr adapter path: doc {} arrived on B in <15s",
        doc_id
    );
}

/// Pin the **accept-side initial-sync push** introduced by `cfa9023`.
///
/// Pre-fix, `SyncProtocolHandler::accept` registered the inbound
/// connection and spawned the sync-receiver loop but never proactively
/// pushed its own docs to the new peer. The asymmetry (only the dial
/// side called `sync_all_documents_with_peer`) was latent under
/// sidecar-pair tests because the dialer typically had docs to push
/// too. It surfaced the moment an empty-store dialer (peat-cli or any
/// newly-spawned ephemeral consumer) dialed an established sidecar —
/// the dialer iterated zero local docs, the acceptor never reciprocated,
/// and the dialer's store stayed empty even though the connection was
/// authenticated and idle.
///
/// This test exactly reproduces that asymmetry in peat-mesh's own CI:
///
///   - Backend B (the **acceptor**) upserts a doc *before* A connects.
///   - Backend A (the **dialer**) is fresh — empty store.
///   - A dials B via `connect_and_authenticate_with_addr` followed by
///     `start_sync_connection`, mirroring the peat-cli/peat-node consumer
///     pattern (the `DataSyncBackend::connect_to_peer` adapter only does
///     the first step; consumers wire the second step manually).
///   - A polls its own `document_store` for B's doc.
///   - Pre-fix: A never sees the doc. The dial-side
///     `sync_all_documents_with_peer` ran against A's empty store
///     (`Syncing 0 existing documents`), B's accept handler never
///     pushed, and A's poll times out.
///   - Post-fix: B's accept handler spawns a one-shot
///     `sync_all_documents_with_peer(A)` after `start_sync_connection`,
///     B's doc reaches A, A's poll observes it.
///
/// If a future refactor drops the accept-side spawn, this test catches
/// it deterministically — the dialer-side push has nothing to do, so
/// the accept-side path is the only way the doc can reach A.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn accept_side_pushes_existing_docs_to_fresh_dialer() {
    use peat_mesh::sync::Document;
    use std::collections::HashMap;

    let shared_key = FormationKey::generate_secret();
    let formation_id = "peat-mesh-205-accept-side-push";

    // A is the dialer (fresh, empty store). B is the acceptor (will have a doc
    // already in its store BEFORE A connects).
    let (backend_a, _dir_a) = build_target(formation_id, &shared_key).await;
    let (backend_b, _dir_b) = build_target(formation_id, &shared_key).await;

    let addr_b = backend_b.blob_store().endpoint().addr();
    let id_b = backend_b.blob_store().endpoint().id();

    let address_hints: Vec<String> = addr_b
        .addrs
        .iter()
        .filter_map(|t| match t {
            iroh::TransportAddr::Ip(sock) => Some(sock.to_string()),
            _ => None,
        })
        .collect();
    assert!(
        !address_hints.is_empty(),
        "backend B must have bound at least one IP — got empty TransportAddr set"
    );

    // Pre-populate B's store BEFORE A connects. This is the scenario
    // that exposed the bug: an established peer with docs accepts a
    // connection from a fresh dialer.
    let doc_id: String = "accept-push-doc".to_string();
    let mut payload = HashMap::new();
    payload.insert(
        "field".to_string(),
        serde_json::Value::String("written-on-B-before-A-connects".to_string()),
    );
    let doc = Document::with_id(&doc_id, payload);
    backend_b
        .document_store()
        .upsert("e2e", doc)
        .await
        .expect("upsert on B (pre-connect)");

    // A's store is intentionally empty — that's the asymmetry that
    // exposed the bug. The dial-side `sync_all_documents_with_peer`
    // (peat-cli's join.rs / peat-node's start_sync) will iterate zero
    // docs on A. The ONLY way B's doc reaches A is the accept-side
    // push that `SyncProtocolHandler::accept` now spawns. We assert
    // that property by reading the specific doc id directly: pre-connect
    // it must not exist on A.
    let a_pre_connect = backend_a
        .document_store()
        .get("e2e", &doc_id)
        .await
        .expect("get on A pre-connect");
    assert!(
        a_pre_connect.is_none(),
        "test premise: A must not have B's doc pre-connect (the bug needs a fresh dialer)"
    );

    // A dials B using the same low-level pattern peat-cli and peat-node
    // wire up manually (`crates/peat-cli/src/join.rs:connect_peer` and
    // `peat-node/src/node.rs:dial_and_attach`):
    //
    //   1. transport.connect_and_authenticate_with_addr → returns Connection
    //   2. transport.start_sync_connection(Connection, coordinator)
    //      → spawns the dial-side `accept_bi()` receiver loop so the
    //        dialer can RECEIVE incoming sync streams from the peer.
    //        Without this, B's accept-side push can't reach A even
    //        though the connection is up.
    //
    // The `DataSyncBackend::connect_to_peer` adapter only does step 1
    // (intentionally — it's a thin wrapper consumers extend). For a
    // realistic reproduction we mirror the consumer pattern.
    let address_hint = address_hints
        .first()
        .expect("at least one address hint")
        .clone();
    let socket_addr: std::net::SocketAddr = address_hint.parse().expect("parse address hint");
    let mut peer_addr = iroh::EndpointAddr::new(id_b);
    peer_addr.addrs.insert(iroh::TransportAddr::Ip(socket_addr));
    let connection = backend_a
        .transport()
        .connect_and_authenticate_with_addr(peer_addr)
        .await
        .expect("connect_and_authenticate_with_addr A → B");
    backend_a
        .transport()
        .start_sync_connection(connection, Arc::clone(backend_a.coordinator()));

    // Poll A's store for B's doc. Pre-fix this times out at deadline
    // because nothing pushes B's doc to A. Post-fix it converges in
    // sub-second on loopback. 10 s budget absorbs CI noise.
    let deadline = Instant::now() + Duration::from_secs(10);
    let observed = loop {
        if let Ok(Some(doc)) = backend_a.document_store().get("e2e", &doc_id).await {
            break doc;
        }
        if Instant::now() >= deadline {
            panic!(
                "fresh dialer A did not observe B's pre-connect doc {doc_id} within 10 s. \
                 The accept-side spawn in SyncProtocolHandler::accept either didn't fire or \
                 didn't reach `sync_all_documents_with_peer`. Pre-#205-cfa9023 the same path \
                 would silently time out — if this test fails today, the spawn at \
                 `src/storage/mesh_sync_transport.rs::SyncProtocolHandler::accept` has regressed."
            );
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    };

    assert_eq!(
        observed.id.as_deref(),
        Some(doc_id.as_str()),
        "observed doc id mismatch on A (expected {doc_id})"
    );
    eprintln!(
        "✓ Accept-side push: B's pre-connect doc {} arrived on fresh dialer A in <10s",
        doc_id
    );
}
