//! Relm4-agnostic client facade over `zodia-net` + `zodia-sync` +
//! `zodia-pipeline` — see `docs/prd/zodia-sdk.md`.
//!
//! `p2panda-stream` (used inside `zodia-pipeline`) is `!Send`, so the whole
//! network/sync/pipeline stack lives on one dedicated OS thread running a
//! single-threaded tokio runtime + `LocalSet`. Everything [`ZodiaClient`]
//! exposes outward — the command methods and the [`StateEvent`] stream —
//! crosses that boundary through ordinary `Send` channels, so the caller
//! can be any async runtime (or none): relm4/glib, plain tokio, a test.

use std::collections::HashMap;
use std::path::PathBuf;
use std::thread;

use ed25519_dalek::SigningKey;
use p2panda_core::{Hash, SigningKey as PandaSigningKey, Timestamp, Topic, VerifyingKey};
use thiserror::Error;
use tokio::sync::{broadcast, mpsc, oneshot, watch};
use tracing::warn;

use p2panda_net::{Endpoint, Gossip};
use zodia_circles::{Access, SpaceId};
use zodia_core::{BirthData, topic_key_global};
use zodia_net::{NetworkConfig, PeerId, ZodiaNetwork};
use zodia_ops::{DocOp, InterpOp};
use zodia_pipeline::ZodiaPipeline;
use zodia_sync::{SyncEvent, SyncError, ZodiaSyncNode};

pub use zodia_circles::{Access as CircleAccess, SpaceId as CircleId};
pub use zodia_pipeline::StateEvent;

/// Reserved `set_display_name` value automated test/dev peers should
/// broadcast so real clients can filter them out of discovery UI instead
/// of showing throwaway test identities as if they were real people to
/// connect with. Not a security boundary — any peer can claim this name —
/// just a courtesy convention for tooling that spins up ephemeral
/// identities on shared infrastructure (mDNS reaches the whole LAN, so a
/// local test run is visible to every real Zodia instance nearby). See
/// `sdk/tests/cucumber.rs`'s `peer_connected` step for the producer side.
pub const TEST_PEER_DISPLAY_NAME: &str = "zodia-test-peer";

/// Per-peer catch-up lifecycle, raw (unlike [`SyncStatus`]'s aggregate
/// counts) — for UIs that want to show per-peer detail (e.g. a "Sync
/// activity" panel listing which peers are caught up and how many ops
/// each sent), not just a summary count.
#[derive(Debug, Clone)]
pub enum SyncLifecycleEvent {
    Started { remote: [u8; 32] },
    Finished { remote: [u8; 32], received_ops: u64 },
    Failed { remote: [u8; 32], error: String },
}

// ── public config / errors ──────────────────────────────────────────────────

/// Everything needed to bring a `ZodiaClient` up.
pub struct ZodiaClientConfig {
    pub signing_key: SigningKey,
    pub birth:       BirthData,
    pub data_dir:    PathBuf,
}

#[derive(Debug, Error)]
pub enum ClientError {
    /// The client's background thread is gone — either it never started
    /// successfully, or the `ZodiaClient` handle was dropped.
    #[error("client thread is gone")]
    Disconnected,
    #[error("network: {0}")]
    Network(String),
    #[error("sync: {0}")]
    Sync(String),
}

/// Catch-up state, derived from `SyncStarted`/`SyncFinished`/`Failed`
/// lifecycle events per peer. Snapshot, not a log — only the latest state
/// matters to a UI status panel.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct SyncStatus {
    pub peers_known:     usize,
    pub peers_caught_up: usize,
}

// ── client handle ────────────────────────────────────────────────────────────

/// `Send + 'static` handle to a running Zodia network/sync/pipeline stack.
/// Cloneable-in-spirit via `events()`/`sync_status()` (each call gets its
/// own receiver); the command methods go through one shared channel to the
/// background thread.
pub struct ZodiaClient {
    cmd_tx:       mpsc::Sender<Command>,
    events_tx:    broadcast::Sender<StateEvent>,
    lifecycle_tx: broadcast::Sender<SyncLifecycleEvent>,
    status_rx:    watch::Receiver<SyncStatus>,
    node_id:      PeerId,
    #[cfg(test)]
    thread: Option<thread::JoinHandle<()>>,
}

impl ZodiaClient {
    /// Spawn the dedicated thread, bring up network + sync + pipeline, and
    /// return once the endpoint is live. Callable from any async runtime.
    /// Owns a fresh `ZodiaNetwork` internally — if the caller also needs
    /// `ZodiaNetwork` directly (e.g. for Tier-1 consent/chat/AV, which this
    /// SDK doesn't cover), use [`Self::attach`] instead so both share one
    /// endpoint rather than running two independent networks under the
    /// same identity.
    pub async fn connect(config: ZodiaClientConfig) -> Result<Self, ClientError> {
        Self::spawn_thread(NetworkSource::Owned(config)).await
    }

