forge_workspace/

parked.rs

//! Payloads that arrived while their slot had no live session, addressed
//! by `(org, project, label)` - the triple the `sessions` table is keyed
//! by.

use std::collections::HashMap;
use std::sync::Arc;

use crate::SessionSlot;
use crate::mcp::gotify::types::GotifyNotification;
use crate::mcp::peers::types::{PeerFailureReason, WrappedPrompt};

/// Everything parked for one slot, in the order a connecting session
/// delivers it.
#[derive(Default)]
pub(crate) struct ParkedForSlot {
    pub peer: Vec<ParkedPeer>,
    /// Prompts held because the slot's session was mid-turn when they
    /// arrived, drained at that turn's end rather than at a connect.
    pub turn_held: Vec<ParkedPeer>,
    pub cron: Vec<crate::crons::PendingCron>,
    pub gotify: Vec<GotifyNotification>,
    pub slack: Vec<forge_primitives::slack::SlackMessage>,
}

/// One peer message waiting for its slot to connect, or held for its
/// turn to end, with the seat that sent it.
///
/// The sender rides here for the DELIVERY ACK alone: if the spawn this
/// waited on fails, the bucket is dropped and the sender is told. It is
/// not outstanding-ask tracking - nothing looks an entry up by id, and
/// nothing here survives the delivery.
pub(crate) struct ParkedPeer {
    pub(crate) sender: SessionSlot,
    pub(crate) wrapped: WrappedPrompt,
}

/// Parked payloads, one bucket per slot.
pub(crate) type ParkedMap = HashMap<SessionSlot, ParkedForSlot>;

impl crate::Workspace {
    /// Park a peer prompt for `slot`, drained by the session that next
    /// connects as that slot.
    pub(crate) fn park_peer_prompt(
        &self,
        slot: &SessionSlot,
        sender: &SessionSlot,
        wrapped: WrappedPrompt,
    ) {
        self.parked_by_slot
            .lock()
            .entry(slot.clone())
            .or_default()
            .peer
            .push(ParkedPeer { sender: sender.clone(), wrapped });
    }

    /// Whether `slot`'s bucket still holds the peer prompt with `id`. The
    /// wake path parks before it dispatches a spawn, so an entry the spawn's
    /// own failure expired means the delivery was not accepted.
    pub(crate) fn peer_prompt_is_parked(
        &self,
        slot: &SessionSlot,
        id: &crate::mcp::peers::types::MessageId,
    ) -> bool {
        self.parked_by_slot
            .lock()
            .get(slot)
            .is_some_and(|bucket| bucket.peer.iter().any(|parked| parked.wrapped.id == *id))
    }

    /// Park a fired cron prompt for `slot`, missed-marked when it came
    /// due while the owner was asleep. The entry's identity rides along so
    /// the drained echo can name the schedule it came from.
    pub(crate) fn park_cron(
        &self,
        slot: &SessionSlot,
        cron: &forge_primitives::CronEntry,
        missed: bool,
    ) {
        self.parked_by_slot.lock().entry(slot.clone()).or_default().cron.push(
            crate::crons::PendingCron {
                text: cron.prompt.clone(),
                missed,
                cron_id: cron.id.as_str().to_owned(),
                description: cron.description.clone(),
            },
        );
    }

    /// Park a Gotify notification for `slot`.
    pub(crate) fn park_gotify(&self, slot: &SessionSlot, notification: GotifyNotification) {
        self.parked_by_slot.lock().entry(slot.clone()).or_default().gotify.push(notification);
    }

    /// Park a Slack batch for `slot`.
    pub(crate) fn park_slack(
        &self,
        slot: &SessionSlot,
        messages: Vec<forge_primitives::slack::SlackMessage>,
    ) {
        self.parked_by_slot.lock().entry(slot.clone()).or_default().slack.extend(messages);
    }

    /// Hold a peer prompt for `slot` until the turn it arrived under ends,
    /// when the seat is idle again and the prompt can start a turn of its
    /// own - unless the turn has already ended, in which case the prompt
    /// comes back for delivery now.
    ///
    /// `turn_running` is read while the bucket's own lock is held, and the
    /// turn-end drain ends the turn's markers before it takes the bucket:
    /// the take's lock is what orders the two, so a delivery is either
    /// enqueued before a drain (and delivered by it) or reads the ended turn
    /// after that drain's take (and is delivered by its caller) - never left
    /// behind a drain that already passed.
    pub(crate) fn hold_peer_prompt_unless_turn_ended(
        &self,
        slot: &SessionSlot,
        sender: &SessionSlot,
        wrapped: WrappedPrompt,
        turn_running: impl FnOnce() -> bool,
    ) -> Option<WrappedPrompt> {
        let mut map = self.parked_by_slot.lock();
        if !turn_running() {
            return Some(wrapped);
        }
        map.entry(slot.clone())
            .or_default()
            .turn_held
            .push(ParkedPeer { sender: sender.clone(), wrapped });
        None
    }

