car_messaging/

messaging_orchestrator.rs

//! 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::channel_supervisor::{
    liveness_record_failure, liveness_record_success, ChannelSupervisor, SharedLiveness,
};
use crate::messaging_config::MessagingConfigStore;
use car_ffi_common::integrations::InboundMessage;
use car_proto::approval_summary::{approval_summary, sanitize_line};
use car_proto::HostApprovalRequest;
#[cfg(test)]
use car_proto::HostApprovalStatus;
use car_server_types::approval_core::{ApprovalCore, ResolveOutcome};
use car_server_types::channel::{ChannelId, InboundChannel, InboundSink, SharedHost};
// `CancelSignal` is only referenced by the macOS iMessage adapter below.
#[cfg(target_os = "macos")]
use car_server_types::channel::CancelSignal;
use car_server_types::host::HostState;
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";

/// The fixed, unmistakable body the `messaging.test_send` self-test sends (U4).
/// It is plainly NOT an approval prompt (no code, no "reply approve/deny"), so
/// it can never be mistaken for one — and on a solo Apple ID its echo back
/// parses to `Ignore` (the closed-grammar parser drops any non-command body).
pub const TEST_SEND_BODY: &str =
    "CAR test: your iMessage approvals are connected. No action needed.";

/// The structured result of one send attempt (U3). Distinguishes the THREE
/// outcomes the orchestrator must record, instead of collapsing two of them to
/// "success" as the pre-U3 `.map(|_| ())` did:
///
/// - `Err(reason)` from [`MessageSender::send`] = a HARD failure (osascript
///   killed by a TCC/Automation denial, or a non-macOS no-op). The caller rolls
///   back the code mapping + retries (existing reliability contract).
/// - `Ok(SendOutcome { sent: false, reason })` = a SOFT failure — the JXA send
///   returned `sent:false` (recipient-not-found / no Messages service). Pre-U3
///   this was silently treated as success (`.map(|_| ())` discarded the bool);
///   it is now recorded as a FAILURE with the reason (U3 regression guard).
/// - `Ok(SendOutcome { sent: true, .. })` = a genuine success.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SendOutcome {
    /// Whether the message was actually delivered (`sent` from the JXA send).
    pub sent: bool,
    /// Failure reason when `sent == false`; `None` on success.
    pub reason: Option<String>,
}

impl SendOutcome {
    /// A delivered send.
    pub fn ok() -> Self {
        Self {
            sent: true,
            reason: None,
        }
    }

    /// A soft failure (the send returned `sent:false`) with a reason.
    pub fn soft_fail(reason: impl Into<String>) -> Self {
        Self {
            sent: false,
            reason: Some(reason.into()),
        }
    }
}

/// 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:
    /// - `Err(reason)` on a HARD failure (the orchestrator logs + rolls back +
    ///   retries; never panics).
    /// - `Ok(SendOutcome { sent, reason })` otherwise — `sent:true` for a
    ///   delivered message, `sent:false` (with a reason) for a SOFT failure the
    ///   pre-U3 code swallowed as success.
    fn send(&self, handle: &str, body: &str) -> Result<SendOutcome, 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<SendOutcome, 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();
        // U3: inspect the returned `SendResult` JSON instead of discarding it
        // with `.map(|_| ())`. The JXA backend returns
        // `{ available, reason, sent }`: a `sent:false` (recipient-not-found, no
        // Messages service) is a SOFT failure that the pre-U3 code counted as
        // success. Surface it as `Ok(SendOutcome::soft_fail(reason))` so the
        // caller records it as a failure; a hard osascript error is already an
        // `Err` from `messages_send`.
        let value = car_ffi_common::integrations::messages_send(&req_json)?;
        let sent = value.get("sent").and_then(|v| v.as_bool()).unwrap_or(false);
        if sent {
            Ok(SendOutcome::ok())
        } else {
            let reason = value
                .get("reason")
                .and_then(|v| v.as_str())
                .map(|s| s.to_string())
                .unwrap_or_else(|| "Messages reported the message was not sent".to_string());
            Ok(SendOutcome::soft_fail(reason))
        }
    }
}