    /// Same as [`Self::connect`], but reuses an already-running
    /// `ZodiaNetwork`'s endpoint and gossip engine instead of spawning a
    /// new one. For a caller (like the relm4 app) that also needs
    /// `ZodiaNetwork` directly for Tier-1 consent/chat/AV — those aren't
    /// part of this SDK's scope, so the caller keeps owning `net` and
    /// handling its `ZodiaNetEvent`s itself; this just attaches the
    /// sync/pipeline layer onto the same transport.
    pub async fn attach(
        net:         &ZodiaNetwork,
        signing_key: SigningKey,
        data_dir:    PathBuf,
    ) -> Result<Self, ClientError> {
        Self::spawn_thread(NetworkSource::Attached {
            endpoint: net.endpoint(),
            gossip:   net.gossip(),
            node_id:  net.node_id(),
            signing_key,
            data_dir,
        }).await
    }

    async fn spawn_thread(source: NetworkSource) -> Result<Self, ClientError> {
        let (ready_tx, ready_rx) = oneshot::channel::<Result<ConnectReady, ClientError>>();
        let (cmd_tx, cmd_rx) = mpsc::channel::<Command>(64);
        let (events_tx, _events_rx) = broadcast::channel::<StateEvent>(256);
        let (lifecycle_tx, _lifecycle_rx) = broadcast::channel::<SyncLifecycleEvent>(256);
        let (status_tx, status_rx) = watch::channel(SyncStatus::default());

        let events_tx_bg = events_tx.clone();
        let lifecycle_tx_bg = lifecycle_tx.clone();
        let handle = thread::Builder::new()
            .name("zodia-sdk".into())
            .spawn(move || {
                let rt = tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                    .expect("zodia-sdk: build current-thread runtime");
                let local = tokio::task::LocalSet::new();
                local.block_on(&rt, run(source, cmd_rx, events_tx_bg, lifecycle_tx_bg, status_tx, ready_tx));
            })
            .expect("zodia-sdk: spawn dedicated thread");

        // Outside tests nothing joins this thread — it runs until `cmd_tx`
        // (held by `Self`, below) drops, then exits on its own.
        #[cfg(not(test))]
        let _ = &handle;

        let ready = ready_rx.await.map_err(|_| ClientError::Disconnected)?;
        let ConnectReady { node_id } = ready?;

        Ok(Self {
            cmd_tx,
            events_tx,
            lifecycle_tx,
            status_rx,
            node_id,
            #[cfg(test)]
            thread: Some(handle),
        })
    }

    /// Every materialised `StateEvent`, network-wide. `broadcast` so
    /// multiple listeners (feed view, bell badge, an open aspect page)
    /// can each hold their own receiver off one internal stream. A slow
    /// listener gets `RecvError::Lagged(n)`, not a stall of everyone else.
    pub fn events(&self) -> broadcast::Receiver<StateEvent> {
        self.events_tx.subscribe()
    }

    /// Latest catch-up snapshot. `watch`, not a log — only the current
    /// state matters for a status panel.
    pub fn sync_status(&self) -> watch::Receiver<SyncStatus> {
        self.status_rx.clone()
    }

    /// Raw per-peer catch-up lifecycle — for a UI that lists individual
    /// peers and their status, rather than just [`Self::sync_status`]'s
    /// aggregate counts. `broadcast`, same multi-listener rationale as
    /// [`Self::events`].
    pub fn sync_lifecycle_events(&self) -> broadcast::Receiver<SyncLifecycleEvent> {
        self.lifecycle_tx.subscribe()
    }

    pub fn node_id(&self) -> PeerId {
        self.node_id.clone()
    }

    /// Open a key's per-key sync topic (Phase C-2). Idempotent.
    pub async fn subscribe(&self, interp_key: &str) -> Result<(), ClientError> {
        self.call(|reply| Command::Subscribe { interp_key: interp_key.to_string(), reply }).await
    }

    /// Close a key's per-key sync topic. Idempotent.
    pub async fn unsubscribe(&self, interp_key: &str) -> Result<(), ClientError> {
        self.call(|reply| Command::Unsubscribe { interp_key: interp_key.to_string(), reply }).await
    }

    /// Subscribe to `interp_key` (idempotent, same as [`Self::subscribe`])
    /// and (re)start a `grace` countdown: if nothing touches this key
    /// again before `grace` elapses, it auto-unsubscribes. Call again
    /// whenever the key is still in active use (e.g. its page is still
    /// open) to keep the subscription alive — completes the lazy-subscribe
    /// half of Phase C-2 (docs/prd/granular-topic-subscription.md) for
    /// keys outside the user's always-subscribed own-chart set.
    pub async fn touch_subscription(&self, interp_key: &str, grace: std::time::Duration) -> Result<(), ClientError> {
        self.call(|reply| Command::TouchSubscription {
            interp_key: interp_key.to_string(), grace, reply,
        }).await
    }

