storkit/server/src/whatsapp.rs

//! WhatsApp Business API integration.
//!
//! Provides:
//! - [`WhatsAppTransport`] — a [`ChatTransport`] that sends messages via the
//!   Meta Graph API (`graph.facebook.com/v21.0/{phone_number_id}/messages`).
//! - [`MessagingWindowTracker`] — tracks the 24-hour messaging window per user.
//! - [`webhook_verify`] / [`webhook_receive`] — Poem handlers for the WhatsApp
//!   webhook (GET verification handshake + POST incoming messages).

use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::Mutex as TokioMutex;

use crate::agents::AgentPool;
use crate::matrix::{ConversationEntry, ConversationRole, RoomConversation};
use crate::slog;
use crate::transport::{ChatTransport, MessageId};

// ── Graph API base URL (overridable for tests) ──────────────────────────

const GRAPH_API_BASE: &str = "https://graph.facebook.com/v21.0";

/// Graph API error code indicating the 24-hour messaging window has elapsed.
///
/// When Meta returns this code the caller must fall back to an approved message
/// template instead of free-form text.
const ERROR_CODE_OUTSIDE_WINDOW: i64 = 131047;

/// Sentinel error string returned by [`WhatsAppTransport::send_text`] when the
/// Graph API reports that the 24-hour messaging window has expired.
const OUTSIDE_WINDOW_ERR: &str = "OUTSIDE_MESSAGING_WINDOW";

// ── Messaging window tracker ─────────────────────────────────────────────

/// Tracks the 24-hour messaging window per WhatsApp phone number.
///
/// Meta's Business API only permits free-form text messages within 24 hours of
/// the last *inbound* message from that user.  After that window expires, only
/// approved message templates may be sent.
///
/// Call [`record_message`] whenever an inbound message is received.  Before
/// sending a proactive notification, call [`is_within_window`] to choose
/// between free-form text and a template message.
pub struct MessagingWindowTracker {
    last_message: std::sync::Mutex<HashMap<String, std::time::Instant>>,
    window_duration: std::time::Duration,
}

impl Default for MessagingWindowTracker {
    fn default() -> Self {
        Self::new()
    }
}

impl MessagingWindowTracker {
    /// Create a tracker with the standard 24-hour window.
    pub fn new() -> Self {
        Self {
            last_message: std::sync::Mutex::new(HashMap::new()),
            window_duration: std::time::Duration::from_secs(24 * 60 * 60),
        }
    }

    /// Create a tracker with a custom window duration (useful in tests).
    #[cfg(test)]
    fn with_duration(window_duration: std::time::Duration) -> Self {
        Self {
            last_message: std::sync::Mutex::new(HashMap::new()),
            window_duration,
        }
    }

    /// Record that `phone` sent an inbound message right now.
    pub fn record_message(&self, phone: &str) {
        self.last_message
            .lock()
            .unwrap()
            .insert(phone.to_string(), std::time::Instant::now());
    }

    /// Returns `true` when the last inbound message from `phone` arrived within
    /// the 24-hour window, meaning free-form replies are still permitted.
    pub fn is_within_window(&self, phone: &str) -> bool {
        let map = self.last_message.lock().unwrap();
        match map.get(phone) {
            Some(&instant) => instant.elapsed() < self.window_duration,
            None => false,
        }
    }
}

// ── WhatsApp Transport ──────────────────────────────────────────────────

/// Real WhatsApp Business API transport.
///
/// Sends text messages via `POST {GRAPH_API_BASE}/{phone_number_id}/messages`.
/// Falls back to approved notification templates when the 24-hour window has
/// elapsed (Meta error 131047).
pub struct WhatsAppTransport {
    phone_number_id: String,
    access_token: String,
    client: reqwest::Client,
    /// Name of the approved Meta message template used for notifications
    /// outside the 24-hour messaging window.
    notification_template_name: String,
    /// Optional base URL override for tests.
    api_base: String,
}

impl WhatsAppTransport {
    pub fn new(
        phone_number_id: String,
        access_token: String,
        notification_template_name: String,
    ) -> Self {
        Self {
            phone_number_id,
            access_token,
            client: reqwest::Client::new(),
            notification_template_name,
            api_base: GRAPH_API_BASE.to_string(),
        }
    }

    #[cfg(test)]
    fn with_api_base(
        phone_number_id: String,
        access_token: String,
        api_base: String,
    ) -> Self {
        Self {
            phone_number_id,
            access_token,
            client: reqwest::Client::new(),
            notification_template_name: "pipeline_notification".to_string(),
            api_base,
        }
    }

