car-server-core 0.30.0

//! In-process iMessage approval-transport orchestrator (Units 2 & 4).
//!
//! The ENTIRE transport loop is daemon-side Rust, owned by `car-server-core`,
//! running Rust-to-Rust with **no new FFI/WS surface on the approval seam and
//! no Swift in the transport path** (the only new WS surface in this feature is
//! the host-gated `messaging.config.*` channel in Unit 3). This module is the
//! two halves of that loop:
//!
//! - **Outbound (Unit 4).** [`MessagingOrchestrator::observe_and_notify`]
//!   polls [`HostState::approvals`] for new *fire-and-return* approvals
//!   (`action` NOT prefixed `ws.method:` — the only discriminator the
//!   persisted row carries, the blocking-gate convention from
//!   `handler.rs:2183-2184`). For each new eligible approval, when the
//!   feature is enabled and a paired handle exists, it sends ONE iMessage to
//!   the paired handle (action summary + a short per-approval code) **in
//!   process via the un-gated [`car_ffi_common::integrations::messages_send`]**
//!   so the send does not itself raise a `messages.send` approval (gate loop).
//!   An in-memory code↔approval_id map correlates the later reply.
//!
//! - **Inbound (Unit 4).** [`MessagingOrchestrator::handle_inbound`] takes one
//!   [`InboundMessage`] and FIRST drops it if the sender handle is not
//!   allowlisted (SC-7 — before ANY parse). It then parses the body to exactly
//!   one [`InboundIntent`] — `{Approve, Deny, PairingCode, Ignore}` — and maps
//!   it: a leading code resolves THAT approval; a bare `approve`/`deny`
//!   resolves the sole pending eligible approval; 2+ pending and
//!   bare/ambiguous resolves NOTHING and sends one disambiguation reply listing
//!   the pending codes; an unknown/already-resolved code resolves nothing; a
//!   pairing code routes to [`MessagingConfigStore::validate_and_consume_pairing_code`].
//!   **v1 single-handle invariant:** the config store rejects a second
//!   allowlisted handle (`MessagingConfigStore::add_handle`/`set_allowlist`),
//!   so there is exactly one paired user. "The sole pending approval" is
//!   therefore correct-by-invariant — all eligible pending approvals belong to
//!   the one paired handle; there is no per-sender scoping to do in v1.
//!   Resolution happens **in-process via [`HostState::resolve_approval`]**
//!   (system-level rows resolve directly, no WS session, no per-session ACL).
//!
//! - **Poller (Unit 2).** [`MessagingOrchestrator::poll_once`] reads new
//!   chat.db rows past the persisted watermark (via the Unit 1 reader),
//!   advances the watermark, and forwards each new row to `handle_inbound`.
//!   [`MessagingOrchestrator::run_inbound_loop`] drives `poll_once` on a bounded
//!   interval (`max_iterations` — the no-runaway-loop primitive, mirroring the
//!   `car-scheduler` `dream_loop` precedent), with a cancel token.
//!
//! ## The anti-injection wall (SC-6, structural half)
//!
//! The inbound path NEVER calls any config/allowlist mutator. Its only effects
//! are: (1) resolve a known pending fire-and-return approval id, (2) send a
//! disambiguation reply, or (3) validate-and-consume a pairing code. The
//! `MessagingConfigStore` privileged mutators (the enabled-flag and allowlist
//! setters) are reachable ONLY from the host-gated `messaging.config.*` WS
//! surface — an inbound iMessage carries no such credential. A grep over THIS
//! file for any config-mutator call name finds ZERO matches: the inbound path
//! touches only `resolve_approval` and `validate_and_consume_pairing_code`.

use crate::approval_core::{ApprovalCore, ResolveOutcome};
use crate::channel::{
    CancelSignal, ChannelId, ChannelRegistryHandles, InboundChannel, InboundSink, SharedHost,
};
use crate::host::HostState;
use crate::messaging_config::MessagingConfigStore;
use car_ffi_common::integrations::InboundMessage;
use car_proto::HostApprovalRequest;
#[cfg(test)]
use car_proto::HostApprovalStatus;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::Mutex;

/// The system-raised principal the iMessage adapter supplies to
/// [`ApprovalCore::resolve`]. The literal lives HERE (in the iMessage adapter),
/// passed INTO the channel-agnostic core — it must NOT appear in
/// `approval_core.rs` (MC-3 edge).
const IMESSAGE_PRINCIPAL: &str = "imessage-transport";

/// The resolution string the transport supplies on an approve.
/// Matches the gate's `approve_label` (`handler.rs:2212` → `"approve"`), so
/// `classify_resolution` reads an inbound approve exactly as a CarHost click.
const APPROVE: &str = "approve";
/// The resolution string on a deny. Any non-`approve` string classifies as
/// denied; `"deny"` is the explicit, human-readable choice.
const DENY: &str = "deny";

/// Synchronous, injectable outbound-send seam. The production impl
/// ([`RealMessageSender`]) calls the un-gated
/// [`car_ffi_common::integrations::messages_send`]; tests substitute a
/// capturing spy recording `(handle, body)` so SC-3/SC-5 assert the outbound
/// behavior with no Messages.app.
pub trait MessageSender: Send + Sync {
    /// Send one iMessage `body` to `handle`. Returns an error string on
    /// failure (which the orchestrator logs but never panics on).
    fn send(&self, handle: &str, body: &str) -> Result<(), String>;
}

/// Production send: routes through the un-gated plain Rust
/// [`car_ffi_common::integrations::messages_send`] (`integrations.rs:105`) so
/// the transport's own send does NOT raise a `messages.send` approval and loop
/// the gate. On non-macOS the underlying backend returns an `Err` at runtime
/// (the symbol still links), so this builds on every platform.
#[derive(Debug, Clone, Default)]
pub struct RealMessageSender;