    /// Permanently delete locally-stored operations older than `retention`,
    /// except anything authored by this device's own identity (own
    /// contributions are never pruned — see `zodia_sync`'s `prune_older_than`).
    /// Local-storage-only: peers who still have the pruned history are
    /// unaffected, and this device can re-receive it later via normal
    /// catch-up sync if it re-subscribes to the relevant topic. Returns the
    /// number of operations removed.
    ///
    /// This is a first slice of Phase D's pruning scope
    /// (docs/prd/operations-and-streams-rearchitecture.md) — it does not
    /// yet exempt content this device has merely *affirmed* (as opposed to
    /// authored), since correlating an affirmed doc revision back to the
    /// specific operations that produced it needs app/store-level state
    /// this SDK doesn't have. Deliberately deferred, not silently dropped —
    /// see docs/prd/pruning.md.
    pub async fn prune(&self, retention: std::time::Duration) -> Result<u64, ClientError> {
        let now_micros: u64 = Timestamp::now().into();
        let cutoff = Timestamp::new(now_micros.saturating_sub(retention.as_micros() as u64));
        let (reply_tx, reply_rx) = oneshot::channel();
        self.cmd_tx.send(Command::Prune { cutoff, reply: reply_tx }).await
            .map_err(|_| ClientError::Disconnected)?;
        reply_rx.await.map_err(|_| ClientError::Disconnected)?
    }