    /// Send a free-form text message to a WhatsApp user via the Graph API.
    ///
    /// Returns [`OUTSIDE_WINDOW_ERR`] if the API responds with error code
    /// 131047 (messaging window expired).  All other errors are returned as
    /// descriptive strings.
    async fn send_text(&self, to: &str, body: &str) -> Result<String, String> {
        let url = format!("{}/{}/messages", self.api_base, self.phone_number_id);

        let payload = GraphSendMessage {
            messaging_product: "whatsapp",
            to,
            r#type: "text",
            text: GraphTextBody { body },
        };

        let resp = self
            .client
            .post(&url)
            .bearer_auth(&self.access_token)
            .json(&payload)
            .send()
            .await
            .map_err(|e| format!("WhatsApp API request failed: {e}"))?;

        let status = resp.status();
        let resp_text = resp
            .text()
            .await
            .unwrap_or_else(|_| "<no body>".to_string());

        if !status.is_success() {
            // Check for 'outside messaging window' (code 131047).  Return a
            // distinct sentinel so callers can fall back to a template without
            // crashing.
            if let Ok(err_body) = serde_json::from_str::<GraphApiErrorResponse>(&resp_text)
                && err_body.error.as_ref().and_then(|e| e.code) == Some(ERROR_CODE_OUTSIDE_WINDOW)
            {
                slog!(
                    "[whatsapp] Outside 24-hour messaging window for {to}; \
                     template required (error 131047)"
                );
                return Err(OUTSIDE_WINDOW_ERR.to_string());
            }
            return Err(format!("WhatsApp API returned {status}: {resp_text}"));
        }

        // Extract the message ID from the response.
        let parsed: GraphSendResponse = serde_json::from_str(&resp_text).map_err(|e| {
            format!("Failed to parse WhatsApp API response: {e} — body: {resp_text}")
        })?;

        let msg_id = parsed
            .messages
            .first()
            .map(|m| m.id.clone())
            .unwrap_or_default();

        Ok(msg_id)
    }

    /// Send an approved template notification message.
    ///
    /// Used when the 24-hour window has expired and free-form text is not
    /// permitted.  The template must already be approved in the Meta Business
    /// Manager under the name configured in `bot.toml`
    /// (`whatsapp_notification_template`, default `pipeline_notification`).
    ///
    /// The template body is expected to accept two positional parameters:
    /// `{{1}}` = story name, `{{2}}` = pipeline stage.
    pub async fn send_template_notification(
        &self,
        to: &str,
        story_name: &str,
        stage: &str,
    ) -> Result<String, String> {
        let url = format!("{}/{}/messages", self.api_base, self.phone_number_id);

        let payload = GraphTemplateMessage {
            messaging_product: "whatsapp",
            to,
            r#type: "template",
            template: GraphTemplate {
                name: &self.notification_template_name,
                language: GraphLanguage { code: "en_US" },
                components: vec![GraphTemplateComponent {
                    r#type: "body",
                    parameters: vec![
                        GraphTemplateParameter {
                            r#type: "text",
                            text: story_name.to_string(),
                        },
                        GraphTemplateParameter {
                            r#type: "text",
                            text: stage.to_string(),
                        },
                    ],
                }],
            },
        };

        let resp = self
            .client
            .post(&url)
            .bearer_auth(&self.access_token)
            .json(&payload)
            .send()
            .await
            .map_err(|e| format!("WhatsApp template API request failed: {e}"))?;

        let status = resp.status();
        let resp_text = resp
            .text()
            .await
            .unwrap_or_else(|_| "<no body>".to_string());

        if !status.is_success() {
            return Err(format!(
                "WhatsApp template API returned {status}: {resp_text}"
            ));
        }

        let parsed: GraphSendResponse = serde_json::from_str(&resp_text).map_err(|e| {
            format!("Failed to parse WhatsApp template API response: {e} — body: {resp_text}")
        })?;

        let msg_id = parsed
            .messages
            .first()
            .map(|m| m.id.clone())
            .unwrap_or_default();

        Ok(msg_id)
    }

    /// Send a pipeline notification, respecting the 24-hour messaging window.
    ///
    /// - Within the window: sends a free-form text message.
    /// - Outside the window (or if the API returns 131047): sends an approved
    ///   template message instead.
    ///
    /// This method never crashes on a messaging-window error — it always
    /// attempts the template fallback and logs what happened.
    pub async fn send_notification(
        &self,
        to: &str,
        tracker: &MessagingWindowTracker,
        story_name: &str,
        stage: &str,
    ) -> Result<String, String> {
        if tracker.is_within_window(to) {
            let text = format!("Story '{story_name}' has moved to {stage}.");
            match self.send_text(to, &text).await {
                Ok(id) => return Ok(id),
                Err(ref e) if e == OUTSIDE_WINDOW_ERR => {
                    // Window expired between our check and the API call —
                    // fall through to the template path.
                    slog!(
                        "[whatsapp] Window expired mid-flight for {to}; \
                         falling back to template"
                    );
                }
                Err(e) => return Err(e),
            }
        }

        // Outside window — use the approved template.
        slog!("[whatsapp] Sending template notification to {to} (outside 24h window)");
        self.send_template_notification(to, story_name, stage).await
    }
}

#[async_trait]
impl ChatTransport for WhatsAppTransport {
    async fn send_message(
        &self,
        recipient: &str,
        plain: &str,
        _html: &str,
    ) -> Result<MessageId, String> {
        slog!("[whatsapp] send_message to {recipient}: {plain:.80}");
        match self.send_text(recipient, plain).await {
            Ok(id) => Ok(id),
            Err(ref e) if e == OUTSIDE_WINDOW_ERR => {
                // Graceful degradation: log and surface a meaningful error
                // rather than crashing.  Callers sending command responses
                // should normally be within the window; this handles the edge
                // case where processing was delayed.
                slog!(
                    "[whatsapp] Cannot send to {recipient}: outside 24h window \
                     (message dropped)"
                );
                Err(format!(
                    "Outside 24-hour messaging window for {recipient}; \
                     send a message to the bot first to re-open the window"
                ))
            }
            Err(e) => Err(e),
        }
    }

    async fn edit_message(
        &self,
        recipient: &str,
        _original_message_id: &str,
        plain: &str,
        html: &str,
    ) -> Result<(), String> {
        // WhatsApp does not support message editing — send a new message.
        slog!("[whatsapp] edit_message — WhatsApp does not support edits, sending new message");
        self.send_message(recipient, plain, html).await.map(|_| ())
    }