impl MessageSender for RealMessageSender {
    fn send(&self, handle: &str, body: &str) -> Result<(), String> {
        // Build the `SendRequest` JSON the plain-Rust `messages_send` parses.
        let req = serde_json::json!({ "recipient": handle, "body": body });
        let req_json = req.to_string();
        car_ffi_common::integrations::messages_send(&req_json).map(|_| ())
    }
}

/// The parsed meaning of an inbound iMessage body. **Closed set** — the parser
/// yields exactly one of these and nothing else, so there is no inbound→config
/// edge by construction (SC-6).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum InboundIntent {
    /// `approve`, optionally naming a specific per-approval code.
    Approve { code: Option<String> },
    /// `deny`, optionally naming a specific per-approval code.
    Deny { code: Option<String> },
    /// A standalone token that looks like a pairing code (the high-entropy
    /// base64url-no-pad code minted by `messaging.pairing.start`).
    PairingCode(String),
    /// Anything else — silently ignored. NOT a config mutation, ever.
    Ignore,
}

/// Length of a minted pairing code (`mint_pairing_code` → 32 bytes
/// base64url-no-pad = 43 chars). Used by the parser to recognize a standalone
/// token as a pairing-code candidate (vs. a short per-approval code).
const PAIRING_CODE_LEN: usize = 43;

/// Short per-approval code length (e.g. `A7`). Kept tiny so it is easy to type
/// back. The leading letter avoids collision with a bare number.
const APPROVAL_CODE_LEN: usize = 2;

/// In-process iMessage adapter (re-homed from #403's orchestrator). One per
/// daemon; held behind an `Arc` and driven by its `InboundChannel::run()` poll
/// loop. Cheap to clone the `Arc`. The channel-agnostic approval semantics live
/// in [`ApprovalCore`]; this struct owns the iMessage-specific transport
/// (chat.db poll, CodeMap text-code correlation, parse grammar).
pub struct MessagingOrchestrator {
    /// Channel-agnostic approval semantics (eligibility + resolve). The
    /// `"imessage-transport"` principal is supplied by this adapter at each
    /// `core.resolve(...)` call — it is not baked into the core.
    core: ApprovalCore,
    config: MessagingConfigStore,
    sender: Arc<dyn MessageSender>,
    /// Base dir for the inbound-read watermark (the `~/.car/` equivalent).
    /// Injectable so tests never touch the developer's real `~/.car/`.
    base_dir: std::path::PathBuf,
    /// Bidirectional correlation between the short per-approval code we put in
    /// the outbound prompt and the approval id. `code → approval_id` resolves
    /// an inbound `<code> approve`; `approval_id → code` lets us (a) skip
    /// re-sending a prompt for an approval we already notified, and (b) list
    /// the pending codes in a disambiguation reply.
    codes: Mutex<CodeMap>,
    /// SC-7 instrumentation: count of inbound bodies that reached the parser.
    /// A non-allowlisted sender's row must NOT increment this (it is dropped
    /// before parse). Tests assert this stays 0 for a dropped row.
    parse_calls: std::sync::atomic::AtomicU64,
}

/// The code↔approval_id correlation, kept together so both directions stay
/// consistent under one lock.
#[derive(Default)]
struct CodeMap {
    code_to_id: HashMap<String, String>,
    id_to_code: HashMap<String, String>,
    /// Monotonic counter feeding the short code suffix, so codes don't repeat
    /// within a daemon uptime even after the map is cleaned.
    next: u64,
}

impl CodeMap {
    /// Mint the next short per-approval code (e.g. `A7`, `B12`). Letter cycles
    /// A–Z, number increments — readable and easy to type back.
    fn mint(&mut self) -> String {
        let n = self.next;
        self.next += 1;
        let letter = (b'A' + (n % 26) as u8) as char;
        let num = n / 26;
        format!("{letter}{num}")
    }
}

impl MessagingOrchestrator {
    /// Build an orchestrator over the given host, config store, send seam, and
    /// `~/.car/`-equivalent base dir. Production passes
    /// `RealMessageSender`/`MessagingConfigStore::from_home()`/`~/.car`; tests
    /// pass a spy sender and a temp base dir.
    pub fn new(
        host: Arc<HostState>,
        config: MessagingConfigStore,
        sender: Arc<dyn MessageSender>,
        base_dir: impl Into<std::path::PathBuf>,
    ) -> Self {
        Self {
            core: ApprovalCore::new(host),
            config,
            sender,
            base_dir: base_dir.into(),
            codes: Mutex::new(CodeMap::default()),
            parse_calls: std::sync::atomic::AtomicU64::new(0),
        }
    }

    /// The shared host this adapter resolves against (the eviction sweep and
    /// pending-code queries borrow it). Routed through the core so there is one
    /// `Arc<HostState>`.
    fn host(&self) -> &Arc<HostState> {
        self.core.host()
    }

    /// Test/diagnostic accessor: how many inbound bodies reached the parser.
    /// SC-7 asserts this is 0 after a non-allowlisted row is fed.
    pub fn parse_call_count(&self) -> u64 {
        self.parse_calls.load(std::sync::atomic::Ordering::Relaxed)
    }

    // -------------------------------------------------------------------
    // OUTBOUND (Unit 4): observe new approvals → send one prompt each
    // -------------------------------------------------------------------