    /// Create a private circle. This device is always added with
    /// `Access::manage()` automatically (see `Manager::create_space`'s own
    /// doc comment) — `initial_members` is anyone *besides* yourself to
    /// add up front; `invite_to_circle` can add more later. Returns the
    /// new circle's id.
    pub async fn create_circle(
        &self,
        initial_members: Vec<(VerifyingKey, CircleAccess<()>)>,
    ) -> Result<CircleId, ClientError> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.cmd_tx.send(Command::CreateCircle { initial_members, reply: reply_tx }).await
            .map_err(|_| ClientError::Disconnected)?;
        reply_rx.await.map_err(|_| ClientError::Disconnected)?
    }

    /// Start listening to `circle_id`'s topic — call this once you've
    /// learned a circle's id (an invite notification, a shared link;
    /// out of scope here). Inviting someone only updates the inviter's
    /// own side; the invitee has to explicitly join to receive anything
    /// — see `zodia_sync::ZodiaSyncNode::open_circle`'s doc comment.
    pub async fn open_circle(&self, circle_id: CircleId) -> Result<(), ClientError> {
        self.call(|reply| Command::OpenCircle { circle_id, reply }).await
    }

    /// Grant `member` `access` in `circle_id`. `member` must be a peer
    /// whose key-bundle this device has already seen — true for any peer
    /// this device has ever been online alongside (see `zodia_sync`'s
    /// circle directory topic).
    pub async fn invite_to_circle(
        &self,
        circle_id: CircleId,
        member:    VerifyingKey,
        access:    CircleAccess<()>,
    ) -> Result<(), ClientError> {
        self.call(|reply| Command::InviteToCircle { circle_id, member, access, reply }).await
    }

    /// Revoke `member`'s access to `circle_id`. Key rotation for the
    /// remaining members happens automatically (see
    /// `docs/prd/circles.md`'s "Key rotation on revocation is automatic").
    pub async fn revoke_from_circle(&self, circle_id: CircleId, member: VerifyingKey) -> Result<(), ClientError> {
        self.call(|reply| Command::RevokeFromCircle { circle_id, member, reply }).await
    }

    /// Broadcast an invite notification for `recipient` (`InterpOp::CircleInviteNotify`,
    /// log 0) — separate from `invite_to_circle` itself, which only updates
    /// the inviter's own membership state. Call this right after a
    /// successful invite so the invitee's own client can surface an
    /// in-app "you've been invited" prompt, instead of the invitee having
    /// no way to learn the circle exists at all. Carries no circle name
    /// (see `InterpOp::CircleInviteNotify`'s doc comment on why) — every
    /// other connected peer also technically receives this since it rides
    /// the always-on global topic, but only `recipient`'s own client
    /// should act on it.
    pub async fn notify_circle_invite(&self, circle_id: CircleId, recipient: VerifyingKey) -> Result<(), ClientError> {
        self.call(|reply| Command::NotifyCircleInvite { circle_id, recipient, reply }).await
    }

    /// Broadcast a join notification (`InterpOp::CircleJoinNotify`, log 0)
    /// — call this right after a successful `open_circle` so existing
    /// members' clients can show "X joined" instead of only finding out
    /// passively the next time they happen to call `circle_members()`.
    /// Same "everyone technically receives this" reasoning as
    /// `notify_circle_invite`.
    pub async fn notify_circle_join(&self, circle_id: CircleId) -> Result<(), ClientError> {
        self.call(|reply| Command::NotifyCircleJoin { circle_id, reply }).await
    }

    /// Share an interpretation privately with `circle_id`'s current
    /// members instead of the public network — encodes the same
    /// `InterpOp::Author` the public path (`Self::author`) uses, so once
    /// decrypted it materialises into the identical `StateEvent::InterpAuthored`
    /// (see `zodia_pipeline::materialize_circle_content`).
    pub async fn share_interp_to_circle(
        &self,
        circle_id:  CircleId,
        interp_key: &str,
        body:       String,
    ) -> Result<(), ClientError> {
        let plaintext = InterpOp::Author { interp_key: interp_key.to_string(), body }.encode();
        self.call(|reply| Command::ShareToCircle { circle_id, plaintext, reply }).await
    }

    /// Current members of `circle_id` and their access levels.
    pub async fn circle_members(&self, circle_id: CircleId) -> Result<Vec<(VerifyingKey, CircleAccess<()>)>, ClientError> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.cmd_tx.send(Command::CircleMembers { circle_id, reply: reply_tx }).await
            .map_err(|_| ClientError::Disconnected)?;
        reply_rx.await.map_err(|_| ClientError::Disconnected)?
    }

    /// Legacy whole-interpretation authoring (`InterpOp::Author`, log 0).
    pub async fn author(&self, interp_key: &str, body: String) -> Result<(), ClientError> {
        self.call(|reply| Command::Author { interp_key: interp_key.to_string(), body, reply }).await
    }

    /// Revoke a previously authored `InterpOp::Author` (or anything else
    /// content-addressed the same way — `target_log_id` is
    /// `BLAKE3(interp_key || body)`, see `zodia_ops::InterpOp::Revoke`).
    /// Legacy model, log 0: propagates over the always-on global topic,
    /// not a per-key one — no subscription is needed to observe it.
    pub async fn revoke(&self, target_log_id: Hash) -> Result<(), ClientError> {
        self.call(|reply| Command::Revoke { target_log_id, reply }).await
    }

    /// Broadcast a self-chosen display name (`InterpOp::SetDisplayName`,
    /// log 0) — shown to other peers in place of their truncated hex
    /// pubkey for this identity. Same always-on-global-topic propagation
    /// as `author()`/`revoke()`; not re-sent automatically on reconnect,
    /// callers should call this again after every successful connect if
    /// they want it to stay visible to newly-met peers (mirrors
    /// `ZodiaNetwork::publish_announce()`'s per-connect pattern).
    pub async fn set_display_name(&self, name: String) -> Result<(), ClientError> {
        self.call(|reply| Command::SetDisplayName { name, reply }).await
    }

    /// Apply a CRDT edit to a key's collaborative doc.
    pub async fn edit(
        &self,
        interp_key:      &str,
        base_rev:        Hash,
        crdt_update:     Vec<u8>,
        affected_blocks: Vec<[u8; 16]>,
    ) -> Result<(), ClientError> {
        self.call(|reply| Command::Edit {
            interp_key: interp_key.to_string(), base_rev, crdt_update, affected_blocks, reply,
        }).await
    }

    /// Veto a specific edit, within the author-ring's window.
    pub async fn veto(&self, interp_key: &str, target_edit_op_id: Hash) -> Result<(), ClientError> {
        self.call(|reply| Command::Veto {
            interp_key: interp_key.to_string(), target_edit_op_id, reply,
        }).await
    }

    /// Affirm a specific revision of a key's doc.
    pub async fn affirm_rev(&self, interp_key: &str, target_rev: [u8; 32]) -> Result<(), ClientError> {
        self.call(|reply| Command::AffirmRev {
            interp_key: interp_key.to_string(), target_rev, reply,
        }).await
    }

    /// Presence heartbeat for a key's editor session.
    pub async fn set_editor_presence(&self, interp_key: &str, joined: bool) -> Result<(), ClientError> {
        self.call(|reply| Command::SetEditorPresence {
            interp_key: interp_key.to_string(), joined, reply,
        }).await
    }

    /// Send `build(reply_tx)` to the background thread and await its reply.
    /// Every command method above is this, mechanically — see
    /// `docs/prd/zodia-sdk.md`'s note on why `try_send`-and-forget (today's
    /// `SyncPublishMsg` pattern) doesn't carry over: callers here get a
    /// real `Result`.
    async fn call(
        &self,
        build: impl FnOnce(oneshot::Sender<Result<(), ClientError>>) -> Command,
    ) -> Result<(), ClientError> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.cmd_tx.send(build(reply_tx)).await.map_err(|_| ClientError::Disconnected)?;
        reply_rx.await.map_err(|_| ClientError::Disconnected)?
    }

    /// Test-only: hand back the background thread's `JoinHandle` so a test
    /// can drop the client and then `.join()` to prove the thread actually
    /// exits rather than leaking. Not part of the public API.
    #[cfg(test)]
    fn take_thread_handle(&mut self) -> thread::JoinHandle<()> {
        self.thread.take().expect("thread handle already taken")
    }
}

// ── background thread ────────────────────────────────────────────────────────

struct ConnectReady {
    node_id: PeerId,
}

/// Where `run()` gets its network transport from — see [`ZodiaClient::connect`]
/// vs [`ZodiaClient::attach`].
enum NetworkSource {
    /// Spawn a fresh `ZodiaNetwork` inside `run()` and keep it alive for
    /// the connection's lifetime (dropping it tears down discovery/mDNS —
    /// see `ZodiaNetwork`'s field comments in `net/src/network.rs`).
    Owned(ZodiaClientConfig),
    /// Reuse a `ZodiaNetwork` the caller already owns and keeps alive
    /// independently — `run()` never touches `ZodiaNetwork` itself here,
    /// only the `Endpoint`/`Gossip` handles extracted from it, which are
    /// cheap `Send + Clone` references to the same underlying actors.
    Attached {
        endpoint:    Endpoint,
        gossip:      Gossip,
        node_id:     PeerId,
        signing_key: SigningKey,
        data_dir:    PathBuf,
    },
}