    async fn send_typing(&self, _recipient: &str, _typing: bool) -> Result<(), String> {
        // WhatsApp Business API does not expose typing indicators.
        Ok(())
    }
}

// ── Graph API request/response types ────────────────────────────────────

#[derive(Serialize)]
struct GraphSendMessage<'a> {
    messaging_product: &'a str,
    to: &'a str,
    r#type: &'a str,
    text: GraphTextBody<'a>,
}

#[derive(Serialize)]
struct GraphTextBody<'a> {
    body: &'a str,
}

#[derive(Deserialize)]
struct GraphSendResponse {
    #[serde(default)]
    messages: Vec<GraphMessageId>,
}

#[derive(Deserialize)]
struct GraphMessageId {
    id: String,
}

// ── Graph API error response types ─────────────────────────────────────

#[derive(Deserialize)]
struct GraphApiErrorResponse {
    error: Option<GraphApiError>,
}

#[derive(Deserialize)]
struct GraphApiError {
    code: Option<i64>,
    #[allow(dead_code)]
    message: Option<String>,
}

// ── Template message types ──────────────────────────────────────────────

#[derive(Serialize)]
struct GraphTemplateMessage<'a> {
    messaging_product: &'a str,
    to: &'a str,
    r#type: &'a str,
    template: GraphTemplate<'a>,
}

#[derive(Serialize)]
struct GraphTemplate<'a> {
    name: &'a str,
    language: GraphLanguage,
    components: Vec<GraphTemplateComponent>,
}

#[derive(Serialize)]
struct GraphLanguage {
    code: &'static str,
}

#[derive(Serialize)]
struct GraphTemplateComponent {
    r#type: &'static str,
    parameters: Vec<GraphTemplateParameter>,
}

#[derive(Serialize)]
struct GraphTemplateParameter {
    r#type: &'static str,
    text: String,
}

// ── Webhook types (Meta → us) ───────────────────────────────────────────

/// Top-level webhook payload from Meta.
#[derive(Deserialize, Debug)]
pub struct WebhookPayload {
    #[serde(default)]
    pub entry: Vec<WebhookEntry>,
}

#[derive(Deserialize, Debug)]
pub struct WebhookEntry {
    #[serde(default)]
    pub changes: Vec<WebhookChange>,
}

#[derive(Deserialize, Debug)]
pub struct WebhookChange {
    pub value: Option<WebhookValue>,
}

#[derive(Deserialize, Debug)]
pub struct WebhookValue {
    #[serde(default)]
    pub messages: Vec<WebhookMessage>,
    pub metadata: Option<WebhookMetadata>,
}

#[derive(Deserialize, Debug)]
pub struct WebhookMetadata {
    pub phone_number_id: Option<String>,
}

#[derive(Deserialize, Debug)]
pub struct WebhookMessage {
    pub from: Option<String>,
    pub r#type: Option<String>,
    pub text: Option<WebhookText>,
}

#[derive(Deserialize, Debug)]
pub struct WebhookText {
    pub body: Option<String>,
}

/// Extract text messages from a webhook payload.
///
/// Returns `(sender_phone, message_body)` pairs.
pub fn extract_text_messages(payload: &WebhookPayload) -> Vec<(String, String)> {
    let mut messages = Vec::new();
    for entry in &payload.entry {
        for change in &entry.changes {
            if let Some(value) = &change.value {
                for msg in &value.messages {
                    if msg.r#type.as_deref() == Some("text")
                        && let (Some(from), Some(text)) = (&msg.from, &msg.text)
                        && let Some(body) = &text.body
                    {
                        messages.push((from.clone(), body.clone()));
                    }
                }
            }
        }
    }
    messages
}

// ── WhatsApp message size limit ──────────────────────────────────────

/// WhatsApp Business API maximum message body size in characters.
const WHATSAPP_MAX_MESSAGE_LEN: usize = 4096;

/// Split a text into chunks that fit within WhatsApp's message size limit.
///
/// Tries to split on paragraph boundaries (`\n\n`), falling back to line
/// boundaries (`\n`), and finally hard-splitting at the character limit.
pub fn chunk_for_whatsapp(text: &str) -> Vec<String> {
    if text.len() <= WHATSAPP_MAX_MESSAGE_LEN {
        return vec![text.to_string()];
    }

    let mut chunks = Vec::new();
    let mut remaining = text;

    while !remaining.is_empty() {
        if remaining.len() <= WHATSAPP_MAX_MESSAGE_LEN {
            chunks.push(remaining.to_string());
            break;
        }

        // Find the best split point within the limit.
        let window = &remaining[..WHATSAPP_MAX_MESSAGE_LEN];

        // Prefer paragraph boundary.
        let split_pos = window
            .rfind("\n\n")
            .or_else(|| window.rfind('\n'))
            .unwrap_or(WHATSAPP_MAX_MESSAGE_LEN);

        let (chunk, rest) = remaining.split_at(split_pos);
        let chunk = chunk.trim();
        if !chunk.is_empty() {
            chunks.push(chunk.to_string());
        }

        // Skip the delimiter.
        remaining = rest.trim_start_matches('\n');
    }

    chunks
}

// ── Conversation history persistence ─────────────────────────────────

/// Per-sender conversation history, keyed by phone number.
pub type WhatsAppConversationHistory = Arc<TokioMutex<HashMap<String, RoomConversation>>>;

/// On-disk format for persisted WhatsApp conversation history.
#[derive(Serialize, Deserialize)]
struct PersistedWhatsAppHistory {
    senders: HashMap<String, RoomConversation>,
}