    /// Observe new eligible approvals on [`HostState`] and send ONE iMessage
    /// prompt per newly-seen one. Idempotent per approval: an approval we have
    /// already minted a code for is skipped, so calling this every tick sends
    /// at most one prompt per approval.
    ///
    /// Gating (the enabled-flag wall): if the feature is disabled OR there is
    /// no paired handle, this is a silent no-op — zero sends. Excluded rows:
    /// any `ws.method:*` blocking-gate row (wrong producer) and any
    /// already-resolved row.
    ///
    /// The single paired recipient is the FIRST allowlisted handle (v1 is one
    /// paired/allowlisted user — ledger item 2). A send failure is logged and
    /// does NOT poison the loop; the code stays mapped so the inbound reply
    /// still correlates if the message did go through.
    pub async fn observe_and_notify(&self) {
        // Enabled-flag gate FIRST — feature off ⇒ no reads of approvals state
        // matter, but cheap-exit anyway so an off feature does zero work.
        if !self.config.is_enabled().unwrap_or(false) {
            return;
        }
        let Some(recipient) = self.paired_handle() else {
            return; // No paired handle ⇒ nobody to notify.
        };

        let approvals = self.host().approvals().await;

        // Eviction sweep: drop code↔id entries whose approval is no longer
        // present-and-Pending (resolved via a CarHost click, reaped, or pruned).
        // Without this the map grows unbounded over a long daemon uptime, since
        // the inbound resolve path only evicts codes IT consumes. Bound it here
        // every tick against the live approval set.
        {
            let live_pending: std::collections::HashSet<&str> = approvals
                .iter()
                .filter(|a| ApprovalCore::is_eligible_pending(a))
                .map(|a| a.id.as_str())
                .collect();
            let mut codes = self.codes.lock().await;
            let stale: Vec<String> = codes
                .id_to_code
                .keys()
                .filter(|id| !live_pending.contains(id.as_str()))
                .cloned()
                .collect();
            for id in stale {
                if let Some(code) = codes.id_to_code.remove(&id) {
                    codes.code_to_id.remove(&code);
                }
            }
        }

        for approval in approvals {
            if !ApprovalCore::is_eligible_pending(&approval) {
                continue;
            }
            // Skip approvals we've already prompted for (idempotency).
            let code = {
                let mut codes = self.codes.lock().await;
                if codes.id_to_code.contains_key(&approval.id) {
                    continue;
                }
                let code = codes.mint();
                codes.code_to_id.insert(code.clone(), approval.id.clone());
                codes.id_to_code.insert(approval.id.clone(), code.clone());
                code
            };
            let body = outbound_body(&approval, &code);
            if let Err(e) = self.sender.send(&recipient, &body) {
                // Roll back the just-minted mapping so this approval is NOT
                // permanently suppressed. The idempotency guard above skips any
                // approval already in `id_to_code`, and the eviction sweep only
                // fires once the approval leaves the pending set — so leaving a
                // mapped-but-unsent code here would drop the prompt forever on a
                // single transient Messages failure. Removing it lets the next
                // tick re-mint (a fresh code via the monotonic counter — fine)
                // and re-send.
                {
                    let mut codes = self.codes.lock().await;
                    if let Some(c) = codes.id_to_code.remove(&approval.id) {
                        codes.code_to_id.remove(&c);
                    }
                }
                tracing::warn!(
                    approval_id = %approval.id,
                    error = %e,
                    "iMessage approval prompt send failed; rolled back code, will retry next tick"
                );
            }
        }
    }

    /// Fan-out outbound (Unit 5): send the iMessage prompt for `approval` using
    /// a SHARED `code` minted ONCE by the fan-out coordinator (not this
    /// adapter's `CodeMap.mint`), so iMessage and Slack carry the SAME code
    /// (MC-8). Records `code ↔ approval_id` into this adapter's `CodeMap` so the
    /// inbound text-resolve path still correlates a reply. Idempotent per
    /// approval: an approval already mapped here is skipped (no duplicate send).
    /// A send failure rolls back the mapping (same reliability contract as
    /// `observe_and_notify`).
    ///
    /// The rendered body is byte-for-byte the same `outbound_body` the iMessage
    /// poll path uses — only the code SOURCE differs (shared vs self-minted), so
    /// the iMessage grammar/behavior is unchanged (MC-1). Gating (enabled flag +
    /// paired handle) is applied here too.
    pub async fn send_shared_prompt(&self, approval: &HostApprovalRequest, code: &str) {
        if !self.config.is_enabled().unwrap_or(false) {
            return;
        }
        let Some(recipient) = self.paired_handle() else {
            return;
        };
        // Idempotency + record the shared code in this adapter's map so an
        // inbound `<code> approve` still resolves. Skip if already mapped.
        {
            let mut codes = self.codes.lock().await;
            if codes.id_to_code.contains_key(&approval.id) {
                return;
            }
            codes
                .code_to_id
                .insert(code.to_string(), approval.id.clone());
            codes
                .id_to_code
                .insert(approval.id.clone(), code.to_string());
        }
        let body = outbound_body(approval, code);
        if let Err(e) = self.sender.send(&recipient, &body) {
            // Roll back so a transient failure does not permanently suppress.
            let mut codes = self.codes.lock().await;
            if let Some(c) = codes.id_to_code.remove(&approval.id) {
                codes.code_to_id.remove(&c);
            }
            tracing::warn!(
                approval_id = %approval.id,
                error = %e,
                "iMessage shared-code prompt send failed; rolled back, will retry next tick"
            );
        }
    }

    // -------------------------------------------------------------------
    // INBOUND (Unit 4): allowlist-drop → parse → resolve / pair / disambiguate
    // -------------------------------------------------------------------