    /// Take (and clear) the prompts held for `slot`'s running turn, in
    /// the order they arrived.
    pub(crate) fn take_turn_held_for_slot(&self, slot: &SessionSlot) -> Vec<ParkedPeer> {
        let mut map = self.parked_by_slot.lock();
        let Some(bucket) = map.get_mut(slot) else { return Vec::new() };
        std::mem::take(&mut bucket.turn_held)
    }

    /// Take (and clear) everything parked for the connecting session's
    /// slot. The caller passes the triple it was spawned with, rather
    /// than resolving one from its cwd: a cwd under no configured
    /// project would resolve to nothing and leave the payloads parked
    /// forever, silently.
    ///
    /// A prompt held for a turn end rides the peer list: the slot that
    /// connected is a different, idle session, and dropping the hold with
    /// the bucket would lose it silently.
    pub(crate) fn take_parked_for_slot(&self, slot: &SessionSlot) -> ParkedForSlot {
        let mut taken = self.parked_by_slot.lock().remove(slot).unwrap_or_default();
        taken.peer.append(&mut taken.turn_held);
        taken
    }

    /// Drop everything parked for `slot`, acknowledging each peer message
    /// back to its sender so a message that never landed is not left
    /// looking delivered. The parked message is the only payload with a
    /// sender, so the Gotify and Slack drops have no one to tell and are
    /// logged instead.
    pub(crate) fn expire_parked_for_slot(
        self: &Arc<Self>,
        slot: &SessionSlot,
        reason: PeerFailureReason,
    ) {
        let Some(parked) = self.parked_by_slot.lock().remove(slot) else { return };
        for entry in parked.peer.into_iter().chain(parked.turn_held) {
            self.notice_undelivered_message(&entry.sender, slot, reason);
        }
        for notification in parked.gotify {
            tracing::warn!(
                target: "forge_workspace::spawn",
                org = %slot.org(),
                project = %slot.project(),
                label = %slot.label(),
                app = %notification.app,
                title = %notification.title,
                "gotify notification dropped: the spawn it was parked for failed",
            );
        }
        for message in parked.slack {
            tracing::warn!(
                target: "forge_workspace::spawn",
                org = %slot.org(),
                project = %slot.project(),
                label = %slot.label(),
                conversation = %message.conversation,
                ts = %message.ts,
                "slack message dropped: the spawn it was parked for failed",
            );
        }
    }
}

#[cfg(test)]
mod tests {
    use super::ParkedForSlot;
    use crate::SessionSlot;
    use crate::mcp::peers::types::{MessageId, PeerFailureReason, WrappedKind, WrappedPrompt};
    use crate::protocol::SessionUpdate;

    fn slot(label: Option<&str>) -> SessionSlot {
        SessionSlot::for_label("TestOrg", "parked-proj", label)
    }

    fn sender() -> SessionSlot {
        SessionSlot::for_label("TestOrg", "sender-proj", None)
    }

    fn wrapped(id: &MessageId, body: &str) -> WrappedPrompt {
        WrappedPrompt {
            id: id.clone(),
            kind: WrappedKind::Message,
            sender_name: "forge".to_owned(),
            sender_org: "Personal".to_owned(),
            body: body.to_owned(),
        }
    }

    /// A cron entry carrying what a park reads: the id, the prompt and a
    /// description.
    fn cron(id: &str, prompt: &str) -> forge_primitives::cron::CronEntry {
        use forge_primitives::cron::{CronEntry, CronId, CronKind};
        CronEntry {
            id: CronId::from(id),
            project_name: "parked-proj".to_owned(),
            kind: CronKind::Recurring("0 9 * * *".to_owned()),
            prompt: prompt.to_owned(),
            created_at: std::time::SystemTime::UNIX_EPOCH,
            description: Some(format!("{id} summary")),
            last_fire: None,
            next_fire: std::time::SystemTime::UNIX_EPOCH,
            team_role: None,
            wake: true,
        }
    }

    /// A payload parked for a sleeping project is taken by the session
    /// that connects as that slot, and the take is a drain, not a read.
    #[test]
    fn a_parked_payload_lands_on_the_slots_connect() {
        let (ws, _rx) = crate::Workspace::testing_stub();
        let id = MessageId::mint();
        ws.park_peer_prompt(&slot(None), &sender(), wrapped(&id, "are you up?"));

        let taken = ws.take_parked_for_slot(&slot(None));
        assert_eq!(taken.peer.len(), 1, "the slot's own connect takes its bucket");
        assert_eq!(taken.peer[0].wrapped.id, id, "and it is the payload that was parked");
        assert!(
            ws.take_parked_for_slot(&slot(None)).peer.is_empty(),
            "the take drains: a second connect does not re-deliver",
        );
    }