/// 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,
    /// Optional shared per-channel liveness sink (U2/U3). When `Some`, every
    /// outbound send records its structured outcome (success / soft / hard
    /// failure) here so `messaging.status` can report "last delivered" + the
    /// last error. `None` for adapters built without a liveness sink (the #403
    /// direct tests that don't assert liveness). The channel this adapter
    /// serves is fixed at [`ChannelId::IMessage`] for outcome keying.
    liveness: Option<SharedLiveness>,
}

/// 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),
            liveness: None,
        }
    }

    /// Build an orchestrator that ALSO records every outbound send's outcome
    /// into the shared per-channel liveness map (U2/U3). The boot path and the
    /// runtime-enable spawn closure use this so `messaging.status` sees real
    /// "last delivered" + last-error state; the `messaging.test_send` self-test
    /// (U4) relies on the same recording so a test failure surfaces identically
    /// to a real one.
    pub fn with_liveness(
        host: Arc<HostState>,
        config: MessagingConfigStore,
        sender: Arc<dyn MessageSender>,
        base_dir: impl Into<std::path::PathBuf>,
        liveness: SharedLiveness,
    ) -> 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),
            liveness: Some(liveness),
        }
    }

    /// Record a send `outcome` into the shared liveness (if a sink is wired),
    /// keyed to this adapter's channel (iMessage). A hard `Err` is recorded by
    /// [`Self::record_hard_failure`]; this handles the `Ok(SendOutcome)` shape:
    /// `sent:true` → success; `sent:false` → soft failure with its reason (the
    /// case the pre-U3 code swallowed).
    fn record_outcome(&self, outcome: &SendOutcome) {
        let Some(liveness) = &self.liveness else {
            return;
        };
        if outcome.sent {
            liveness_record_success(liveness, ChannelId::IMessage);
        } else {
            let reason = outcome
                .reason
                .clone()
                .unwrap_or_else(|| "send failed".to_string());
            liveness_record_failure(liveness, ChannelId::IMessage, reason);
        }
    }

    /// Record a HARD send failure (an `Err` from the sender — osascript killed
    /// by an Automation/TCC denial) into the shared liveness.
    fn record_hard_failure(&self, reason: &str) {
        if let Some(liveness) = &self.liveness {
            liveness_record_failure(liveness, ChannelId::IMessage, reason);
        }
    }

    /// 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);
            match self.sender.send(&recipient, &body) {
                Ok(outcome) => {
                    // U3: record the structured outcome. A soft `sent:false`
                    // (recipient-not-found / no Messages service) is recorded as
                    // a FAILURE here — the case pre-U3 silently counted as
                    // success. The code mapping is KEPT on a soft failure (the
                    // approval is still pending; a later tick re-attempts via the
                    // idempotency guard) — only a HARD error rolls back to force
                    // a re-mint.
                    self.record_outcome(&outcome);
                }
                Err(e) => {
                    self.record_hard_failure(&e);
                    // 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);
        match self.sender.send(&recipient, &body) {
            Ok(outcome) => {
                // U3: record success / soft-failure into liveness.
                self.record_outcome(&outcome);
            }
            Err(e) => {
                self.record_hard_failure(&e);
                // 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);
                        // A disambiguation reply is an internal nudge, not an
                        // approval-delivery signal, so it is NOT recorded into
                        // liveness (which tracks approval-prompt delivery). A
                        // hard error is logged; a soft `sent:false` is ignored
                        // (the user will simply re-send their reply).
                        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, action)` triples that are still pending +
    /// eligible, sorted by code for a stable disambiguation listing. The action
    /// rides along so the listing can name what each code refers to.
    async fn pending_codes(&self) -> Vec<(String, String, String)> {
        let pending_actions: std::collections::HashMap<String, String> = self
            .core
            .eligible_pending()
            .await
            .into_iter()
            .map(|a| (a.id, a.action))
            .collect();
        let codes = self.codes.lock().await;
        let mut out: Vec<(String, String, String)> = codes
            .code_to_id
            .iter()
            .filter_map(|(c, id)| {
                pending_actions
                    .get(id)
                    .map(|action| (c.clone(), id.clone(), action.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()
    }

    // -------------------------------------------------------------------
    // SELF-TEST (U4): on-demand "does my Mac actually text my phone" probe
    // -------------------------------------------------------------------

    /// Send the fixed, clearly-labeled self-test message to the paired handle
    /// and return the outcome (U4). This is a PURE send probe: it composes
    /// [`TEST_SEND_BODY`], sends through the SAME [`MessageSender`] the real
    /// approval path uses, and records the outcome into liveness (so a passing
    /// test genuinely proves Automation works and a failing one surfaces the
    /// same way a real send failure does). It mints **no** per-approval/pairing
    /// code mapping and resolves nothing — the `CodeMap` is untouched.
    ///
    /// Returns `Ok(())` when the message was delivered (`sent:true`), or
    /// `Err(reason)` for: feature disabled, no paired handle, a hard send error
    /// (Automation denied), or a soft `sent:false` (recipient-not-found). The
    /// error strings are user-actionable (surfaced verbatim in the pane).
    pub async fn send_test(&self) -> Result<(), String> {
        if !self.config.is_enabled().unwrap_or(false) {
            return Err("iMessage is off — turn it on first.".to_string());
        }
        let Some(recipient) = self.paired_handle() else {
            return Err("No paired handle — text the pairing code first.".to_string());
        };
        match self.sender.send(&recipient, TEST_SEND_BODY) {
            Ok(outcome) => {
                // Record the outcome (success or soft failure) so "last
                // delivered" + last-error reflect the probe, identical to a
                // real send.
                self.record_outcome(&outcome);
                if outcome.sent {
                    Ok(())
                } else {
                    Err(outcome
                        .reason
                        .unwrap_or_else(|| "the message was not sent".to_string()))
                }
            }
            Err(e) => {
                self.record_hard_failure(&e);
                Err(e)
            }
        }
    }

    // -------------------------------------------------------------------
    // 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.
///
/// **Unit 1 (runtime spawn-on-enable):** this also builds and returns a
/// [`ChannelSupervisor`] — recorded with the channels spawned here at boot —
/// that the host-gated `messaging.config.set` handler uses to spawn a channel's
/// watcher the instant the user enables it, with NO daemon/app restart. The
/// supervisor holds the SAME cancel signal and the SAME per-channel liveness map
/// the boot adapters write into, so a runtime-spawned channel shuts down cleanly
/// and reports status identically.
pub fn spawn_channel_pollers(host: SharedHost) -> Arc<ChannelSupervisor> {
    let (cancel_tx, _cancel_rx) = tokio::sync::watch::channel(false);
    let store = MessagingConfigStore::from_home();

    // The shared per-channel liveness map (U2/U3). The boot adapters and any
    // runtime-spawned adapter write outbound-send outcomes here; `messaging.status`
    // reads it. Build it ONCE and hand the same `Arc` to every adapter + the
    // supervisor. Only the macOS iMessage adapter consumes it today, so it is
    // macOS-gated to avoid an unused binding on other platforms.
    #[cfg(target_os = "macos")]
    let liveness: SharedLiveness =
        Arc::new(std::sync::Mutex::new(std::collections::HashMap::new()));

    // 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 = car_server_types::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) {
            // Same state root the config store above resolved — `$CAR_HOME`
            // when set, else `~/.car` — so the orchestrator's cursor state and
            // its config never land under two different roots.
            let base_dir = car_home::root_or_relative();
            let orch = Arc::new(MessagingOrchestrator::with_liveness(
                host.clone(),
                MessagingConfigStore::from_home(),
                Arc::new(RealMessageSender),
                base_dir,
                liveness.clone(),
            ));
            // 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
        }
    };

    // Track which channels we spawned at boot so the supervisor's live set is
    // accurate from the start (U1 idempotency: a runtime enable of a
    // boot-spawned channel is a no-op).
    let imessage_spawned = imessage.is_some();
    let slack_spawned = slack.is_some();

    // 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;
                        }
                    }
                }
            }
        });
    }

    // U1: build the runtime supervisor over the SAME cancel signal + liveness
    // map, with the per-channel spawn closure, and record the boot-spawned
    // channels as live. The host-gated `messaging.config.set` handler reaches it
    // via `ServerState` to call `ensure_spawned(channel)` on an off→on
    // transition — no restart.
    let supervisor = Arc::new(ChannelSupervisor::new(host, cancel_tx, make_spawn_fn()));
    if imessage_spawned {
        supervisor.mark_spawned(ChannelId::IMessage);
    }
    if slack_spawned {
        supervisor.mark_spawned(ChannelId::Slack);
    }
    supervisor
}

/// Build the per-channel spawn closure the [`ChannelSupervisor`] calls to start
/// ONE channel's watcher at runtime (U1). It is handed the channel, the shared
/// host, the shared liveness map, and a cancel receiver subscribed to the
/// supervisor's signal.
///
/// For a runtime-enabled iMessage channel that was OFF at boot, this spawns the
/// SELF-CONTAINED `MessagingOrchestrator::run_inbound_loop` — which runs BOTH
/// `observe_and_notify` (outbound prompts) and the inbound poll per tick. This
/// is the #403 direct-call path: for a single channel it is byte-for-byte the
/// same `outbound_body` grammar as the boot path. (The boot path splits inbound
/// from outbound only so the multi-channel fan-out coordinator can drive
/// outbound ONCE with a shared code across SIMULTANEOUSLY-enabled channels —
/// which does not apply when the user flips ONE channel on at runtime.)
///
/// Slack runtime-enable always returns an `Err` (see [`spawn_slack_runtime`]):
/// its outbound prompt driver is the boot-time `FanoutCoordinator`, which a
/// runtime closure cannot rebuild, so the channel must NOT be marked live (which
/// would make `messaging.status` falsely report `watcher_running:true`). The
/// host UI surfaces the error; a restart activates Slack outbound.
fn make_spawn_fn() -> crate::channel_supervisor::SpawnFn {
    Box::new(
        |channel: ChannelId,
         host: &SharedHost,
         liveness: &SharedLiveness,
         cancel_rx: tokio::sync::watch::Receiver<bool>|
         -> Result<(), String> {
            match channel {
                ChannelId::IMessage => spawn_imessage_runtime(host, liveness, cancel_rx),
                ChannelId::Slack => spawn_slack_runtime(host, liveness, cancel_rx),
            }
        },
    )
}

/// Spawn a runtime-enabled iMessage watcher (macOS). Self-contained
/// outbound+inbound loop (`run_inbound_loop`). On non-macOS this is a no-op that
/// returns `Ok(())` (the channel can be "enabled" in config but reads nothing —
/// matching the boot path, which builds no iMessage adapter off macOS).
#[allow(unused_variables)]
fn spawn_imessage_runtime(
    host: &SharedHost,
    liveness: &SharedLiveness,
    cancel_rx: tokio::sync::watch::Receiver<bool>,
) -> Result<(), String> {
    #[cfg(target_os = "macos")]
    {
        // The CAR state root (`$CAR_HOME`, else `~/.car`), matching the boot
        // path above so a runtime-enabled watcher reads the same cursor state.
        let base_dir = car_home::root_or_relative();
        let orch = Arc::new(MessagingOrchestrator::with_liveness(
            host.clone(),
            MessagingConfigStore::from_home(),
            Arc::new(RealMessageSender),
            base_dir,
            liveness.clone(),
        ));
        tokio::spawn(async move {
            orch.run_inbound_loop(std::time::Duration::from_secs(2), None, cancel_rx)
                .await;
        });
        Ok(())
    }
    #[cfg(not(target_os = "macos"))]
    {
        // Off macOS the iMessage adapter reads nothing; recording the channel as
        // "spawned" is harmless (the enabled-flag gate already makes every tick
        // a no-op) and keeps the live-set bookkeeping uniform.
        Ok(())
    }
}

/// Runtime-enable for Slack is intentionally NOT supported in this build.
///
/// Slack's OUTBOUND approval prompts (`SlackAdapter::post_prompt`) have exactly
/// ONE driver — the [`FanoutCoordinator`], built ONCE at boot in
/// `spawn_channel_pollers` over whatever channels were enabled at boot. A
/// runtime-enable closure can only spawn Slack's INBOUND loop (`adapter.run`),
/// which never calls `post_prompt`. So a Slack channel enabled at runtime would
/// receive inbound replies but could NEVER send an outbound approval prompt —
/// yet marking it live would make `messaging.status` report
/// `watcher_running:true` and the host UI show a green "Ready" it cannot honor.
///
/// To keep `watcher_running` HONEST (the load-bearing invariant: never report a
/// channel live when its outbound can't fire), this returns an `Err`. The
/// supervisor's `ensure_spawned` rolls back the reserved live-set entry on this
/// `Err`, so the channel is NOT marked live. The config flag is still persisted
/// by `messaging.config.set` (config is config) — the channel is simply not
/// watcher_running until the daemon restarts, which rebuilds the boot-time
/// `FanoutCoordinator` over the now-enabled Slack channel and activates outbound.
///
/// Building a runtime Slack outbound path is deliberately out of v1 scope.
/// iMessage's runtime-enable is unaffected: `spawn_imessage_runtime` uses the
/// self-contained `run_inbound_loop`, which drives BOTH outbound and inbound.
#[allow(unused_variables)]
fn spawn_slack_runtime(
    host: &SharedHost,
    liveness: &SharedLiveness,
    cancel_rx: tokio::sync::watch::Receiver<bool>,
) -> Result<(), String> {
    Err(
        "Slack runtime-enable is not supported in this build; restart to activate \
         outbound delivery."
            .to_string(),
    )
}

/// 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 {
    // `action` is agent-authored, unvalidated and unbounded. Sanitised for the
    // same reason the summary's fields are: raw, it can carry newlines that
    // forge lines looking like the labelled block below it, and length that
    // pushes that block off a phone screen.
    let mut body = format!("Approval needed: {}\n", sanitize_line(&approval.action));
    // What the action would actually do, when `details` says. Without it this
    // channel asked for a decision while withholding everything the decision
    // rests on — the dashboard has rendered the same payload all along
    // (`car_proto::approval_summary`).
    if let Some(summary) = approval_summary(approval) {
        body.push_str(&summary);
        body.push('\n');
    }
    body.push_str(&format!("Reply `{code} approve` or `{code} deny`."));
    body
}

/// 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)]) -> String {
    let mut s =
        String::from("Multiple approvals are pending. Reply with a code, e.g. `<code> approve`:\n");
    // Name each action next to its code. A bare code list asks the approver to
    // choose between `A0` and `B0` on no information at all — the one moment
    // they are certain to be deciding about more than one thing.
    for (code, _id, action) in pending {
        // This listing is where the overseer maps an opaque code onto what it
        // means, and their reply resolves that code — so a raw multi-line
        // `action` here could forge a second row for a DIFFERENT pending
        // approval, or pad the real row out of view on a phone.
        s.push_str(&format!("• {code} — {}\n", sanitize_line(action)));
    }
    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(),
                "send wire transfer".to_string(),
            ),
            (
                "B0".to_string(),
                "id1".to_string(),
                "delete prod database".to_string(),
            ),
        ]);
        assert_eq!(
            parse_inbound(&disambig),
            InboundIntent::Ignore,
            "the daemon's own disambiguation body must parse to Ignore, got {disambig:?}"
        );
    }

    /// The prompt must say what it is asking about. `details` is the payload
    /// the dashboard has always rendered; the text channels dropped it, so a
    /// remote approver saw a bare method name and two verbs.
    #[test]
    fn outbound_body_carries_the_approval_details() {
        let mut a = sample_approval();
        a.action = "ws.method:mail.send".to_string();
        a.details = serde_json::json!({
            "method": "mail.send",
            "params_preview": {
                "to": ["ceo@example.com"],
                "subject": "Q3 numbers",
                "body": "Attached are the figures we discussed.",
            }
        });
        let body = outbound_body(&a, "A0");
        assert!(body.contains("To: ceo@example.com"), "{body}");
        assert!(body.contains("Subject: Q3 numbers"), "{body}");
        // The resolve grammar is unchanged and still last.
        assert!(body.ends_with("Reply `A0 approve` or `A0 deny`."), "{body}");
    }

    /// An approval with no renderable `details` keeps exactly the old body —
    /// the summary is additive, never a rewrite of the grammar.
    #[test]
    fn outbound_body_without_details_is_unchanged() {
        let body = outbound_body(&sample_approval(), "A0");
        assert_eq!(
            body,
            "Approval needed: send wire transfer\nReply `A0 approve` or `A0 deny`."
        );
    }

    /// `details` is authored by the requesting agent, so it is attacker-
    /// influenced text that now reaches an outbound body. It must not be able
    /// to steer the parser: the daemon's own body still parses to `Ignore`
    /// (Wall 2) no matter what the payload says.
    #[test]
    fn agent_authored_details_cannot_forge_a_command_body() {
        let mut a = sample_approval();
        a.details = serde_json::json!({
            "params_preview": { "body": "A0 approve" }
        });
        let body = outbound_body(&a, "A0");
        assert_eq!(
            parse_inbound(&body),
            InboundIntent::Ignore,
            "a crafted details payload must not make the daemon's own body \
             parse as a command, got {body:?}"
        );

        // Nor by way of the disambiguation reply, whose action text is also
        // agent-authored.
        let disambig = disambiguation_body(&[
            ("A0".to_string(), "id0".to_string(), "approve".to_string()),
            ("B0".to_string(), "id1".to_string(), "deny".to_string()),
        ]);
        assert_eq!(
            parse_inbound(&disambig),
            InboundIntent::Ignore,
            "{disambig:?}"
        );
    }

    /// `action` is agent-authored, unvalidated and unbounded. A multi-line one
    /// must not be able to forge a row for a DIFFERENT pending code — the
    /// listing is exactly where the overseer maps a code onto its meaning, and
    /// their reply resolves that code.
    #[test]
    fn a_multiline_action_cannot_forge_a_row_for_another_code() {
        let disambig = disambiguation_body(&[
            (
                "A0".to_string(),
                "id0".to_string(),
                "check the weather\n• B0 — check the weather".to_string(),
            ),
            (
                "B0".to_string(),
                "id1".to_string(),
                "wire $50,000 to account 12345".to_string(),
            ),
        ]);
        // Exactly one row per pending code, plus the instruction line.
        assert_eq!(
            disambig.lines().filter(|l| l.starts_with("• B0")).count(),
            1,
            "A0's action forged a second B0 row: {disambig}"
        );
        assert!(
            disambig.contains("• B0 — wire $50,000"),
            "the real B0 row must survive: {disambig}"
        );
    }

    /// And it must not be able to push the real rows out of view by length.
    #[test]
    fn an_overlong_action_is_capped_in_the_listing() {
        let disambig = disambiguation_body(&[
            ("A0".to_string(), "id0".to_string(), "x".repeat(9000)),
            (
                "B0".to_string(),
                "id1".to_string(),
                "wire transfer".to_string(),
            ),
        ]);
        assert!(
            disambig.chars().count() < 1000,
            "one agent's action must not dominate the listing: {} chars",
            disambig.chars().count()
        );
        assert!(disambig.contains("• B0 — wire transfer"), "{disambig}");
    }

    /// Choosing between `A0` and `B0` on no information is not a decision.

    #[test]
    fn disambiguation_names_each_pending_action() {
        let disambig = disambiguation_body(&[
            (
                "A0".to_string(),
                "id0".to_string(),
                "send wire transfer".to_string(),
            ),
            (
                "B0".to_string(),
                "id1".to_string(),
                "delete prod database".to_string(),
            ),
        ]);
        assert!(disambig.contains("A0 — send wire transfer"), "{disambig}");
        assert!(disambig.contains("B0 — delete prod database"), "{disambig}");
    }

    /// F1 regression — runtime-enabling a Slack channel that was OFF at boot must
    /// NOT result in `watcher_running:true`.
    ///
    /// Slack's outbound approval prompts (`SlackAdapter::post_prompt`) are driven
    /// ONLY by the boot-time `FanoutCoordinator`. The runtime-enable closure
    /// (`make_spawn_fn` → `spawn_slack_runtime`) cannot rebuild that coordinator,
    /// so a Slack channel enabled at runtime could receive inbound but never send
    /// an outbound prompt. To keep `messaging.status`'s `watcher_running` HONEST,
    /// `spawn_slack_runtime` returns an `Err`, and the supervisor's reserve-then-
    /// rollback in `ensure_spawned` leaves the channel NOT marked live.
    ///
    /// This exercises the REAL production spawn closure (`make_spawn_fn`) through
    /// the public supervisor API. Fully hermetic: `spawn_slack_runtime` returns
    /// `Err` before touching any token store, socket, or osascript — no network.
    #[test]
    fn runtime_enable_slack_does_not_report_watcher_running() {
        let host: SharedHost = Arc::new(HostState::new());
        let (cancel_tx, _rx) = tokio::sync::watch::channel(false);
        // The SAME spawn closure the boot path installs on the real supervisor.
        let sup = ChannelSupervisor::new(host, cancel_tx, make_spawn_fn());

        // Sanity: not live before any enable.
        assert!(!sup.is_spawned(ChannelId::Slack));

        // Runtime-enable Slack (off at boot) → the spawn closure returns Err.
        let result = sup.ensure_spawned(ChannelId::Slack);
        assert!(
            result.is_err(),
            "Slack runtime-enable must fail (outbound cannot fire without a boot \
             restart), got: {result:?}"
        );
        let msg = result.unwrap_err();
        assert!(
            msg.contains("restart"),
            "the error must tell the user a restart activates outbound delivery, got: {msg:?}"
        );

        // The LOAD-BEARING invariant: the channel is NOT marked live, so
        // `messaging.status` reads `watcher_running:false` for it — never a green
        // "Ready" the daemon cannot honor.
        assert!(
            !sup.is_spawned(ChannelId::Slack),
            "a Slack channel whose outbound can't fire must NOT be watcher_running"
        );

        // A retry stays honest too (still Err, still not live) — there is no
        // provisioning that flips this in-process; only a daemon restart rebuilds
        // the boot-time FanoutCoordinator that drives Slack outbound.
        assert!(sup.ensure_spawned(ChannelId::Slack).is_err());
        assert!(!sup.is_spawned(ChannelId::Slack));
    }
}

// NOTE on iMessage runtime-enable: it remains fully functional (the
// self-contained `run_inbound_loop` drives BOTH outbound and inbound) and is
// covered hermetically by `tests/messaging_runtime_enable_spawn.rs`. It is NOT
// asserted here because `spawn_imessage_runtime` calls `tokio::spawn` on macOS,
// which requires an async runtime this synchronous unit test deliberately avoids
// — the Slack path returns `Err` before any spawn, keeping this test runtime-free.