    /// Handle one inbound row. SC-7: a non-allowlisted sender is dropped
    /// BEFORE any parse (the parse counter does not move). Then the body is
    /// parsed to one [`InboundIntent`] and mapped:
    ///
    /// - `<code> approve|deny` → resolve THAT approval id (if it is a known,
    ///   still-pending, eligible approval).
    /// - bare `approve|deny` → resolve the SOLE pending eligible approval; if
    ///   2+ are pending, resolve nothing and send one disambiguation reply.
    ///   (v1 has one paired handle — see the module-level single-handle
    ///   invariant — so "the sole pending approval" needs no per-sender scope.)
    /// - unknown / already-resolved code → resolve nothing.
    /// - pairing code → validate-and-consume (the ONLY inbound-reachable
    ///   config mutation; binds the sender only on a constant-time code match).
    /// - ignore → no-op.
    ///
    /// The enabled-flag gate applies: feature off ⇒ the row is dropped with no
    /// parse/resolve/send.
    pub async fn handle_inbound(&self, msg: &InboundMessage) {
        // Enabled-flag wall: a disabled feature does zero inbound work.
        if !self.config.is_enabled().unwrap_or(false) {
            return;
        }
        // SC-7: drop non-allowlisted senders BEFORE any parse. is_allowlisted
        // errs only on a malformed config file; treat an error as "not
        // allowlisted" (fail closed).
        if !self.config.is_allowlisted(&msg.handle_id).unwrap_or(false) {
            return;
        }

        // Past the allowlist wall: now (and only now) we parse.
        self.parse_calls
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let intent = parse_inbound(&msg.body);

        match intent {
            InboundIntent::Approve { code } => {
                self.resolve_intent(&msg.handle_id, code, APPROVE).await;
            }
            InboundIntent::Deny { code } => {
                self.resolve_intent(&msg.handle_id, code, DENY).await;
            }
            InboundIntent::PairingCode(candidate) => {
                // The ONLY inbound-reachable config mutation. Binds the sender
                // ONLY on a constant-time match of the locally-minted code.
                let _ = self
                    .config
                    .validate_and_consume_pairing_code(&msg.handle_id, &candidate);
            }
            InboundIntent::Ignore => {}
        }
    }

    /// Map an approve/deny intent (with or without a code) onto a resolve.
    /// Encapsulates the ledger rule: a code names exactly one approval; a bare
    /// word resolves the sole pending eligible approval; 2+ pending and
    /// bare/ambiguous resolves nothing and sends a disambiguation reply. (v1's
    /// single-handle invariant means all eligible pending approvals belong to
    /// the one paired user, so the bare-word "sole pending" rule is unambiguous
    /// without per-sender scoping.)
    async fn resolve_intent(&self, handle: &str, code: Option<String>, resolution: &str) {
        match code {
            Some(code) => {
                // Named code: resolve THAT approval iff it maps to a known,
                // still-pending, eligible approval. Unknown / already-resolved
                // ⇒ resolve nothing (no disambiguation — the sender named a
                // specific code, it just isn't actionable).
                if let Some(approval_id) = self.lookup_pending_code(&code).await {
                    self.resolve(&approval_id, resolution).await;
                }
            }
            None => {
                // Bare word: resolve the SOLE pending eligible approval. 0 ⇒
                // nothing to do; 1 ⇒ resolve it; 2+ ⇒ disambiguate.
                let pending = self.pending_codes().await;
                match pending.len() {
                    0 => {}
                    1 => {
                        let approval_id = pending[0].1.clone();
                        self.resolve(&approval_id, resolution).await;
                    }
                    _ => {
                        let reply = disambiguation_body(&pending);
                        if let Err(e) = self.sender.send(handle, &reply) {
                            tracing::warn!(error = %e, "disambiguation reply send failed");
                        }
                    }
                }
            }
        }
    }

    /// Resolve `approval_id` in-process via the channel-agnostic
    /// [`ApprovalCore::resolve`] (which calls the untouched
    /// `HostState::resolve_approval`). The iMessage adapter supplies its own
    /// system-raised principal ([`IMESSAGE_PRINCIPAL`] = `"imessage-transport"`)
    /// — the literal lives in this adapter, not in the core (MC-3 edge).
    ///
    /// On a real resolve ([`ResolveOutcome::Resolved`]) the consumed code is
    /// dropped so a re-send of the same word doesn't re-resolve and it stops
    /// showing in disambiguation lists. On a fan-out non-resolve
    /// ([`ResolveOutcome::StillPending`]) or an error the code is NOT evicted —
    /// the approval did not move, so the user's reply should be able to retry.
    async fn resolve(&self, approval_id: &str, resolution: &str) {
        if self.core.resolve(IMESSAGE_PRINCIPAL, approval_id, resolution).await
            == ResolveOutcome::Resolved
        {
            let mut codes = self.codes.lock().await;
            if let Some(code) = codes.id_to_code.remove(approval_id) {
                codes.code_to_id.remove(&code);
            }
        }
    }

    /// Resolve a named code to its approval id IFF that approval is still
    /// pending and eligible. Unknown code, or a code whose approval has been
    /// resolved/excluded, yields `None` (resolve nothing).
    async fn lookup_pending_code(&self, code: &str) -> Option<String> {
        let approval_id = {
            let codes = self.codes.lock().await;
            codes.code_to_id.get(code).cloned()?
        };
        // Confirm the approval is still a pending, eligible row.
        self.core
            .is_id_eligible_pending(&approval_id)
            .await
            .then_some(approval_id)
    }