enum Command {
    CreateCircle {
        initial_members: Vec<(VerifyingKey, Access<()>)>,
        reply:           oneshot::Sender<Result<SpaceId, ClientError>>,
    },
    OpenCircle { circle_id: SpaceId, reply: Reply },
    InviteToCircle { circle_id: SpaceId, member: VerifyingKey, access: Access<()>, reply: Reply },
    RevokeFromCircle { circle_id: SpaceId, member: VerifyingKey, reply: Reply },
    NotifyCircleInvite { circle_id: SpaceId, recipient: VerifyingKey, reply: Reply },
    NotifyCircleJoin { circle_id: SpaceId, reply: Reply },
    ShareToCircle { circle_id: SpaceId, plaintext: Vec<u8>, reply: Reply },
    CircleMembers {
        circle_id: SpaceId,
        reply:     oneshot::Sender<Result<Vec<(VerifyingKey, Access<()>)>, ClientError>>,
    },
    Author { interp_key: String, body: String, reply: Reply },
    Revoke { target_log_id: Hash, reply: Reply },
    SetDisplayName { name: String, reply: Reply },
    Edit {
        interp_key:      String,
        base_rev:        Hash,
        crdt_update:     Vec<u8>,
        affected_blocks: Vec<[u8; 16]>,
        reply:           Reply,
    },
    Veto { interp_key: String, target_edit_op_id: Hash, reply: Reply },
    AffirmRev { interp_key: String, target_rev: [u8; 32], reply: Reply },
    SetEditorPresence { interp_key: String, joined: bool, reply: Reply },
    Subscribe { interp_key: String, reply: Reply },
    Unsubscribe { interp_key: String, reply: Reply },
    TouchSubscription { interp_key: String, grace: std::time::Duration, reply: Reply },
    Prune { cutoff: Timestamp, reply: oneshot::Sender<Result<u64, ClientError>> },
}

type Reply = oneshot::Sender<Result<(), ClientError>>;