/// Path to the persisted WhatsApp conversation history file.
const WHATSAPP_HISTORY_FILE: &str = ".storkit/whatsapp_history.json";

/// Load WhatsApp conversation history from disk.
pub fn load_whatsapp_history(
    project_root: &std::path::Path,
) -> HashMap<String, RoomConversation> {
    let path = project_root.join(WHATSAPP_HISTORY_FILE);
    let data = match std::fs::read_to_string(&path) {
        Ok(d) => d,
        Err(_) => return HashMap::new(),
    };
    let persisted: PersistedWhatsAppHistory = match serde_json::from_str(&data) {
        Ok(p) => p,
        Err(e) => {
            slog!("[whatsapp] Failed to parse history file: {e}");
            return HashMap::new();
        }
    };
    persisted.senders
}

/// Save WhatsApp conversation history to disk.
fn save_whatsapp_history(
    project_root: &std::path::Path,
    history: &HashMap<String, RoomConversation>,
) {
    let persisted = PersistedWhatsAppHistory {
        senders: history.clone(),
    };
    let path = project_root.join(WHATSAPP_HISTORY_FILE);
    match serde_json::to_string_pretty(&persisted) {
        Ok(json) => {
            if let Err(e) = std::fs::write(&path, json) {
                slog!("[whatsapp] Failed to write history file: {e}");
            }
        }
        Err(e) => slog!("[whatsapp] Failed to serialise history: {e}"),
    }
}

// ── Webhook handlers (Poem) ────────────────────────────────────────────

use poem::{Request, Response, handler, http::StatusCode, web::Query};
use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Mutex;

/// Query parameters for the webhook verification GET request.
#[derive(Deserialize)]
pub struct VerifyQuery {
    #[serde(rename = "hub.mode")]
    pub hub_mode: Option<String>,
    #[serde(rename = "hub.verify_token")]
    pub hub_verify_token: Option<String>,
    #[serde(rename = "hub.challenge")]
    pub hub_challenge: Option<String>,
}

/// Shared context for webhook handlers, injected via Poem's `Data` extractor.
pub struct WhatsAppWebhookContext {
    pub verify_token: String,
    pub transport: Arc<WhatsAppTransport>,
    pub project_root: PathBuf,
    pub agents: Arc<AgentPool>,
    pub bot_name: String,
    /// The bot's "user ID" for command dispatch (e.g. "whatsapp-bot").
    pub bot_user_id: String,
    pub ambient_rooms: Arc<Mutex<HashSet<String>>>,
    /// Per-sender conversation history for LLM passthrough.
    pub history: WhatsAppConversationHistory,
    /// Maximum number of conversation entries to keep per sender.
    pub history_size: usize,
    /// Tracks the 24-hour messaging window per user phone number.
    pub window_tracker: Arc<MessagingWindowTracker>,
}

/// GET /webhook/whatsapp — Meta verification handshake.
///
/// Meta sends `hub.mode=subscribe&hub.verify_token=<token>&hub.challenge=<challenge>`.
/// We return the challenge if the token matches.
#[handler]
pub async fn webhook_verify(
    Query(q): Query<VerifyQuery>,
    ctx: poem::web::Data<&Arc<WhatsAppWebhookContext>>,
) -> Response {
    if q.hub_mode.as_deref() == Some("subscribe")
        && q.hub_verify_token.as_deref() == Some(&ctx.verify_token)
        && let Some(challenge) = q.hub_challenge
    {
        slog!("[whatsapp] Webhook verification succeeded");
        return Response::builder()
            .status(StatusCode::OK)
            .body(challenge);
    }
    slog!("[whatsapp] Webhook verification failed");
    Response::builder()
        .status(StatusCode::FORBIDDEN)
        .body("Verification failed")
}

/// POST /webhook/whatsapp — receive incoming messages from Meta.
#[handler]
pub async fn webhook_receive(
    req: &Request,
    body: poem::Body,
    ctx: poem::web::Data<&Arc<WhatsAppWebhookContext>>,
) -> Response {
    let _ = req;
    let bytes = match body.into_bytes().await {
        Ok(b) => b,
        Err(e) => {
            slog!("[whatsapp] Failed to read webhook body: {e}");
            return Response::builder()
                .status(StatusCode::BAD_REQUEST)
                .body("Bad request");
        }
    };

    let payload: WebhookPayload = match serde_json::from_slice(&bytes) {
        Ok(p) => p,
        Err(e) => {
            slog!("[whatsapp] Failed to parse webhook payload: {e}");
            // Meta expects 200 even on parse errors to avoid retries.
            return Response::builder()
                .status(StatusCode::OK)
                .body("ok");
        }
    };

    let messages = extract_text_messages(&payload);
    if messages.is_empty() {
        // Status updates, read receipts, etc. — acknowledge silently.
        return Response::builder()
            .status(StatusCode::OK)
            .body("ok");
    }

    let ctx = Arc::clone(*ctx);
    tokio::spawn(async move {
        for (sender, text) in messages {
            slog!("[whatsapp] Message from {sender}: {text}");
            handle_incoming_message(&ctx, &sender, &text).await;
        }
    });

    Response::builder()
        .status(StatusCode::OK)
        .body("ok")
}