    /// The `(code, approval_id)` pairs that are still pending + eligible,
    /// sorted by code for a stable disambiguation listing.
    async fn pending_codes(&self) -> Vec<(String, String)> {
        let pending_ids: std::collections::HashSet<String> = self
            .core
            .eligible_pending()
            .await
            .into_iter()
            .map(|a| a.id)
            .collect();
        let codes = self.codes.lock().await;
        let mut out: Vec<(String, String)> = codes
            .code_to_id
            .iter()
            .filter(|(_, id)| pending_ids.contains(*id))
            .map(|(c, id)| (c.clone(), id.clone()))
            .collect();
        out.sort_by(|a, b| a.0.cmp(&b.0));
        out
    }

    /// The single paired recipient: the first allowlisted handle (v1 is one
    /// paired user). `None` when the allowlist is empty.
    fn paired_handle(&self) -> Option<String> {
        self.config.allowlist().ok()?.into_iter().next()
    }

    // -------------------------------------------------------------------
    // POLLER (Unit 2): bounded interval read past the watermark
    // -------------------------------------------------------------------

    /// One poll tick: read new chat.db rows past the persisted watermark,
    /// advance the watermark, and forward each new row to `handle_inbound`.
    /// Does NO parse/resolve itself — that is `handle_inbound`'s job; the
    /// poller is purely the change-detection tick.
    ///
    /// Watermark semantics: on a fresh/missing watermark, seed it to the
    /// current `MAX(ROWID)` so no pre-existing text is ever replayed as new
    /// (and return without forwarding anything that first tick). Otherwise read
    /// strictly past `last_rowid`, forward, then persist the new high-water =
    /// the max rowid seen this tick. The watermark advances monotonically.
    ///
    /// `read_max`/`read_new` are injected so the test can drive a temp chat.db
    /// (or a synthetic source) with no macOS dependency; production passes the
    /// macOS-gated default-DB readers.
    pub async fn poll_once<FMax, FNew>(
        &self,
        read_max: FMax,
        read_new: FNew,
    ) -> Result<usize, String>
    where
        FMax: Fn() -> Result<i64, String>,
        FNew: Fn(i64) -> Result<Vec<InboundMessage>, String>,
    {
        // Enabled-flag wall: a disabled feature performs NO chat.db read.
        if !self.config.is_enabled().unwrap_or(false) {
            return Ok(0);
        }

        use car_ffi_common::integrations::Watermark;
        let existing = Watermark::load(&self.base_dir).map_err(|e| e.to_string())?;

        let last = match existing {
            Some(w) => w.last_rowid,
            None => {
                // Fresh boot: seed to MAX(ROWID) so old texts never replay.
                let seed = read_max()?;
                Watermark::new(seed)
                    .persist(&self.base_dir)
                    .map_err(|e| e.to_string())?;
                return Ok(0);
            }
        };

        let rows = read_new(last)?;
        if rows.is_empty() {
            return Ok(0);
        }

        // Forward each new row, then advance the watermark to the max rowid
        // seen this tick. Persist AFTER forwarding so a crash mid-tick replays
        // (at-least-once) rather than silently dropping a reply.
        let mut max_seen = last;
        for row in &rows {
            if row.rowid > max_seen {
                max_seen = row.rowid;
            }
            self.handle_inbound(row).await;
        }
        if max_seen > last {
            Watermark::new(max_seen)
                .persist(&self.base_dir)
                .map_err(|e| e.to_string())?;
        }
        Ok(rows.len())
    }

    /// One poll tick against the user's REAL Messages library (macOS only).
    /// Wraps `poll_once` with the production default-DB readers
    /// (`messages_max_rowid` / `messages_read_inbound`). Returns 0 on a fresh
    /// seed or an empty tick.
    #[cfg(target_os = "macos")]
    pub async fn poll_once_default_db(&self) -> Result<usize, String> {
        self.poll_once(
            || car_ffi_common::integrations::messages_max_rowid().map_err(|e| e.to_string()),
            |min| {
                car_ffi_common::integrations::messages_read_inbound(min).map_err(|e| e.to_string())
            },
        )
        .await
    }

    /// Bounded interval poll loop — the no-runaway-loop primitive, and the body
    /// of this adapter's [`InboundChannel::run`]. Each tick runs
    /// `observe_and_notify` (outbound) then one `poll_once` against the real
    /// Messages library (inbound), sleeps `interval`, and stops after
    /// `max_iterations` ticks (when `Some`) or on cancel. Mirrors the
    /// `car-scheduler` `dream_loop` bounded-interval precedent rather than a
    /// hand-rolled `loop {}`.
    ///
    /// A failing tick is logged and the loop continues (one bad read must not
    /// kill the transport). `cancel` is a `tokio::sync::watch` the daemon
    /// shutdown path flips to `true`.
    #[cfg(target_os = "macos")]
    pub async fn run_inbound_loop(
        &self,
        interval: std::time::Duration,
        max_iterations: Option<u32>,
        mut cancel: tokio::sync::watch::Receiver<bool>,
    ) {
        let mut iterations: u32 = 0;
        loop {
            if let Some(max) = max_iterations {
                if iterations >= max {
                    break;
                }
            }

            // Outbound first (prompt newly-raised approvals), then inbound
            // (drain replies). Both are best-effort; a panic in one is the
            // caller's concern (the boot site wraps the spawn).
            self.observe_and_notify().await;
            if let Err(e) = self.poll_once_default_db().await {
                tracing::debug!(error = %e, "messaging poll tick failed");
            }
            iterations += 1;

            if let Some(max) = max_iterations {
                if iterations >= max {
                    break;
                }
            }

            tokio::select! {
                _ = tokio::time::sleep(interval) => {}
                _ = cancel.changed() => {
                    if *cancel.borrow() {
                        break;
                    }
                }
            }
        }
    }