/// Runs for the lifetime of the client, on the dedicated thread's
/// `LocalSet`. Brings up network + sync + pipeline, reports readiness via
/// `ready_tx`, then pumps commands in and `StateEvent`s out until `cmd_rx`
/// closes (i.e. the `ZodiaClient` was dropped).
async fn run(
    source:       NetworkSource,
    mut cmd_rx:   mpsc::Receiver<Command>,
    events_tx:    broadcast::Sender<StateEvent>,
    lifecycle_tx: broadcast::Sender<SyncLifecycleEvent>,
    status_tx:    watch::Sender<SyncStatus>,
    ready_tx:     oneshot::Sender<Result<ConnectReady, ClientError>>,
) {
    // `_owned_net` keeps an owned `ZodiaNetwork` alive for this whole
    // function's scope (dropping it early would tear down discovery) —
    // `None` in `Attached` mode, where the caller keeps their own alive.
    let (endpoint, gossip, node_id, signing_key, data_dir, _owned_net, net_events) = match source {
        NetworkSource::Owned(config) => {
            let net_config = NetworkConfig { signing_key: config.signing_key.clone() };
            let (net, net_events) = match ZodiaNetwork::spawn(net_config, &config.birth).await {
                Ok(pair) => pair,
                Err(e) => { let _ = ready_tx.send(Err(ClientError::Network(e.to_string()))); return; }
            };
            let _ = net.publish_announce().await;
            let endpoint = net.endpoint();
            let gossip = net.gossip();
            let node_id = net.node_id();
            (endpoint, gossip, node_id, config.signing_key, config.data_dir, Some(net), Some(net_events))
        }
        NetworkSource::Attached { endpoint, gossip, node_id, signing_key, data_dir } => {
            (endpoint, gossip, node_id, signing_key, data_dir, None, None)
        }
    };

    let panda_key = PandaSigningKey::from_bytes(signing_key.as_bytes());
    let sync_topic = Topic::from(topic_key_global().0);
    let mut node = match ZodiaSyncNode::spawn(
        panda_key, endpoint, gossip, sync_topic, &data_dir,
    ).await {
        Ok(n) => n,
        Err(e) => { let _ = ready_tx.send(Err(ClientError::Sync(e.to_string()))); return; }
    };

    if ready_tx.send(Ok(ConnectReady { node_id })).is_err() {
        // Caller dropped the `connect()`/`attach()` future (or the whole
        // client) before we finished — nobody left to serve, exit quietly.
        return;
    }

    // Only in `Owned` mode: v1 doesn't act on peer-discovery/consent
    // events, but the sender side (`spawn_gossip_listener` et al) does
    // `.send().await` into this channel — an undrained receiver would
    // eventually stall those tasks. In `Attached` mode the caller already
    // owns and drains their `ZodiaNetwork`'s events themselves.
    if let Some(mut net_events) = net_events {
        tokio::task::spawn_local(async move {
            while net_events.recv().await.is_some() {}
        });
    }

    let pipeline = ZodiaPipeline::new();
    let mut peer_caught_up: HashMap<[u8; 32], bool> = HashMap::new();

    // Grace-period unsubscribe (docs/prd/granular-topic-subscription.md's
    // remaining scope): `TouchSubscription` records when a key was last
    // touched and spawns a timer on this same LocalSet; `expiry_rx` below
    // re-checks elapsed time when a timer fires rather than cancelling
    // stale timers outright — a later touch just makes an earlier timer's
    // check come up short, so no task-cancellation bookkeeping is needed.
    let mut last_touch: HashMap<String, (std::time::Instant, std::time::Duration)> = HashMap::new();
    let (expiry_tx, mut expiry_rx) = mpsc::channel::<String>(64);

    loop {
        tokio::select! {
            cmd = cmd_rx.recv() => {
                match cmd {
                    Some(cmd) => handle_command(&mut node, cmd, &mut last_touch, &expiry_tx).await,
                    None      => break, // ZodiaClient dropped — tear down.
                }
            }
            Some(expired_key) = expiry_rx.recv() => {
                let still_idle = last_touch.get(&expired_key)
                    .is_some_and(|(touched_at, grace)| touched_at.elapsed() >= *grace);
                if still_idle {
                    node.unsubscribe(&expired_key);
                    last_touch.remove(&expired_key);
                }
            }
            Some(sync_event) = node.inbound.recv() => {
                match sync_event {
                    SyncEvent::OperationReceived(op) => {
                        if pipeline.process(*op).await.is_err() {
                            warn!("zodia-sdk: pipeline closed unexpectedly");
                            break;
                        }
                        match pipeline.next().await {
                            Ok(event) => { let _ = events_tx.send(event); }
                            Err(e)    => warn!("zodia-sdk: pipeline next: {e}"),
                        }
                    }
                    SyncEvent::SyncStarted { remote } => {
                        let remote_bytes = *remote.as_bytes();
                        peer_caught_up.insert(remote_bytes, false);
                        let _ = status_tx.send(status_from(&peer_caught_up));
                        let _ = lifecycle_tx.send(SyncLifecycleEvent::Started { remote: remote_bytes });
                    }
                    SyncEvent::SyncFinished { remote, received_ops, .. } => {
                        let remote_bytes = *remote.as_bytes();
                        peer_caught_up.insert(remote_bytes, true);
                        let _ = status_tx.send(status_from(&peer_caught_up));
                        let _ = lifecycle_tx.send(SyncLifecycleEvent::Finished {
                            remote: remote_bytes, received_ops,
                        });
                    }
                    SyncEvent::Failed { remote, error } => {
                        warn!("zodia-sdk: sync failed: {error}");
                        let remote_bytes = *remote.as_bytes();
                        peer_caught_up.insert(remote_bytes, false);
                        let _ = status_tx.send(status_from(&peer_caught_up));
                        let _ = lifecycle_tx.send(SyncLifecycleEvent::Failed {
                            remote: remote_bytes, error,
                        });
                    }
                    SyncEvent::CircleContentReceived { op_id, author, plaintext, .. } => {
                        // `space_id` isn't threaded into `StateEvent` — by
                        // design (docs/prd/circles.md), a decrypted circle
                        // share materialises into the exact same StateEvent
                        // shape a public InterpOp/DocOp would, so the rest
                        // of the app (affirm/thread/revoke handling) needs
                        // no separate code path for circle content.
                        let event = zodia_pipeline::materialize_circle_content(
                            op_id, author, Timestamp::now(), &plaintext,
                        );
                        let _ = events_tx.send(event);
                    }
                }
            }
            else => break,
        }
    }
}

fn status_from(peer_caught_up: &HashMap<[u8; 32], bool>) -> SyncStatus {
    SyncStatus {
        peers_known:     peer_caught_up.len(),
        peers_caught_up: peer_caught_up.values().filter(|&&caught_up| caught_up).count(),
    }
}

