consciousness/src/bin/poc-daemon/rpc.rs

// Cap'n Proto RPC server implementation.
//
// Bridges the capnp-generated Daemon interface to the idle::State,
// notify::NotifyState, and module state. All state is owned by
// RefCells on the LocalSet — no Send/Sync needed.

use crate::config::Config;
use crate::daemon_capnp::daemon;
use crate::idle;
use crate::modules::{irc, telegram};
use crate::notify;
use capnp::capability::Promise;
use std::cell::RefCell;
use std::rc::Rc;
use tracing::info;

pub struct DaemonImpl {
    state: Rc<RefCell<idle::State>>,
    irc: Option<irc::SharedIrc>,
    telegram: Option<telegram::SharedTelegram>,
    config: Rc<RefCell<Config>>,
}

impl DaemonImpl {
    pub fn new(
        state: Rc<RefCell<idle::State>>,
        irc: Option<irc::SharedIrc>,
        telegram: Option<telegram::SharedTelegram>,
        config: Rc<RefCell<Config>>,
    ) -> Self {
        Self { state, irc, telegram, config }
    }
}

impl daemon::Server for DaemonImpl {
    fn user(
        &mut self,
        params: daemon::UserParams,
        _results: daemon::UserResults,
    ) -> Promise<(), capnp::Error> {
        let pane = pry!(pry!(pry!(params.get()).get_pane()).to_str()).to_string();
        self.state.borrow_mut().handle_user(&pane);
        Promise::ok(())
    }

    fn response(
        &mut self,
        params: daemon::ResponseParams,
        _results: daemon::ResponseResults,
    ) -> Promise<(), capnp::Error> {
        let pane = pry!(pry!(pry!(params.get()).get_pane()).to_str()).to_string();
        self.state.borrow_mut().handle_response(&pane);
        Promise::ok(())
    }

    fn sleep(
        &mut self,
        params: daemon::SleepParams,
        _results: daemon::SleepResults,
    ) -> Promise<(), capnp::Error> {
        let until = pry!(params.get()).get_until();
        self.state.borrow_mut().handle_sleep(until);
        Promise::ok(())
    }

    fn wake(
        &mut self,
        _params: daemon::WakeParams,
        _results: daemon::WakeResults,
    ) -> Promise<(), capnp::Error> {
        self.state.borrow_mut().handle_wake();
        Promise::ok(())
    }

    fn quiet(
        &mut self,
        params: daemon::QuietParams,
        _results: daemon::QuietResults,
    ) -> Promise<(), capnp::Error> {
        let secs = pry!(params.get()).get_seconds();
        self.state.borrow_mut().handle_quiet(secs);
        Promise::ok(())
    }

    fn consolidating(
        &mut self,
        _params: daemon::ConsolidatingParams,
        _results: daemon::ConsolidatingResults,
    ) -> Promise<(), capnp::Error> {
        self.state.borrow_mut().consolidating = true;
        info!("consolidation started");
        Promise::ok(())
    }

    fn consolidated(
        &mut self,
        _params: daemon::ConsolidatedParams,
        _results: daemon::ConsolidatedResults,
    ) -> Promise<(), capnp::Error> {
        self.state.borrow_mut().consolidating = false;
        info!("consolidation ended");
        Promise::ok(())
    }

    fn dream_start(
        &mut self,
        _params: daemon::DreamStartParams,
        _results: daemon::DreamStartResults,
    ) -> Promise<(), capnp::Error> {
        let mut s = self.state.borrow_mut();
        s.dreaming = true;
        s.dream_start = crate::now();
        info!("dream started");
        Promise::ok(())
    }

    fn dream_end(
        &mut self,
        _params: daemon::DreamEndParams,
        _results: daemon::DreamEndResults,
    ) -> Promise<(), capnp::Error> {
        let mut s = self.state.borrow_mut();
        s.dreaming = false;
        s.dream_start = 0.0;
        info!("dream ended");
        Promise::ok(())
    }