    /// Inbound-ONLY bounded poll loop (Unit 5 boot path). Each tick drains
    /// chat.db replies via `poll_once_default_db` — but does NOT run
    /// `observe_and_notify`, because under multi-channel fan-out the OUTBOUND
    /// prompt is driven once by the [`crate::fanout::FanoutCoordinator`] (so
    /// iMessage and Slack carry the SAME shared code, MC-8). Separating inbound
    /// from outbound here is what lets the boot path notify both channels with
    /// one shared code without double-prompting iMessage. `observe_and_notify`
    /// and `run_inbound_loop` stay intact for the #403 direct-call tests (MC-1).
    #[cfg(target_os = "macos")]
    pub async fn run_inbound_only_loop(
        &self,
        interval: std::time::Duration,
        mut cancel: tokio::sync::watch::Receiver<bool>,
    ) {
        loop {
            if let Err(e) = self.poll_once_default_db().await {
                tracing::debug!(error = %e, "messaging inbound poll tick failed");
            }
            tokio::select! {
                _ = tokio::time::sleep(interval) => {}
                _ = cancel.changed() => {
                    if *cancel.borrow() {
                        break;
                    }
                }
            }
        }
    }
}

// ===================================================================
// The iMessage adapter as the FIRST `InboundChannel` (Unit 3)
// ===================================================================

/// The iMessage adapter implements the channel-agnostic [`InboundChannel`]
/// seam: it names its [`ChannelId::IMessage`] and OWNS its poll loop, feeding
/// observed inbound rows into the supplied sink. macOS-gated — the underlying
/// `poll_once_default_db` reads the local Messages library.
///
/// The watermark/rowid stay PRIVATE to this adapter's poll loop (they never
/// cross the sink — the sink sees only `handle_id` + `body`).
#[cfg(target_os = "macos")]
#[async_trait::async_trait]
impl InboundChannel for MessagingOrchestrator {
    fn channel(&self) -> ChannelId {
        ChannelId::IMessage
    }

    async fn run(&self, _sink: &dyn InboundSink, cancel: CancelSignal) {
        // The iMessage adapter is its own delivery path: each polled row goes to
        // `self.handle_inbound`, which is the channel-agnostic approval sink for
        // this channel. (The trait's `sink` param exists for adapters whose
        // delivery is externally supplied, e.g. Slack in Unit 4.)
        self.run_inbound_loop(std::time::Duration::from_secs(2), None, cancel)
            .await;
    }
}