/// Dispatch an incoming WhatsApp message to bot commands.
async fn handle_incoming_message(
    ctx: &WhatsAppWebhookContext,
    sender: &str,
    message: &str,
) {
    use crate::matrix::commands::{CommandDispatch, try_handle_command};

    // Record this inbound message to keep the 24-hour window open.
    ctx.window_tracker.record_message(sender);

    let dispatch = CommandDispatch {
        bot_name: &ctx.bot_name,
        bot_user_id: &ctx.bot_user_id,
        project_root: &ctx.project_root,
        agents: &ctx.agents,
        ambient_rooms: &ctx.ambient_rooms,
        room_id: sender,
    };

    if let Some(response) = try_handle_command(&dispatch, message) {
        slog!("[whatsapp] Sending command response to {sender}");
        if let Err(e) = ctx.transport.send_message(sender, &response, "").await {
            slog!("[whatsapp] Failed to send reply to {sender}: {e}");
        }
        return;
    }

    // Check for async commands (htop, delete).
    if let Some(htop_cmd) = crate::matrix::htop::extract_htop_command(
        message,
        &ctx.bot_name,
        &ctx.bot_user_id,
    ) {
        use crate::matrix::htop::HtopCommand;
        slog!("[whatsapp] Handling htop command from {sender}");
        match htop_cmd {
            HtopCommand::Stop => {
                // htop stop — no-op on WhatsApp since there's no persistent
                // editable message; just acknowledge.
                let _ = ctx
                    .transport
                    .send_message(sender, "htop stopped.", "")
                    .await;
            }
            HtopCommand::Start { duration_secs } => {
                // On WhatsApp, send a single snapshot instead of a live-updating
                // dashboard since we can't edit messages.
                let snapshot =
                    crate::matrix::htop::build_htop_message(&ctx.agents, 0, duration_secs);
                let _ = ctx
                    .transport
                    .send_message(sender, &snapshot, "")
                    .await;
            }
        }
        return;
    }

    if let Some(del_cmd) = crate::matrix::delete::extract_delete_command(
        message,
        &ctx.bot_name,
        &ctx.bot_user_id,
    ) {
        let response = match del_cmd {
            crate::matrix::delete::DeleteCommand::Delete { story_number } => {
                slog!("[whatsapp] Handling delete command from {sender}: story {story_number}");
                crate::matrix::delete::handle_delete(
                    &ctx.bot_name,
                    &story_number,
                    &ctx.project_root,
                    &ctx.agents,
                )
                .await
            }
            crate::matrix::delete::DeleteCommand::BadArgs => {
                format!("Usage: `{} delete <number>`", ctx.bot_name)
            }
        };
        let _ = ctx.transport.send_message(sender, &response, "").await;
        return;
    }

    // No command matched — forward to LLM for conversational response.
    slog!("[whatsapp] No command matched, forwarding to LLM for {sender}");
    handle_llm_message(ctx, sender, message).await;
}