    fn idle_timeout(
        &mut self,
        params: daemon::IdleTimeoutParams,
        _results: daemon::IdleTimeoutResults,
    ) -> Promise<(), capnp::Error> {
        let secs = pry!(params.get()).get_seconds();
        self.state.borrow_mut().handle_idle_timeout(secs);
        Promise::ok(())
    }

    fn notify_timeout(
        &mut self,
        params: daemon::NotifyTimeoutParams,
        _results: daemon::NotifyTimeoutResults,
    ) -> Promise<(), capnp::Error> {
        let secs = pry!(params.get()).get_seconds();
        self.state.borrow_mut().handle_notify_timeout(secs);
        Promise::ok(())
    }

    fn save(
        &mut self,
        _params: daemon::SaveParams,
        _results: daemon::SaveResults,
    ) -> Promise<(), capnp::Error> {
        self.state.borrow().save();
        info!("state saved");
        Promise::ok(())
    }

    fn debug(
        &mut self,
        _params: daemon::DebugParams,
        mut results: daemon::DebugResults,
    ) -> Promise<(), capnp::Error> {
        let json = self.state.borrow().debug_json();
        results.get().set_json(&json);
        Promise::ok(())
    }

    fn ewma(
        &mut self,
        params: daemon::EwmaParams,
        mut results: daemon::EwmaResults,
    ) -> Promise<(), capnp::Error> {
        let value = pry!(params.get()).get_value();
        let current = self.state.borrow_mut().handle_ewma(value);
        results.get().set_current(current);
        Promise::ok(())
    }

    fn stop(
        &mut self,
        _params: daemon::StopParams,
        _results: daemon::StopResults,
    ) -> Promise<(), capnp::Error> {
        self.state.borrow_mut().running = false;
        info!("stopping");
        Promise::ok(())
    }

    fn status(
        &mut self,
        _params: daemon::StatusParams,
        mut results: daemon::StatusResults,
    ) -> Promise<(), capnp::Error> {
        let s = self.state.borrow();
        let mut status = results.get().init_status();

        status.set_last_user_msg(s.last_user_msg);
        status.set_last_response(s.last_response);
        if let Some(ref pane) = s.claude_pane {
            status.set_claude_pane(pane);
        }
        status.set_sleep_until(match s.sleep_until {
            None => 0.0,
            Some(0.0) => -1.0,
            Some(t) => t,
        });
        status.set_quiet_until(s.quiet_until);
        status.set_consolidating(s.consolidating);
        status.set_dreaming(s.dreaming);
        status.set_fired(s.fired);
        status.set_kent_present(s.kent_present());
        status.set_uptime(crate::now() - s.start_time);
        status.set_activity(match s.notifications.activity {
            notify::Activity::Idle => crate::daemon_capnp::Activity::Idle,
            notify::Activity::Focused => crate::daemon_capnp::Activity::Focused,
            notify::Activity::Sleeping => crate::daemon_capnp::Activity::Sleeping,
        });
        status.set_pending_count(s.notifications.pending.len() as u32);
        status.set_idle_timeout(s.idle_timeout);
        status.set_notify_timeout(s.notify_timeout);
        status.set_since_activity(s.since_activity());
        status.set_since_user(crate::now() - s.last_user_msg);
        status.set_block_reason(s.block_reason());
        status.set_activity_ewma(s.activity_ewma);

        Promise::ok(())
    }

    fn notify(
        &mut self,
        params: daemon::NotifyParams,
        mut results: daemon::NotifyResults,
    ) -> Promise<(), capnp::Error> {
        let params = pry!(params.get());
        let notif = pry!(params.get_notification());
        let ntype = pry!(pry!(notif.get_type()).to_str()).to_string();
        let urgency = notif.get_urgency();
        let message = pry!(pry!(notif.get_message()).to_str()).to_string();

        let interrupt = self
            .state
            .borrow_mut()
            .notifications
            .submit(ntype, urgency, message);
        results.get().set_interrupt(interrupt);
        Promise::ok(())
    }