/// Spawn every ENABLED approval-transport adapter at daemon boot (Unit 3 —
/// cross-platform registry). Iterates [`ChannelId::ALL`]; for each channel that
/// is enabled in `~/.car/messaging.json` it spawns that adapter's
/// [`InboundChannel::run`] loop on the shared cancel signal. The iMessage
/// adapter alone is `#[cfg(target_os = "macos")]`-gated (it reads the local
/// Messages library); the registry/trait/`ChannelId` are unconditional — there
/// are NO cargo feature flags.
///
/// The enabled-flag gate ALSO lives inside each adapter, so a channel that is
/// off does zero work per tick even if spawned; the registry both (a) skips
/// spawning disabled channels at boot, and (b) relies on the adapter's internal
/// gate so a live toggle still flips behavior. Returns the cancel sender so the
/// daemon shutdown path can stop every spawned loop.
pub fn spawn_channel_pollers(host: SharedHost) -> ChannelRegistryHandles {
    let (cancel_tx, _cancel_rx) = tokio::sync::watch::channel(false);
    let store = MessagingConfigStore::from_home();

    // The fan-out coordinator (Unit 5) drives OUTBOUND once across all enabled
    // channels with ONE shared code (MC-8). Each enabled channel contributes its
    // concrete adapter `Arc` to the coordinator (for outbound) AND spawns its
    // INBOUND loop (for replies/clicks). iMessage's inbound is poll-only at boot
    // (`run_inbound_only_loop`) so outbound is not double-driven.
    let core = crate::approval_core::ApprovalCore::new(host.clone());

    // Build the iMessage adapter Arc (macOS only). Hold it both for the
    // coordinator (outbound) and to spawn its inbound poll loop.
    #[cfg(target_os = "macos")]
    let imessage: Option<Arc<MessagingOrchestrator>> = {
        if store.is_enabled_for(ChannelId::IMessage).unwrap_or(false) {
            let base_dir = std::env::var_os("HOME")
                .map(std::path::PathBuf::from)
                .unwrap_or_else(|| std::path::PathBuf::from("."))
                .join(".car");
            let orch = Arc::new(MessagingOrchestrator::new(
                host.clone(),
                MessagingConfigStore::from_home(),
                Arc::new(RealMessageSender),
                base_dir,
            ));
            // Spawn the inbound-only poll loop.
            let orch_in = orch.clone();
            let cancel_rx = cancel_tx.subscribe();
            tokio::spawn(async move {
                orch_in
                    .run_inbound_only_loop(std::time::Duration::from_secs(2), cancel_rx)
                    .await;
            });
            Some(orch)
        } else {
            None
        }
    };
    #[cfg(not(target_os = "macos"))]
    let imessage: Option<Arc<MessagingOrchestrator>> = None;

    // Build the Slack adapter Arc (UNCONDITIONAL — no cfg gate, MC-11). Its
    // tokens live in the OS keychain (MC-9); the on-disk config holds NO bearer
    // value, only a keychain REFERENCE (the key names) persisted by the
    // host-gated `messaging.config.set` provisioning path. The transport is
    // constructed FROM that persisted ref, so the live adapter fetches the real
    // creds the host provisioned. The post-channel id is read from CONFIG (the
    // `slack_channel_id` in `messaging.json`, set on the same host-gated call as
    // the tokens) — it is configuration, not a secret, so it lives in the config
    // file, not the keychain. If no token has been provisioned (no
    // `slack_token_ref` in the config), the adapter cannot run — log + SKIP
    // rather than hard-crashing the boot (mirrors how iMessage skips a missing
    // prerequisite / chat.db). A misconfigured/absent post-channel id ⇒ the
    // adapter is built but cannot post (no-op outbound), still safe.
    let slack: Option<Arc<crate::slack_adapter::SlackAdapter>> = {
        if store.is_enabled_for(ChannelId::Slack).unwrap_or(false) {
            match store.slack_token_ref_for(ChannelId::Slack).unwrap_or(None) {
                Some(token_ref) => {
                    let post_channel = store
                        .slack_channel_id_for(ChannelId::Slack)
                        .unwrap_or(None)
                        .unwrap_or_default();
                    if post_channel.is_empty() {
                        // No post-channel configured: outbound `post_prompt` is a
                        // no-op (it early-returns on an empty channel), so we'd
                        // post nothing — but inbound (pairing + button resolve)
                        // still works, so build the adapter anyway. Warn ONCE
                        // here at boot rather than every doomed outbound tick.
                        tracing::warn!(
                            "slack channel enabled but no slack_channel id configured \
                             (no slack_channel in messaging.json) — outbound approval \
                             prompts will be skipped; inbound pairing + button resolve \
                             still active. Set via messaging.config.set \
                             {{ channel: \"slack\", slack_channel: \"<id>\" }}"
                        );
                    }
                    let transport = Arc::new(crate::slack_adapter::RealSlackTransport::new(
                        token_ref.bot_token_key.clone(),
                        token_ref.app_token_key.clone(),
                    ));
                    transport.clone().spawn_socket_loop(cancel_tx.subscribe());
                    let adapter = Arc::new(crate::slack_adapter::SlackAdapter::new(
                        host.clone(),
                        MessagingConfigStore::from_home(),
                        transport as Arc<dyn crate::slack_adapter::SlackTransport>,
                        post_channel,
                    ));
                    // Spawn the Slack inbound `run()` loop (push-based; pulls
                    // events off the transport seam and dispatches each).
                    let adapter_in = adapter.clone();
                    let cancel_rx = cancel_tx.subscribe();
                    tokio::spawn(async move {
                        let sink = NoopSink;
                        adapter_in.run(&sink, cancel_rx).await;
                    });
                    Some(adapter)
                }
                None => {
                    tracing::warn!(
                        "slack channel enabled but no tokens provisioned \
                         (no slack_token_ref in messaging.json) — skipping slack adapter; \
                         provision via messaging.config.set {{ channel: \"slack\", bot_token, app_token }}"
                    );
                    None
                }
            }
        } else {
            None
        }
    };

    // Spawn the single fan-out OUTBOUND ticker over the enabled channels. One
    // shared code per approval reaches every enabled channel (MC-8). If no
    // channel is enabled this still runs but is a cheap no-op (eligible-pending
    // query over an empty/gated set).
    if imessage.is_some() || slack.is_some() {
        let coordinator = crate::fanout::FanoutCoordinator::new(core, imessage, slack);
        let mut cancel_rx = cancel_tx.subscribe();
        tokio::spawn(async move {
            loop {
                coordinator.observe_and_fanout().await;
                tokio::select! {
                    _ = tokio::time::sleep(std::time::Duration::from_secs(2)) => {}
                    _ = cancel_rx.changed() => {
                        if *cancel_rx.borrow() {
                            break;
                        }
                    }
                }
            }
        });
    }

    ChannelRegistryHandles { cancel_tx }
}

/// A no-op [`InboundSink`] for adapters that own their own delivery (the
/// iMessage adapter delivers via its internal `handle_inbound`; the Slack
/// adapter resolves by `approval_id` + pairs via the code primitive — neither
/// routes through the `handle_id`+`body` sink, so its `run` ignores the passed
/// sink). Present so the registry can drive any adapter through the seam
/// signature uniformly.
struct NoopSink;

#[async_trait::async_trait]
impl InboundSink for NoopSink {
    async fn deliver(&self, _channel: ChannelId, _msg: &InboundMessage) {}
}

/// Build the outbound prompt body: a short action summary plus the
/// per-approval code and how to reply. Kept compact for a text message.
///
/// `pub` so the MC-1 fan-out gate can assert the fan-out iMessage body
/// (rendered by [`MessagingOrchestrator::send_shared_prompt`]) is byte-for-byte
/// the poll-path grammar — pinning the MC-1 guarantee against a future drift in
/// `send_shared_prompt`.
pub fn outbound_body(approval: &HostApprovalRequest, code: &str) -> String {
    format!(
        "Approval needed: {action}\nReply `{code} approve` or `{code} deny`.",
        action = approval.action,
        code = code,
    )
}

/// Build the disambiguation reply listing the pending codes (when a bare
/// approve/deny is ambiguous because 2+ approvals are pending).
fn disambiguation_body(pending: &[(String, String)]) -> String {
    let mut s = String::from(
        "Multiple approvals are pending. Reply with a code, e.g. `<code> approve`:\n",
    );
    for (code, _id) in pending {
        s.push_str(&format!("• {code}\n"));
    }
    s
}