async fn handle_command(
    node:       &mut ZodiaSyncNode,
    cmd:        Command,
    last_touch: &mut HashMap<String, (std::time::Instant, std::time::Duration)>,
    expiry_tx:  &mpsc::Sender<String>,
) {
    match cmd {
        Command::CreateCircle { initial_members, reply } => {
            let res = node.create_circle(&initial_members).await.map_err(sync_err);
            let _ = reply.send(res);
        }
        Command::OpenCircle { circle_id, reply } => {
            let res = node.open_circle(circle_id).await.map_err(sync_err);
            let _ = reply.send(res);
        }
        Command::InviteToCircle { circle_id, member, access, reply } => {
            let res = node.invite_to_circle(circle_id, member, access).await.map_err(sync_err);
            let _ = reply.send(res);
        }
        Command::RevokeFromCircle { circle_id, member, reply } => {
            let res = node.revoke_from_circle(circle_id, member).await.map_err(sync_err);
            let _ = reply.send(res);
        }
        Command::NotifyCircleInvite { circle_id, recipient, reply } => {
            let res = node.publish(InterpOp::CircleInviteNotify { circle_id, recipient }).await.map_err(sync_err);
            let _ = reply.send(res);
        }
        Command::NotifyCircleJoin { circle_id, reply } => {
            let res = node.publish(InterpOp::CircleJoinNotify { circle_id }).await.map_err(sync_err);
            let _ = reply.send(res);
        }
        Command::ShareToCircle { circle_id, plaintext, reply } => {
            let res = node.share_to_circle(circle_id, plaintext).await.map_err(sync_err);
            let _ = reply.send(res);
        }
        Command::CircleMembers { circle_id, reply } => {
            let res = node.circle_members(circle_id).await.map_err(sync_err);
            let _ = reply.send(res);
        }
        Command::Author { interp_key, body, reply } => {
            let res = node.publish(InterpOp::Author { interp_key, body }).await.map_err(sync_err);
            let _ = reply.send(res);
        }
        Command::Revoke { target_log_id, reply } => {
            let res = node.publish(InterpOp::Revoke { target_log_id }).await.map_err(sync_err);
            let _ = reply.send(res);
        }
        Command::SetDisplayName { name, reply } => {
            let res = node.publish(InterpOp::SetDisplayName { name }).await.map_err(sync_err);
            let _ = reply.send(res);
        }
        Command::Edit { interp_key, base_rev, crdt_update, affected_blocks, reply } => {
            let op = DocOp::Edit { interp_key, base_rev, crdt_update, affected_blocks };
            let _ = reply.send(node.publish_doc(op).await.map_err(sync_err));
        }
        Command::Veto { interp_key, target_edit_op_id, reply } => {
            let op = DocOp::Veto { interp_key, target_edit_op_id };
            let _ = reply.send(node.publish_doc(op).await.map_err(sync_err));
        }
        Command::AffirmRev { interp_key, target_rev, reply } => {
            let op = DocOp::AffirmRev { interp_key, target_rev };
            let _ = reply.send(node.publish_doc(op).await.map_err(sync_err));
        }
        Command::SetEditorPresence { interp_key, joined, reply } => {
            let op = DocOp::EditorPresence { interp_key, joined };
            let _ = reply.send(node.publish_doc(op).await.map_err(sync_err));
        }
        Command::Subscribe { interp_key, reply } => {
            let _ = reply.send(node.subscribe(&interp_key).await.map_err(sync_err));
        }
        Command::Unsubscribe { interp_key, reply } => {
            node.unsubscribe(&interp_key);
            let _ = reply.send(Ok(()));
        }
        Command::TouchSubscription { interp_key, grace, reply } => {
            let res = node.subscribe(&interp_key).await.map_err(sync_err);
            if res.is_ok() {
                last_touch.insert(interp_key.clone(), (std::time::Instant::now(), grace));
                let expiry_tx = expiry_tx.clone();
                tokio::task::spawn_local(async move {
                    tokio::time::sleep(grace).await;
                    let _ = expiry_tx.send(interp_key).await;
                });
            }
            let _ = reply.send(res);
        }
        Command::Prune { cutoff, reply } => {
            let res = node.prune_older_than(cutoff).await.map_err(sync_err);
            let _ = reply.send(res);
        }
    }
}

fn sync_err(e: SyncError) -> ClientError {
    ClientError::Sync(e.to_string())
}