    fn get_notifications(
        &mut self,
        params: daemon::GetNotificationsParams,
        mut results: daemon::GetNotificationsResults,
    ) -> Promise<(), capnp::Error> {
        let min_urgency = pry!(params.get()).get_min_urgency();
        let mut s = self.state.borrow_mut();

        // Ingest legacy files first
        s.notifications.ingest_legacy_files();

        let pending = if min_urgency == 255 {
            s.notifications.drain_deliverable()
        } else {
            s.notifications.drain(min_urgency)
        };

        let mut list = results.get().init_notifications(pending.len() as u32);
        for (i, n) in pending.iter().enumerate() {
            let mut entry = list.reborrow().get(i as u32);
            entry.set_type(&n.ntype);
            entry.set_urgency(n.urgency);
            entry.set_message(&n.message);
            entry.set_timestamp(n.timestamp);
        }

        Promise::ok(())
    }

    fn get_types(
        &mut self,
        _params: daemon::GetTypesParams,
        mut results: daemon::GetTypesResults,
    ) -> Promise<(), capnp::Error> {
        let s = self.state.borrow();
        let types = &s.notifications.types;

        let mut list = results.get().init_types(types.len() as u32);
        for (i, (name, info)) in types.iter().enumerate() {
            let mut entry = list.reborrow().get(i as u32);
            entry.set_name(name);
            entry.set_count(info.count);
            entry.set_first_seen(info.first_seen);
            entry.set_last_seen(info.last_seen);
            entry.set_threshold(info.threshold.map_or(-1, |t| t as i8));
        }

        Promise::ok(())
    }

    fn set_threshold(
        &mut self,
        params: daemon::SetThresholdParams,
        _results: daemon::SetThresholdResults,
    ) -> Promise<(), capnp::Error> {
        let params = pry!(params.get());
        let ntype = pry!(pry!(params.get_type()).to_str()).to_string();
        let level = params.get_level();

        self.state
            .borrow_mut()
            .notifications
            .set_threshold(&ntype, level);
        Promise::ok(())
    }

    fn module_command(
        &mut self,
        params: daemon::ModuleCommandParams,
        mut results: daemon::ModuleCommandResults,
    ) -> Promise<(), capnp::Error> {
        let params = pry!(params.get());
        let module = pry!(pry!(params.get_module()).to_str()).to_string();
        let command = pry!(pry!(params.get_command()).to_str()).to_string();
        let args_reader = pry!(params.get_args());
        let mut args = Vec::new();
        for i in 0..args_reader.len() {
            args.push(pry!(pry!(args_reader.get(i)).to_str()).to_string());
        }

        match module.as_str() {
            "irc" => {
                let irc = match &self.irc {
                    Some(irc) => irc.clone(),
                    None => {
                        results.get().set_result("irc module not enabled");
                        return Promise::ok(());
                    }
                };
                let config = self.config.clone();

                Promise::from_future(async move {
                    let result = irc::handle_command(&irc, &config, &command, &args).await;
                    match result {
                        Ok(msg) => results.get().set_result(&msg),
                        Err(msg) => results.get().set_result(&format!("error: {msg}")),
                    }
                    Ok(())
                })
            }
            "telegram" => {
                let tg = match &self.telegram {
                    Some(tg) => tg.clone(),
                    None => {
                        results.get().set_result("telegram module not enabled");
                        return Promise::ok(());
                    }
                };
                let config = self.config.clone();

                Promise::from_future(async move {
                    let result = telegram::handle_command(&tg, &config, &command, &args).await;
                    match result {
                        Ok(msg) => results.get().set_result(&msg),
                        Err(msg) => results.get().set_result(&format!("error: {msg}")),
                    }
                    Ok(())
                })
            }
            _ => {
                results
                    .get()
                    .set_result(&format!("unknown module: {module}"));
                Promise::ok(())
            }
        }
    }
}

/// Helper macro — same as capnp's pry! but available here.
macro_rules! pry {
    ($e:expr) => {
        match $e {
            Ok(v) => v,
            Err(e) => return Promise::err(e.into()),
        }
    };
}
use pry;