/// Parse an inbound body to exactly one [`InboundIntent`]. **Closed output** —
/// the only things this can ever produce are approve/deny (optionally
/// code-prefixed), a pairing-code candidate, or ignore. There is NO branch
/// that returns a config mutation. Recognizes (case-insensitive):
///
/// - `approve` / `deny` (bare)
/// - `<code> approve` / `<code> deny` (leading short code)
/// - `approve <code>` / `deny <code>` (trailing short code — lenient)
/// - a standalone token of pairing-code length ⇒ a pairing candidate
///
/// Everything else ⇒ `Ignore`.
fn parse_inbound(body: &str) -> InboundIntent {
    let trimmed = body.trim();
    if trimmed.is_empty() {
        return InboundIntent::Ignore;
    }
    let tokens: Vec<&str> = trimmed.split_whitespace().collect();

    // Single token: a bare word, or a standalone pairing-code candidate.
    if tokens.len() == 1 {
        let t = tokens[0];
        if t.eq_ignore_ascii_case(APPROVE) {
            return InboundIntent::Approve { code: None };
        }
        if t.eq_ignore_ascii_case(DENY) {
            return InboundIntent::Deny { code: None };
        }
        if looks_like_pairing_code(t) {
            return InboundIntent::PairingCode(t.to_string());
        }
        return InboundIntent::Ignore;
    }

    // Two-ish tokens: a short code paired with approve/deny, in either order.
    if tokens.len() == 2 {
        let (a, b) = (tokens[0], tokens[1]);
        // `<code> approve|deny`
        if let Some(intent) = code_word(a, b) {
            return intent;
        }
        // `approve|deny <code>` (lenient trailing-code form)
        if let Some(intent) = code_word(b, a) {
            return intent;
        }
    }

    InboundIntent::Ignore
}

/// If `word` is approve/deny and `code` looks like a short per-approval code,
/// build the corresponding code-named intent. Returns `None` otherwise.
fn code_word(code: &str, word: &str) -> Option<InboundIntent> {
    if !looks_like_approval_code(code) {
        return None;
    }
    if word.eq_ignore_ascii_case(APPROVE) {
        return Some(InboundIntent::Approve {
            code: Some(code.to_uppercase()),
        });
    }
    if word.eq_ignore_ascii_case(DENY) {
        return Some(InboundIntent::Deny {
            code: Some(code.to_uppercase()),
        });
    }
    None
}

/// A short per-approval code looks like `A7`/`B12` — a leading ASCII letter
/// then digits, short. (Distinguishes it from the long pairing code.)
fn looks_like_approval_code(t: &str) -> bool {
    let bytes = t.as_bytes();
    if bytes.len() < APPROVAL_CODE_LEN || bytes.len() > 5 {
        return false;
    }
    bytes[0].is_ascii_alphabetic() && bytes[1..].iter().all(|b| b.is_ascii_digit())
}

/// A standalone token of exactly pairing-code length and base64url charset is a
/// pairing-code candidate. The constant-time compare in
/// `validate_and_consume_pairing_code` is the real gate; this is just the
/// parser's recognizer so a 43-char token routes to pairing rather than ignore.
fn looks_like_pairing_code(t: &str) -> bool {
    t.len() == PAIRING_CODE_LEN
        && t.bytes()
            .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')
}

#[cfg(test)]
mod tests {
    //! SC-C — Parser-strictness backstop (Wall 2 pinned).
    //!
    //! On a single-Apple-ID (solo) account the daemon's OWN outbound bodies echo
    //! back into chat.db as `is_from_me = 0` received rows. Wall 1 (the
    //! `is_from_me = 0` SQL filter, `read.rs`) drops those echoes before a row is
    //! constructed. This unit pins the SECOND wall: even IF a daemon-authored body
    //! reached `parse_inbound` (defense-in-depth, or a future refactor that lets
    //! one through), the closed-grammar parser must classify it `Ignore` — never a
    //! command. Both `outbound_body` and `disambiguation_body` are multi-token
    //! strings with leading non-command words, so they fall through to the
    //! terminal `Ignore`. This test fails the moment a future rephrase of either
    //! body collapses it to a 1–2-token command-shaped form. Pure in-process —
    //! no host, no async.

    use super::*;

    /// Build a representative system-level `HostApprovalRequest` for feeding into
    /// `outbound_body`. Only `.action` is load-bearing for the body; the rest of
    /// the field set mirrors what the host produces for a fire-and-return row.
    fn sample_approval() -> HostApprovalRequest {
        HostApprovalRequest {
            id: "id0".to_string(),
            agent_id: None,
            client_id: None,
            action: "send wire transfer".to_string(),
            details: serde_json::Value::Null,
            options: vec![],
            status: HostApprovalStatus::Pending,
            created_at: chrono::Utc::now(),
            resolved_at: None,
            resolution: None,
        }
    }

    #[test]
    fn parse_inbound_ignores_daemon_authored_bodies() {
        // The outbound approval prompt the daemon sends (and which echoes back
        // `is_from_me = 0` on a solo account) must parse to Ignore.
        let prompt = outbound_body(&sample_approval(), "A0");
        assert_eq!(
            parse_inbound(&prompt),
            InboundIntent::Ignore,
            "the daemon's own outbound prompt body must parse to Ignore, got {prompt:?}"
        );

        // The disambiguation reply the daemon sends (also echoes back
        // `is_from_me = 0` on a solo account) must parse to Ignore.
        let disambig = disambiguation_body(&[
            ("A0".to_string(), "id0".to_string()),
            ("B0".to_string(), "id1".to_string()),
        ]);
        assert_eq!(
            parse_inbound(&disambig),
            InboundIntent::Ignore,
            "the daemon's own disambiguation body must parse to Ignore, got {disambig:?}"
        );
    }
}