/// Forward a message to Claude Code and send the response back via WhatsApp.
async fn handle_llm_message(
    ctx: &WhatsAppWebhookContext,
    sender: &str,
    user_message: &str,
) {
    use crate::llm::providers::claude_code::{ClaudeCodeProvider, ClaudeCodeResult};
    use crate::matrix::drain_complete_paragraphs;
    use std::sync::atomic::{AtomicBool, Ordering};
    use tokio::sync::watch;

    // Look up existing session ID for this sender.
    let resume_session_id: Option<String> = {
        let guard = ctx.history.lock().await;
        guard
            .get(sender)
            .and_then(|conv| conv.session_id.clone())
    };

    let bot_name = &ctx.bot_name;
    let prompt = format!(
        "[Your name is {bot_name}. Refer to yourself as {bot_name}, not Claude.]\n\n{sender}: {user_message}"
    );

    let provider = ClaudeCodeProvider::new();
    let (_cancel_tx, mut cancel_rx) = watch::channel(false);

    // Channel for sending complete chunks to the WhatsApp posting task.
    let (msg_tx, mut msg_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
    let msg_tx_for_callback = msg_tx.clone();

    // Spawn a task to post messages as they arrive.
    let post_transport = Arc::clone(&ctx.transport);
    let post_sender = sender.to_string();
    let post_task = tokio::spawn(async move {
        while let Some(chunk) = msg_rx.recv().await {
            // Split into WhatsApp-sized chunks.
            for part in chunk_for_whatsapp(&chunk) {
                let _ = post_transport.send_message(&post_sender, &part, "").await;
            }
        }
    });

    // Shared buffer between the sync token callback and the async scope.
    let buffer = Arc::new(std::sync::Mutex::new(String::new()));
    let buffer_for_callback = Arc::clone(&buffer);
    let sent_any_chunk = Arc::new(AtomicBool::new(false));
    let sent_any_chunk_for_callback = Arc::clone(&sent_any_chunk);

    let project_root_str = ctx.project_root.to_string_lossy().to_string();
    let result = provider
        .chat_stream(
            &prompt,
            &project_root_str,
            resume_session_id.as_deref(),
            None,
            &mut cancel_rx,
            move |token| {
                let mut buf = buffer_for_callback.lock().unwrap();
                buf.push_str(token);
                let paragraphs = drain_complete_paragraphs(&mut buf);
                for chunk in paragraphs {
                    sent_any_chunk_for_callback.store(true, Ordering::Relaxed);
                    let _ = msg_tx_for_callback.send(chunk);
                }
            },
            |_thinking| {},
            |_activity| {},
        )
        .await;

    // Flush remaining text.
    let remaining = buffer.lock().unwrap().trim().to_string();
    let did_send_any = sent_any_chunk.load(Ordering::Relaxed);

    let (assistant_reply, new_session_id) = match result {
        Ok(ClaudeCodeResult {
            messages,
            session_id,
        }) => {
            let reply = if !remaining.is_empty() {
                let _ = msg_tx.send(remaining.clone());
                remaining
            } else if !did_send_any {
                let last_text = messages
                    .iter()
                    .rev()
                    .find(|m| {
                        m.role == crate::llm::types::Role::Assistant && !m.content.is_empty()
                    })
                    .map(|m| m.content.clone())
                    .unwrap_or_default();
                if !last_text.is_empty() {
                    let _ = msg_tx.send(last_text.clone());
                }
                last_text
            } else {
                remaining
            };
            slog!("[whatsapp] session_id from chat_stream: {:?}", session_id);
            (reply, session_id)
        }
        Err(e) => {
            slog!("[whatsapp] LLM error: {e}");
            let err_msg = format!("Error processing your request: {e}");
            let _ = msg_tx.send(err_msg.clone());
            (err_msg, None)
        }
    };

    // Signal the posting task to finish and wait for it.
    drop(msg_tx);
    let _ = post_task.await;

    // Record this exchange in conversation history.
    if !assistant_reply.starts_with("Error processing") {
        let mut guard = ctx.history.lock().await;
        let conv = guard.entry(sender.to_string()).or_default();

        if new_session_id.is_some() {
            conv.session_id = new_session_id;
        }

        conv.entries.push(ConversationEntry {
            role: ConversationRole::User,
            sender: sender.to_string(),
            content: user_message.to_string(),
        });
        conv.entries.push(ConversationEntry {
            role: ConversationRole::Assistant,
            sender: String::new(),
            content: assistant_reply,
        });

        // Trim to configured maximum.
        if conv.entries.len() > ctx.history_size {
            let excess = conv.entries.len() - ctx.history_size;
            conv.entries.drain(..excess);
        }

        save_whatsapp_history(&ctx.project_root, &guard);
    }
}

// ── Tests ───────────────────────────────────────────────────────────────

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

    // ── MessagingWindowTracker ────────────────────────────────────────

    #[test]
    fn window_tracker_unknown_user_is_outside_window() {
        let tracker = MessagingWindowTracker::new();
        assert!(!tracker.is_within_window("15551234567"));
    }

    #[test]
    fn window_tracker_records_within_window() {
        let tracker = MessagingWindowTracker::new();
        tracker.record_message("15551234567");
        assert!(tracker.is_within_window("15551234567"));
    }

    #[test]
    fn window_tracker_expired_window_returns_false() {
        // Use a 1-nanosecond window so it expires immediately.
        let tracker = MessagingWindowTracker::with_duration(std::time::Duration::from_nanos(1));
        tracker.record_message("15551234567");
        // Sleep briefly to ensure the instant has elapsed.
        std::thread::sleep(std::time::Duration::from_millis(1));
        assert!(!tracker.is_within_window("15551234567"));
    }

    #[test]
    fn window_tracker_tracks_users_independently() {
        let tracker = MessagingWindowTracker::new();
        tracker.record_message("111");
        assert!(tracker.is_within_window("111"));
        assert!(!tracker.is_within_window("222"));
    }

    // ── send_text error handling ───────────────────────────────────────

    #[tokio::test]
    async fn send_text_handles_131047_outside_window_error() {
        let mut server = mockito::Server::new_async().await;
        server
            .mock("POST", "/123456/messages")
            .with_status(400)
            .with_body(
                r#"{"error":{"message":"More than 24 hours have passed","type":"OAuthException","code":131047}}"#,
            )
            .create_async()
            .await;

        let transport = WhatsAppTransport::with_api_base(
            "123456".to_string(),
            "test-token".to_string(),
            server.url(),
        );

        let result = transport.send_text("15551234567", "hello").await;
        assert!(result.is_err());
        assert_eq!(result.unwrap_err(), OUTSIDE_WINDOW_ERR);
    }

    #[tokio::test]
    async fn send_message_handles_outside_window_gracefully() {
        let mut server = mockito::Server::new_async().await;
        server
            .mock("POST", "/123456/messages")
            .with_status(400)
            .with_body(
                r#"{"error":{"message":"More than 24 hours have passed","type":"OAuthException","code":131047}}"#,
            )
            .create_async()
            .await;

        let transport = WhatsAppTransport::with_api_base(
            "123456".to_string(),
            "test-token".to_string(),
            server.url(),
        );

        // send_message must not panic — it returns Err with a human-readable message.
        let result = transport.send_message("15551234567", "hello", "").await;
        assert!(result.is_err());
        let msg = result.unwrap_err();
        assert!(msg.contains("24-hour messaging window"), "unexpected: {msg}");
    }

    // ── send_template_notification ────────────────────────────────────

    #[tokio::test]
    async fn send_template_notification_calls_graph_api() {
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/123456/messages")
            .match_header("authorization", "Bearer test-token")
            .match_body(mockito::Matcher::PartialJsonString(
                r#"{"type":"template"}"#.to_string(),
            ))
            .with_body(r#"{"messages": [{"id": "wamid.tpl123"}]}"#)
            .create_async()
            .await;

        let transport = WhatsAppTransport::with_api_base(
            "123456".to_string(),
            "test-token".to_string(),
            server.url(),
        );

        let result = transport
            .send_template_notification("15551234567", "my-story", "done")
            .await;
        assert!(result.is_ok(), "unexpected err: {:?}", result.err());
        assert_eq!(result.unwrap(), "wamid.tpl123");
        mock.assert_async().await;
    }

    #[tokio::test]
    async fn send_notification_uses_text_within_window() {
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/123456/messages")
            .match_body(mockito::Matcher::PartialJsonString(
                r#"{"type":"text"}"#.to_string(),
            ))
            .with_body(r#"{"messages": [{"id": "wamid.txt1"}]}"#)
            .create_async()
            .await;

        let transport = WhatsAppTransport::with_api_base(
            "123456".to_string(),
            "test-token".to_string(),
            server.url(),
        );
        let tracker = MessagingWindowTracker::new();
        tracker.record_message("15551234567");

        let result = transport
            .send_notification("15551234567", &tracker, "my-story", "done")
            .await;
        assert!(result.is_ok());
        mock.assert_async().await;
    }

    #[tokio::test]
    async fn send_notification_uses_template_outside_window() {
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/123456/messages")
            .match_body(mockito::Matcher::PartialJsonString(
                r#"{"type":"template"}"#.to_string(),
            ))
            .with_body(r#"{"messages": [{"id": "wamid.tpl2"}]}"#)
            .create_async()
            .await;

        let transport = WhatsAppTransport::with_api_base(
            "123456".to_string(),
            "test-token".to_string(),
            server.url(),
        );
        // No record_message call — user is outside the window.
        let tracker = MessagingWindowTracker::new();

        let result = transport
            .send_notification("15551234567", &tracker, "my-story", "done")
            .await;
        assert!(result.is_ok());
        mock.assert_async().await;
    }

    // ── Existing webhook / transport tests ────────────────────────────

    #[test]
    fn extract_text_messages_parses_valid_payload() {
        let json = r#"{
            "entry": [{
                "changes": [{
                    "value": {
                        "messages": [{
                            "from": "15551234567",
                            "type": "text",
                            "text": { "body": "help" }
                        }],
                        "metadata": { "phone_number_id": "123456" }
                    }
                }]
            }]
        }"#;
        let payload: WebhookPayload = serde_json::from_str(json).unwrap();
        let msgs = extract_text_messages(&payload);
        assert_eq!(msgs.len(), 1);
        assert_eq!(msgs[0].0, "15551234567");
        assert_eq!(msgs[0].1, "help");
    }

    #[test]
    fn extract_text_messages_ignores_non_text() {
        let json = r#"{
            "entry": [{
                "changes": [{
                    "value": {
                        "messages": [{
                            "from": "15551234567",
                            "type": "image",
                            "image": { "id": "img123" }
                        }],
                        "metadata": { "phone_number_id": "123456" }
                    }
                }]
            }]
        }"#;
        let payload: WebhookPayload = serde_json::from_str(json).unwrap();
        let msgs = extract_text_messages(&payload);
        assert!(msgs.is_empty());
    }

    #[test]
    fn extract_text_messages_handles_empty_payload() {
        let json = r#"{ "entry": [] }"#;
        let payload: WebhookPayload = serde_json::from_str(json).unwrap();
        let msgs = extract_text_messages(&payload);
        assert!(msgs.is_empty());
    }

    #[test]
    fn extract_text_messages_handles_multiple_messages() {
        let json = r#"{
            "entry": [{
                "changes": [{
                    "value": {
                        "messages": [
                            { "from": "111", "type": "text", "text": { "body": "status" } },
                            { "from": "222", "type": "text", "text": { "body": "help" } }
                        ],
                        "metadata": { "phone_number_id": "123456" }
                    }
                }]
            }]
        }"#;
        let payload: WebhookPayload = serde_json::from_str(json).unwrap();
        let msgs = extract_text_messages(&payload);
        assert_eq!(msgs.len(), 2);
        assert_eq!(msgs[0].1, "status");
        assert_eq!(msgs[1].1, "help");
    }

    #[tokio::test]
    async fn transport_send_message_calls_graph_api() {
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/123456/messages")
            .match_header("authorization", "Bearer test-token")
            .with_body(r#"{"messages": [{"id": "wamid.abc123"}]}"#)
            .create_async()
            .await;

        let transport = WhatsAppTransport::with_api_base(
            "123456".to_string(),
            "test-token".to_string(),
            server.url(),
        );

        let result = transport
            .send_message("15551234567", "hello", "<p>hello</p>")
            .await;
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), "wamid.abc123");
        mock.assert_async().await;
    }

    #[tokio::test]
    async fn transport_edit_sends_new_message() {
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/123456/messages")
            .with_body(r#"{"messages": [{"id": "wamid.xyz"}]}"#)
            .create_async()
            .await;

        let transport = WhatsAppTransport::with_api_base(
            "123456".to_string(),
            "test-token".to_string(),
            server.url(),
        );

        let result = transport
            .edit_message("15551234567", "old-msg-id", "updated", "<p>updated</p>")
            .await;
        assert!(result.is_ok());
        mock.assert_async().await;
    }

    #[tokio::test]
    async fn transport_send_typing_succeeds() {
        let transport =
            WhatsAppTransport::new("123".to_string(), "tok".to_string(), "tpl".to_string());
        assert!(transport.send_typing("room1", true).await.is_ok());
        assert!(transport.send_typing("room1", false).await.is_ok());
    }

    #[tokio::test]
    async fn transport_handles_api_error() {
        let mut server = mockito::Server::new_async().await;
        server
            .mock("POST", "/123456/messages")
            .with_status(401)
            .with_body(r#"{"error": {"message": "Invalid token"}}"#)
            .create_async()
            .await;

        let transport = WhatsAppTransport::with_api_base(
            "123456".to_string(),
            "bad-token".to_string(),
            server.url(),
        );

        let result = transport.send_message("15551234567", "hello", "").await;
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("401"));
    }

    // ── chunk_for_whatsapp tests ────────────────────────────────────────

    #[test]
    fn chunk_short_message_returns_single_chunk() {
        let chunks = chunk_for_whatsapp("Hello world");
        assert_eq!(chunks, vec!["Hello world"]);
    }

    #[test]
    fn chunk_exactly_at_limit_returns_single_chunk() {
        let text = "a".repeat(WHATSAPP_MAX_MESSAGE_LEN);
        let chunks = chunk_for_whatsapp(&text);
        assert_eq!(chunks.len(), 1);
        assert_eq!(chunks[0].len(), WHATSAPP_MAX_MESSAGE_LEN);
    }

    #[test]
    fn chunk_splits_on_paragraph_boundary() {
        // Create text with a paragraph boundary near the split point.
        let first_para = "a".repeat(4000);
        let second_para = "b".repeat(200);
        let text = format!("{first_para}\n\n{second_para}");
        let chunks = chunk_for_whatsapp(&text);
        assert_eq!(chunks.len(), 2);
        assert_eq!(chunks[0], first_para);
        assert_eq!(chunks[1], second_para);
    }

    #[test]
    fn chunk_splits_on_line_boundary_when_no_paragraph_break() {
        let first_line = "a".repeat(4000);
        let second_line = "b".repeat(200);
        let text = format!("{first_line}\n{second_line}");
        let chunks = chunk_for_whatsapp(&text);
        assert_eq!(chunks.len(), 2);
        assert_eq!(chunks[0], first_line);
        assert_eq!(chunks[1], second_line);
    }

    #[test]
    fn chunk_hard_splits_continuous_text() {
        let text = "x".repeat(WHATSAPP_MAX_MESSAGE_LEN * 2 + 100);
        let chunks = chunk_for_whatsapp(&text);
        assert!(chunks.len() >= 2);
        for chunk in &chunks {
            assert!(chunk.len() <= WHATSAPP_MAX_MESSAGE_LEN);
        }
        // Verify all content is preserved.
        let reassembled: String = chunks.join("");
        assert_eq!(reassembled.len(), text.len());
    }

    #[test]
    fn chunk_empty_string_returns_single_empty() {
        let chunks = chunk_for_whatsapp("");
        assert_eq!(chunks, vec![""]);
    }

    // ── WhatsApp history persistence tests ──────────────────────────────

    #[test]
    fn save_and_load_whatsapp_history_round_trips() {
        let tmp = tempfile::tempdir().unwrap();
        let sk = tmp.path().join(".storkit");
        std::fs::create_dir_all(&sk).unwrap();

        let mut history = HashMap::new();
        history.insert(
            "15551234567".to_string(),
            RoomConversation {
                session_id: Some("sess-abc".to_string()),
                entries: vec![
                    ConversationEntry {
                        role: ConversationRole::User,
                        sender: "15551234567".to_string(),
                        content: "hello".to_string(),
                    },
                    ConversationEntry {
                        role: ConversationRole::Assistant,
                        sender: String::new(),
                        content: "hi there!".to_string(),
                    },
                ],
            },
        );

        save_whatsapp_history(tmp.path(), &history);
        let loaded = load_whatsapp_history(tmp.path());

        assert_eq!(loaded.len(), 1);
        let conv = loaded.get("15551234567").unwrap();
        assert_eq!(conv.session_id.as_deref(), Some("sess-abc"));
        assert_eq!(conv.entries.len(), 2);
        assert_eq!(conv.entries[0].content, "hello");
        assert_eq!(conv.entries[1].content, "hi there!");
    }

    #[test]
    fn load_whatsapp_history_returns_empty_when_file_missing() {
        let tmp = tempfile::tempdir().unwrap();
        let history = load_whatsapp_history(tmp.path());
        assert!(history.is_empty());
    }

    #[test]
    fn load_whatsapp_history_returns_empty_on_invalid_json() {
        let tmp = tempfile::tempdir().unwrap();
        let sk = tmp.path().join(".storkit");
        std::fs::create_dir_all(&sk).unwrap();
        std::fs::write(sk.join("whatsapp_history.json"), "not json {{{").unwrap();
        let history = load_whatsapp_history(tmp.path());
        assert!(history.is_empty());
    }

    #[test]
    fn save_whatsapp_history_preserves_multiple_senders() {
        let tmp = tempfile::tempdir().unwrap();
        let sk = tmp.path().join(".storkit");
        std::fs::create_dir_all(&sk).unwrap();

        let mut history = HashMap::new();
        history.insert(
            "111".to_string(),
            RoomConversation {
                session_id: None,
                entries: vec![ConversationEntry {
                    role: ConversationRole::User,
                    sender: "111".to_string(),
                    content: "msg1".to_string(),
                }],
            },
        );
        history.insert(
            "222".to_string(),
            RoomConversation {
                session_id: Some("sess-222".to_string()),
                entries: vec![ConversationEntry {
                    role: ConversationRole::User,
                    sender: "222".to_string(),
                    content: "msg2".to_string(),
                }],
            },
        );

        save_whatsapp_history(tmp.path(), &history);
        let loaded = load_whatsapp_history(tmp.path());

        assert_eq!(loaded.len(), 2);
        assert!(loaded.contains_key("111"));
        assert!(loaded.contains_key("222"));
        assert_eq!(loaded["222"].session_id.as_deref(), Some("sess-222"));
    }
}