// ── tests ─────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::Duration;

    use rand_core::OsRng;
    use tokio::time::timeout;

    fn test_config(tmp: &tempdir_shim::TempDir, seed: u8) -> ZodiaClientConfig {
        let _ = seed;
        ZodiaClientConfig {
            signing_key: SigningKey::generate(&mut OsRng),
            birth:       zodia_core::birth_from_coords(2_451_545.0, 59.9, 10.7, 9),
            data_dir:    tmp.path().to_path_buf(),
        }
    }

    #[tokio::test]
    async fn connect_reports_a_stable_node_id() {
        let tmp = tempdir_shim::TempDir::new();
        let client = ZodiaClient::connect(test_config(&tmp, 1)).await
            .expect("connect");
        let id_a = client.node_id();
        let id_b = client.node_id();
        assert_eq!(id_a, id_b);
    }

    #[tokio::test]
    async fn attach_reuses_an_existing_zodia_networks_identity() {
        let signing_key = SigningKey::generate(&mut OsRng);
        let birth = zodia_core::birth_from_coords(2_451_545.0, 59.9, 10.7, 9);
        let net_config = zodia_net::NetworkConfig { signing_key: signing_key.clone() };
        let (net, _net_events) = zodia_net::ZodiaNetwork::spawn(net_config, &birth).await
            .expect("network spawns");
        let tmp = tempdir_shim::TempDir::new();

        let client = ZodiaClient::attach(&net, signing_key, tmp.path().to_path_buf()).await
            .expect("attach");

        assert_eq!(client.node_id(), net.node_id());
    }

    /// Proves `attach()` isn't just constructible but carries real sync
    /// traffic — two independent "app instances", each owning its own
    /// `ZodiaNetwork` and attaching a `ZodiaClient` to it, still converge
    /// on an edit, the same guarantee `subscribe_publish_receive_round_trip`
    /// proves for `connect()`.
    #[tokio::test]
    async fn attach_round_trip_matches_connect_round_trip() {
        let birth = zodia_core::birth_from_coords(2_451_545.0, 59.9, 10.7, 9);

        let key_a = SigningKey::generate(&mut OsRng);
        // `net_a`/`net_b` stay bound for this whole test — dropping a
        // ZodiaNetwork tears down its discovery/mDNS components, which
        // the attached ZodiaClient depends on for the entire exchange.
        let (net_a, _events_a) = zodia_net::ZodiaNetwork::spawn(
            zodia_net::NetworkConfig { signing_key: key_a.clone() }, &birth,
        ).await.expect("network a spawns");
        let tmp_a = tempdir_shim::TempDir::new();
        let a = ZodiaClient::attach(&net_a, key_a, tmp_a.path().to_path_buf()).await
            .expect("attach a");

        let key_b = SigningKey::generate(&mut OsRng);
        let (net_b, _events_b) = zodia_net::ZodiaNetwork::spawn(
            zodia_net::NetworkConfig { signing_key: key_b.clone() }, &birth,
        ).await.expect("network b spawns");
        let tmp_b = tempdir_shim::TempDir::new();
        let b = ZodiaClient::attach(&net_b, key_b, tmp_b.path().to_path_buf()).await
            .expect("attach b");

        let key = "natal:sdk_attach_test_round_trip";
        a.subscribe(key).await.expect("a subscribe");
        b.subscribe(key).await.expect("b subscribe");

        let mut a_events = a.events();
        tokio::time::sleep(Duration::from_millis(500)).await;

        b.edit(key, Hash::from_bytes([0u8; 32]), vec![1, 2, 3], vec![[9u8; 16]]).await
            .expect("b edit");

        let event = timeout(Duration::from_secs(15), async {
            loop {
                match a_events.recv().await.expect("events channel closed") {
                    StateEvent::DocEdited { interp_key, .. } if interp_key == key => return true,
                    _ => continue,
                }
            }
        }).await;

        assert!(event.is_ok(), "a did not observe b's edit within 15s (attach path)");
    }

    #[tokio::test]
    async fn dropping_the_client_tears_down_its_thread() {
        let tmp = tempdir_shim::TempDir::new();
        let mut client = ZodiaClient::connect(test_config(&tmp, 2)).await
            .expect("connect");
        let handle = client.take_thread_handle();
        drop(client);
        // Joining must return promptly — a leaked thread would hang here.
        tokio::task::spawn_blocking(move || handle.join())
            .await
            .expect("join task panicked")
            .expect("zodia-sdk thread panicked");
    }

    #[tokio::test]
    async fn subscribe_publish_receive_round_trip() {
        let tmp_a = tempdir_shim::TempDir::new();
        let tmp_b = tempdir_shim::TempDir::new();
        let a = ZodiaClient::connect(test_config(&tmp_a, 3)).await.expect("connect a");
        let b = ZodiaClient::connect(test_config(&tmp_b, 4)).await.expect("connect b");

        let key = "natal:sdk_test_round_trip";
        a.subscribe(key).await.expect("a subscribe");
        b.subscribe(key).await.expect("b subscribe");

        let mut a_events = a.events();

        // Give discovery a moment before publishing — mirrors the existing
        // net/tests/channel.rs pattern of a short settle window.
        tokio::time::sleep(Duration::from_millis(500)).await;

        b.edit(key, Hash::from_bytes([0u8; 32]), vec![1, 2, 3], vec![[9u8; 16]]).await
            .expect("b edit");

        let event = timeout(Duration::from_secs(15), async {
            loop {
                match a_events.recv().await.expect("events channel closed") {
                    StateEvent::DocEdited { interp_key, .. } if interp_key == key => return true,
                    _ => continue,
                }
            }
        }).await;

        assert!(event.is_ok(), "a did not observe b's edit within 15s");
    }

    /// Minimal stand-in for `tempfile::TempDir` — avoids adding a new
    /// workspace dependency for two dev-only tests. Cleans up on drop.
    mod tempdir_shim {
        use std::path::{Path, PathBuf};

        pub struct TempDir(PathBuf);

        impl TempDir {
            pub fn new() -> Self {
                let mut p = std::env::temp_dir();
                let unique = format!(
                    "zodia-sdk-test-{}-{}",
                    std::process::id(),
                    std::time::SystemTime::now()
                        .duration_since(std::time::UNIX_EPOCH)
                        .unwrap()
                        .as_nanos(),
                );
                p.push(unique);
                std::fs::create_dir_all(&p).expect("create temp dir");
                Self(p)
            }

            pub fn path(&self) -> &Path {
                &self.0
            }
        }

        impl Drop for TempDir {
            fn drop(&mut self) {
                let _ = std::fs::remove_dir_all(&self.0);
            }
        }
    }
}