    /// All four parked kinds live in the one bucket, so a cron and a peer
    /// prompt parked for the same slot come back together.
    #[test]
    fn one_bucket_holds_every_kind() {
        let (ws, _rx) = crate::Workspace::testing_stub();
        let id = MessageId::mint();
        ws.park_peer_prompt(&slot(None), &sender(), wrapped(&id, "are you up?"));
        ws.park_cron(&slot(None), &cron("c1", "morning reminder"), true);

        let taken: ParkedForSlot = ws.take_parked_for_slot(&slot(None));
        assert_eq!(taken.peer.len(), 1, "the peer prompt is here");
        assert_eq!(taken.cron.len(), 1, "and so is the cron");
        assert!(taken.cron[0].missed, "with its missed mark intact");
        assert_eq!(taken.cron[0].cron_id, "c1", "and the entry it came from");
        assert_eq!(taken.cron[0].description.as_deref(), Some("c1 summary"));
    }

    /// The label is half the address. A payload parked for a team worker
    /// must not land on the project's lead, which is the silent miss this
    /// bucket shape exists to prevent.
    #[test]
    fn a_workers_parked_payload_is_not_the_leads() {
        let (ws, _rx) = crate::Workspace::testing_stub();
        let id = MessageId::mint();
        ws.park_peer_prompt(&slot(Some("planner")), &sender(), wrapped(&id, "planner"));

        assert!(
            ws.take_parked_for_slot(&slot(None)).peer.is_empty(),
            "the lead does not drain a worker's bucket",
        );
        assert_eq!(
            ws.take_parked_for_slot(&slot(Some("planner"))).peer.len(),
            1,
            "the worker drains its own",
        );
    }

    /// The org is part of the key: a payload parked for one org's project
    /// is not drained by a session under another org that happens to share
    /// the project name.
    #[test]
    fn a_parked_payload_is_not_drained_across_orgs() {
        let (ws, _rx) = crate::Workspace::testing_stub();
        let id = MessageId::mint();
        ws.park_peer_prompt(
            &SessionSlot::lead("OtherOrg", "parked-proj"),
            &sender(),
            wrapped(&id, "elsewhere"),
        );

        assert!(
            ws.take_parked_for_slot(&slot(None)).peer.is_empty(),
            "TestOrg's slot does not take OtherOrg's bucket",
        );
    }

    /// A prompt held for a turn end that never comes - the session died
    /// first - is told to its sender rather than dropped with the bucket.
    /// The hold is a wait, not a promise: only the turn ending delivers it.
    #[test]
    fn expiring_a_slot_acknowledges_its_turn_held_prompts_to_the_sender() {
        let (ws, mut rx) = crate::Workspace::testing_stub();
        let target = slot(Some("planner"));
        let id = MessageId::mint();
        ws.hold_peer_prompt_unless_turn_ended(
            &target,
            &sender(),
            wrapped(&id, "while you work"),
            || true,
        );

        // Nothing else is parked for this slot, so the notice below can only
        // have come from the held entry.
        ws.expire_parked_for_slot(&target, PeerFailureReason::TargetConnectionFailed);

        let mut echo = None;
        while let Ok(frame) = rx.try_recv() {
            if let SessionUpdate::PeerEnvelopeAppended { key, wrapped, .. } =
                std::sync::Arc::unwrap_or_clone(frame.update)
            {
                echo = Some((key, wrapped));
            }
        }
        let (key, notice) = echo.expect("the notice is painted for the sender");
        assert_eq!(key, sender(), "the notice lands on the sender, not the target");
        assert!(
            matches!(notice.kind, WrappedKind::DeliveryFailureNotice),
            "the held prompt's sender hears it never landed: {notice:?}",
        );
        assert!(
            ws.take_turn_held_for_slot(&target).is_empty(),
            "and the hold is gone with the bucket",
        );
    }

    /// A spawn that never connects expires what it parked: the sender is
    /// told its message never landed, and the bucket does not survive to
    /// leak into a later session.
    ///
    /// This is the delivery-ack path, and it is the whole reason the
    /// parked entry carries a sender - with the ask registry gone, nothing
    /// else knows who to tell.
    #[test]
    fn expiring_a_slot_acknowledges_its_parked_messages_to_the_sender() {
        let (ws, mut rx) = crate::Workspace::testing_stub();
        let id = MessageId::mint();
        ws.park_peer_prompt(&slot(None), &sender(), wrapped(&id, "are you up?"));

        ws.expire_parked_for_slot(&slot(None), PeerFailureReason::TargetConnectionFailed);

        let mut echo = None;
        while let Ok(frame) = rx.try_recv() {
            if let SessionUpdate::PeerEnvelopeAppended { key, wrapped, .. } =
                std::sync::Arc::unwrap_or_clone(frame.update)
            {
                echo = Some((key, wrapped));
            }
        }
        let (key, notice) = echo.expect("the notice is painted for the sender");
        assert_eq!(key, sender(), "the notice lands on the sender, not the target");
        assert!(
            matches!(notice.kind, WrappedKind::DeliveryFailureNotice),
            "and it is the delivery failure, not a peer message",
        );
        assert_eq!(
            notice.sender_name, "parked-proj",
            "the notice names the seat that never took the message",
        );
        assert!(
            ws.take_parked_for_slot(&slot(None)).peer.is_empty(),
            "and the bucket is gone, so a later connect cannot deliver it",
        );
    }
}