consciousness/src/mind/mod.rs

// mind/ — Cognitive layer
//
// Mind state machine, DMN, identity, observation socket.
// Everything about how the mind operates, separate from the
// user interface (TUI, CLI) and the agent execution (tools, API).

pub mod dmn;
pub mod identity;
pub mod log;

// consciousness.rs — Mind state machine and event loop
//
// The core runtime for the consciousness binary. Mind manages turns,
// DMN state, compaction, scoring, and slash commands. The event loop
// bridges Mind (cognitive state) with App (TUI rendering).
//
// The event loop uses biased select! so priorities are deterministic:
// keyboard events > turn results > render ticks > DMN timer > UI messages.

use anyhow::Result;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{mpsc, Mutex};

use crate::agent::{Agent, TurnResult};
use crate::agent::api::ApiClient;
use crate::agent::api::types as api_types;
use crate::config::{self, AppConfig, SessionConfig};
use crate::user::{self as tui};
use crate::user::ui_channel::{self, ContextInfo, StatusInfo, StreamTarget, UiMessage};
use crate::subconscious::learn;

/// Compaction threshold — context is rebuilt when prompt tokens exceed this.
fn compaction_threshold(app: &AppConfig) -> u32 {
    (crate::agent::context::context_window() as u32) * app.compaction.hard_threshold_pct / 100
}


/// Shared state between Mind and UI. UI writes, Mind reads.
pub struct MindState {
    /// Pending user input — UI pushes, Mind consumes after turn completes.
    pub input: Vec<String>,
    /// True while a turn is in progress.
    pub turn_active: bool,
}

pub type SharedMindState = Arc<std::sync::Mutex<MindState>>;

pub fn shared_mind_state() -> SharedMindState {
    Arc::new(std::sync::Mutex::new(MindState {
        input: Vec::new(),
        turn_active: false,
    }))
}

// --- Mind: cognitive state machine ---

pub struct Mind {
    pub agent: Arc<Mutex<Agent>>,
    pub shared: SharedMindState,
    config: SessionConfig,
    ui_tx: ui_channel::UiSender,
    turn_tx: mpsc::Sender<(Result<TurnResult>, StreamTarget)>,
    // DMN state
    dmn: dmn::State,
    dmn_turns: u32,
    max_dmn_turns: u32,

    // Turn tracking
    turn_in_progress: bool,
    turn_handle: Option<tokio::task::JoinHandle<()>>,
    /// Broadcast when turn_in_progress changes. Commands can wait
    /// for turns to complete via `turn_watch_rx.wait_for(|&v| !v)`.
    turn_watch: tokio::sync::watch::Sender<bool>,

    // Per-turn tracking for DMN context
    last_user_input: Instant,
    consecutive_errors: u32,
    last_turn_had_tools: bool,

    // Subconscious orchestration
    agent_cycles: crate::subconscious::subconscious::AgentCycleState,
    /// Latest memory importance scores from full matrix scoring (manual /score).
    memory_scores: Option<learn::MemoryScore>,
    /// Whether a full matrix /score task is currently running.
    scoring_in_flight: bool,
}

impl Mind {
    pub fn new(
        agent: Arc<Mutex<Agent>>,
        shared: SharedMindState,
        config: SessionConfig,
        ui_tx: ui_channel::UiSender,
        turn_tx: mpsc::Sender<(Result<TurnResult>, StreamTarget)>,
    ) -> Self {
        let max_dmn_turns = config.app.dmn.max_turns;
        let (turn_watch, _) = tokio::sync::watch::channel(false);

        Self {
            agent,
            shared,
            config,
            ui_tx,
            turn_tx,
            dmn: if dmn::is_off() {
                dmn::State::Off
            } else {
                dmn::State::Resting { since: Instant::now() }
            },
            dmn_turns: 0,
            max_dmn_turns,
            turn_in_progress: false,
            turn_handle: None,
            turn_watch,
            last_user_input: Instant::now(),
            consecutive_errors: 0,
            last_turn_had_tools: false,
            agent_cycles: crate::subconscious::subconscious::AgentCycleState::new(""),
            memory_scores: None,
            scoring_in_flight: false,
        }
    }

    /// Subscribe to turn state changes. Use `rx.wait_for(|&v| !v).await`
    /// to wait until no turn is in progress.
    pub fn turn_watch(&self) -> tokio::sync::watch::Receiver<bool> {
        self.turn_watch.subscribe()
    }

    fn set_turn_active(&mut self, active: bool) {
        self.turn_in_progress = active;
        self.shared.lock().unwrap().turn_active = active;
        let _ = self.turn_watch.send(active);
    }

    /// How long before the next DMN tick.
    fn dmn_interval(&self) -> Duration {
        self.dmn.interval()
    }

    /// Spawn an agent turn in a background task.
    fn spawn_turn(&mut self, input: String, target: StreamTarget) {
        let agent = self.agent.clone();
        let ui_tx = self.ui_tx.clone();
        let result_tx = self.turn_tx.clone();
        self.set_turn_active(true);
        self.turn_handle = Some(tokio::spawn(async move {
            let result = Agent::turn(agent, &input, &ui_tx, target).await;
            let _ = result_tx.send((result, target)).await;
        }));
    }

    /// Submit user input — either queue it (if a turn is running) or
    /// start a new turn immediately.
    /// Check shared state for pending user input, start a turn if available.
    fn check_pending_input(&mut self) {
        if self.turn_in_progress {
            return;
        }
        let input = {
            let mut shared = self.shared.lock().unwrap();
            if shared.input.is_empty() {
                return;
            }
            shared.input.join("\n")
        };
        self.shared.lock().unwrap().input.clear();
        self.dmn_turns = 0;
        self.consecutive_errors = 0;
        self.last_user_input = Instant::now();
        self.dmn = dmn::State::Engaged;
        let _ = self.ui_tx.send(UiMessage::UserInput(input.clone()));
        self.update_status();
        self.spawn_turn(input, StreamTarget::Conversation);
    }

    /// Process a completed turn: update DMN state, check compaction,
    /// drain any queued input.
    async fn handle_turn_result(
        &mut self,
        result: Result<TurnResult>,
        target: StreamTarget,
    ) {
        self.set_turn_active(false);
        self.turn_handle = None;

        match result {
            Ok(turn_result) => {
                if turn_result.tool_errors > 0 {
                    self.consecutive_errors += turn_result.tool_errors;
                } else {
                    self.consecutive_errors = 0;
                }
                self.last_turn_had_tools = turn_result.had_tool_calls;
                self.dmn = dmn::transition(
                    &self.dmn,
                    turn_result.yield_requested,
                    turn_result.had_tool_calls,
                    target == StreamTarget::Conversation,
                );
                if turn_result.dmn_pause {
                    self.dmn = dmn::State::Paused;
                    self.dmn_turns = 0;
                    let _ = self.ui_tx.send(UiMessage::Info(
                        "DMN paused (agent requested). Ctrl+P or /wake to resume.".into(),
                    ));
                }
                if let Some(model_name) = turn_result.model_switch {
                    crate::user::event_loop::cmd_switch_model(
                        &self.agent, &model_name, &self.ui_tx,
                    ).await;
                }
            }
            Err(e) => {
                self.consecutive_errors += 1;
                let msg = match target {
                    StreamTarget::Autonomous => {
                        UiMessage::DmnAnnotation(format!("[error: {:#}]", e))
                    }
                    StreamTarget::Conversation => {
                        UiMessage::Info(format!("Error: {:#}", e))
                    }
                };
                let _ = self.ui_tx.send(msg);
                self.dmn = dmn::State::Resting {
                    since: Instant::now(),
                };
            }
        }

        self.update_status();
        self.check_compaction();
        if !self.config.no_agents {
            self.start_memory_scoring();
        }
        self.drain_pending();
    }

    /// Spawn incremental memory scoring if not already running.
    /// Non-blocking — all async work happens in the spawned task.
    fn start_memory_scoring(&self) {
        let agent = self.agent.clone();
        let ui_tx = self.ui_tx.clone();
        let cfg = crate::config::get();
        let max_age = cfg.scoring_interval_secs;
        let response_window = cfg.scoring_response_window;
        tokio::spawn(async move {
            let (context, client) = {
                let mut agent = agent.lock().await;
                if agent.agent_cycles.memory_scoring_in_flight {
                    return;
                }
                agent.agent_cycles.memory_scoring_in_flight = true;
                let _ = ui_tx.send(UiMessage::AgentUpdate(agent.agent_cycles.snapshots()));
                (agent.context.clone(), agent.client_clone())
            };

            let result = learn::score_memories_incremental(
                &context, max_age as i64, response_window, &client, &ui_tx,
            ).await;

            {
                let mut agent = agent.lock().await;
                agent.agent_cycles.memory_scoring_in_flight = false;
                if let Ok(ref scores) = result {
                    agent.agent_cycles.memory_scores = scores.clone();
                }
            }
            match result {
                Ok(_) => {
                    let agent = agent.lock().await;
                    let _ = ui_tx.send(UiMessage::AgentUpdate(agent.agent_cycles.snapshots()));
                }
                Err(e) => {
                    let _ = ui_tx.send(UiMessage::Debug(format!(
                        "[memory-scoring] failed: {:#}", e,
                    )));
                }
            }
        });
    }

    /// Check if compaction is needed after a turn.
    fn check_compaction(&self) {
        let threshold = compaction_threshold(&self.config.app);
        let agent = self.agent.clone();
        let ui_tx = self.ui_tx.clone();
        tokio::spawn(async move {
            let mut agent_guard = agent.lock().await;
            let tokens = agent_guard.last_prompt_tokens();
            if tokens > threshold {
                let _ = ui_tx.send(UiMessage::Info(format!(
                    "[compaction: {}K > {}K threshold]",
                    tokens / 1000,
                    threshold / 1000,
                )));
                agent_guard.compact();
                let _ = ui_tx.send(UiMessage::Info(
                    "[compacted — journal + recent messages]".into(),
                ));
            }
        });
    }

    /// Send any consolidated pending input as a single turn.
    fn drain_pending(&mut self) {
        self.check_pending_input();
    }

    /// Fire a DMN tick: check max turns, generate prompt, spawn turn.
    fn dmn_tick(&mut self) {
        if matches!(self.dmn, dmn::State::Paused | dmn::State::Off) {
            return;
        }

        self.dmn_turns += 1;
        if self.dmn_turns > self.max_dmn_turns {
            let _ = self.ui_tx.send(UiMessage::DmnAnnotation(format!(
                "[dmn: {} consecutive turns, resting (limit: {})]",
                self.dmn_turns - 1,
                self.max_dmn_turns,
            )));
            self.dmn = dmn::State::Resting {
                since: Instant::now(),
            };
            self.dmn_turns = 0;
            self.update_status();
            return;
        }

        let dmn_ctx = dmn::DmnContext {
            user_idle: self.last_user_input.elapsed(),
            consecutive_errors: self.consecutive_errors,
            last_turn_had_tools: self.last_turn_had_tools,
        };
        let prompt = self.dmn.prompt(&dmn_ctx);
        let _ = self.ui_tx.send(UiMessage::DmnAnnotation(format!(
            "[dmn: {} ({}/{})]",
            self.dmn.label(),
            self.dmn_turns,
            self.max_dmn_turns,
        )));
        self.update_status();
        self.spawn_turn(prompt, StreamTarget::Autonomous);
    }

    async fn cmd_new(&mut self) {
        let new_log = log::ConversationLog::new(
            self.config.session_dir.join("conversation.jsonl"),
        ).ok();
        let mut agent_guard = self.agent.lock().await;
        let shared_ctx = agent_guard.shared_context.clone();
        let shared_tools = agent_guard.active_tools.clone();
        *agent_guard = Agent::new(
            ApiClient::new(&self.config.api_base, &self.config.api_key, &self.config.model),
            self.config.system_prompt.clone(),
            self.config.context_parts.clone(),
            self.config.app.clone(),
            self.config.prompt_file.clone(),
            new_log,
            shared_ctx,
            shared_tools,
        );
        drop(agent_guard);
        self.dmn = dmn::State::Resting { since: Instant::now() };
        let _ = self.ui_tx.send(UiMessage::Info("New session started.".into()));
    }

    fn cmd_score(&mut self) {
        if self.scoring_in_flight {
            let _ = self.ui_tx.send(UiMessage::Info("(scoring already in progress)".into()));
            return;
        }
        let agent = self.agent.clone();
        let ui_tx = self.ui_tx.clone();
        self.scoring_in_flight = true;
        tokio::spawn(async move {
            let (context, client) = {
                let ag = agent.lock().await;
                (ag.context.clone(), ag.client_clone())
            };
            let result = learn::score_memories(&context, &client, &ui_tx).await;
            let ag = agent.lock().await;
            match result {
                Ok(scores) => ag.publish_context_state_with_scores(Some(&scores)),
                Err(e) => { let _ = ui_tx.send(UiMessage::Info(format!("[scoring failed: {:#}]", e))); }
            }
        });
    }

    fn cmd_dmn_query(&self) {
        let _ = self.ui_tx.send(UiMessage::Info(format!("DMN state: {:?}", self.dmn)));
        let _ = self.ui_tx.send(UiMessage::Info(format!("Next tick in: {:?}", self.dmn.interval())));
        let _ = self.ui_tx.send(UiMessage::Info(format!(
            "Consecutive DMN turns: {}/{}", self.dmn_turns, self.max_dmn_turns,
        )));
    }

    fn cmd_dmn_sleep(&mut self) {
        self.dmn = dmn::State::Resting { since: Instant::now() };
        self.dmn_turns = 0;
        let _ = self.ui_tx.send(UiMessage::Info(
            "DMN sleeping (heartbeat every 5 min). Type anything to wake.".into(),
        ));
    }

    fn cmd_dmn_wake(&mut self) {
        let was_paused = matches!(self.dmn, dmn::State::Paused | dmn::State::Off);
        if matches!(self.dmn, dmn::State::Off) {
            dmn::set_off(false);
        }
        self.dmn = dmn::State::Foraging;
        self.dmn_turns = 0;
        let msg = if was_paused { "DMN unpaused — entering foraging mode." }
            else { "DMN waking — entering foraging mode." };
        let _ = self.ui_tx.send(UiMessage::Info(msg.into()));
        self.update_status();
    }

    fn cmd_dmn_pause(&mut self) {
        self.dmn = dmn::State::Paused;
        self.dmn_turns = 0;
        let _ = self.ui_tx.send(UiMessage::Info(
            "DMN paused — no autonomous ticks. Ctrl+P or /wake to resume.".into(),
        ));
        self.update_status();
    }

    /// Interrupt: kill processes, abort current turn, clear pending queue.
    async fn interrupt(&mut self) {
        let count = {
            let agent = self.agent.lock().await;
            let mut tools = agent.active_tools.lock().unwrap();
            let count = tools.len();
            for entry in tools.drain(..) {
                entry.handle.abort();
            }
            count
        };
        if count == 0 {
            if let Some(handle) = self.turn_handle.take() {
                handle.abort();
                self.set_turn_active(false);
                self.dmn = dmn::State::Resting { since: Instant::now() };
                self.update_status();
                let _ = self.ui_tx.send(UiMessage::Activity(String::new()));
            }
        }
        self.shared.lock().unwrap().input.clear();
        let killed = count;
        if killed > 0 || self.turn_in_progress {
            let _ = self.ui_tx.send(UiMessage::Info(format!(
                "(interrupted — killed {} process(es), turn aborted)", killed,
            )));
        } else {
            let _ = self.ui_tx.send(UiMessage::Info("(interrupted)".into()));
        }
    }

    /// Cycle reasoning effort: none → low → high → none.
    /// Cycle DMN autonomy: foraging → resting → paused → off → foraging.
    fn cycle_autonomy(&mut self) {
        let (new_state, label) = match &self.dmn {
            dmn::State::Engaged | dmn::State::Working | dmn::State::Foraging => {
                (dmn::State::Resting { since: Instant::now() }, "resting")
            }
            dmn::State::Resting { .. } => (dmn::State::Paused, "PAUSED"),
            dmn::State::Paused => {
                dmn::set_off(true);
                (dmn::State::Off, "OFF (persists across restarts)")
            }
            dmn::State::Off => {
                dmn::set_off(false);
                (dmn::State::Foraging, "foraging")
            }
        };
        self.dmn = new_state;
        self.dmn_turns = 0;
        let _ = self.ui_tx.send(UiMessage::Info(format!("DMN → {} (Ctrl+P to cycle)", label)));
        self.update_status();
    }

    /// Switch to a named model from the config registry.
    fn load_context_groups(&self) -> Vec<config::ContextGroup> {
        config::get().context_groups.clone()
    }

    fn send_context_info(&self) {
        let context_groups = self.load_context_groups();
        let (instruction_files, memory_files) = identity::context_file_info(
            &self.config.prompt_file,
            self.config.app.memory_project.as_deref(),
            &context_groups,
        );
        let _ = self.ui_tx.send(UiMessage::ContextInfoUpdate(ContextInfo {
            model: self.config.model.clone(),
            available_models: self.config.app.model_names(),
            prompt_file: self.config.prompt_file.clone(),
            backend: self.config.app.backend.clone(),
            instruction_files,
            memory_files,
            system_prompt_chars: self.config.system_prompt.len(),
            context_message_chars: self.config.context_parts.iter().map(|(_, c)| c.len()).sum(),
        }));
    }

    fn update_status(&self) {
        let _ = self.ui_tx.send(UiMessage::StatusUpdate(StatusInfo {
            dmn_state: self.dmn.label().to_string(),
            dmn_turns: self.dmn_turns,
            dmn_max_turns: self.max_dmn_turns,
            prompt_tokens: 0,
            completion_tokens: 0,
            model: String::new(),
            turn_tools: 0,
            context_budget: String::new(),
        }));
    }

    pub async fn shutdown(&mut self) {
        if let Some(handle) = self.turn_handle.take() {
            handle.abort();
        }
    }

    /// Mind event loop — reacts to user input, turn results, DMN ticks.
    pub async fn run(
        &mut self,
        mut input_rx: tokio::sync::mpsc::UnboundedReceiver<crate::user::event_loop::MindMessage>,
        mut turn_rx: mpsc::Receiver<(Result<TurnResult>, StreamTarget)>,
    ) {
        use crate::user::event_loop::MindMessage;
        use crate::user::HotkeyAction;

        loop {
            let timeout = self.dmn_interval();

            tokio::select! {
                biased;

                Some(msg) = input_rx.recv() => {
                    match msg {
                        MindMessage::Hotkey(action) => {
                            match action {
                                HotkeyAction::Interrupt => self.interrupt().await,
                                HotkeyAction::CycleAutonomy => self.cycle_autonomy(),
                                _ => {}
                            }
                        }
                        MindMessage::NewSession => self.cmd_new().await,
                        MindMessage::Score => self.cmd_score(),
                        MindMessage::DmnQuery => self.cmd_dmn_query(),
                        MindMessage::DmnSleep => self.cmd_dmn_sleep(),
                        MindMessage::DmnWake => self.cmd_dmn_wake(),
                        MindMessage::DmnPause => self.cmd_dmn_pause(),
                    }
                    self.check_pending_input();
                }

                Some((result, target)) = turn_rx.recv() => {
                    self.handle_turn_result(result, target).await;
                    self.check_pending_input();
                }

                _ = tokio::time::sleep(timeout), if !self.turn_in_progress => {
                    self.check_pending_input();
                    if !self.turn_in_progress {
                        self.dmn_tick();
                    }
                }
            }
        }
    }
}

// --- Startup ---

pub async fn run(cli: crate::user::CliArgs) -> Result<()> {
    let (config, _figment) = config::load_session(&cli)?;

    if config.app.debug {
        unsafe { std::env::set_var("POC_DEBUG", "1") };
    }

    // Start channel daemons
    let mut channel_supervisor = crate::thalamus::supervisor::Supervisor::new();
    channel_supervisor.load_config();
    channel_supervisor.ensure_running();

    // Initialize idle state machine
    let mut idle_state = crate::thalamus::idle::State::new();
    idle_state.load();

    // Channel status
    let (channel_tx, channel_rx) = tokio::sync::mpsc::channel::<Vec<(String, bool, u32)>>(4);
    {
        let tx = channel_tx.clone();
        tokio::spawn(async move {
            let result = crate::thalamus::channels::fetch_all_channels().await;
            let _ = tx.send(result).await;
        });
    }
    let notify_rx = crate::thalamus::channels::subscribe_all();

    // Create UI channel
    let (ui_tx, ui_rx) = ui_channel::channel();

    // Shared state
    let shared_context = ui_channel::shared_context_state();
    let shared_active_tools = ui_channel::shared_active_tools();

    // Startup info
    let _ = ui_tx.send(UiMessage::Info("consciousness v0.3 (tui)".into()));
    let _ = ui_tx.send(UiMessage::Info(format!(
        "  model: {} (available: {})", config.model, config.app.model_names().join(", "),
    )));
    let client = ApiClient::new(&config.api_base, &config.api_key, &config.model);
    let _ = ui_tx.send(UiMessage::Info(format!("  api:   {} ({})", config.api_base, client.backend_label())));
    let _ = ui_tx.send(UiMessage::Info(format!(
        "  context: {}K chars ({} config, {} memory files)",
        config.context_parts.iter().map(|(_, c)| c.len()).sum::<usize>() / 1024,
        config.config_file_count, config.memory_file_count,
    )));

    let conversation_log_path = config.session_dir.join("conversation.jsonl");
    let conversation_log = log::ConversationLog::new(conversation_log_path.clone())
        .expect("failed to create conversation log");
    let _ = ui_tx.send(UiMessage::Info(format!("  log:   {}", conversation_log.path().display())));

    let agent = Arc::new(Mutex::new(Agent::new(
        client,
        config.system_prompt.clone(),
        config.context_parts.clone(),
        config.app.clone(),
        config.prompt_file.clone(),
        Some(conversation_log),
        shared_context.clone(),
        shared_active_tools.clone(),
    )));

    // Restore conversation from log
    {
        let mut agent_guard = agent.lock().await;
        if agent_guard.restore_from_log() {
            ui_channel::replay_session_to_ui(agent_guard.entries(), &ui_tx);
            let _ = ui_tx.send(UiMessage::Info("--- restored from conversation log ---".into()));
        }
    }

    // Send initial budget to status bar
    {
        let agent_guard = agent.lock().await;
        let _ = ui_tx.send(UiMessage::StatusUpdate(StatusInfo {
            dmn_state: "resting".to_string(),
            dmn_turns: 0, dmn_max_turns: 0,
            prompt_tokens: 0, completion_tokens: 0,
            model: agent_guard.model().to_string(),
            turn_tools: 0,
            context_budget: agent_guard.budget().status_string(),
        }));
    }

    let (turn_tx, turn_rx) = mpsc::channel::<(Result<TurnResult>, StreamTarget)>(1);

    let no_agents = config.no_agents;
    let shared_mind = shared_mind_state();
    let mut mind = Mind::new(agent, shared_mind.clone(), config, ui_tx.clone(), turn_tx);
    mind.update_status();
    if !no_agents {
        mind.start_memory_scoring();
    }
    mind.send_context_info();

    // Start observation socket
    let socket_path = mind.config.session_dir.join("agent.sock");
    let (observe_input_tx, observe_input_rx) = log::input_channel();
    if !no_agents {
        log::start(socket_path, ui_tx.subscribe(), observe_input_tx);
    }

    // Mind ↔ UI channel
    let (mind_tx, mind_rx) = tokio::sync::mpsc::unbounded_channel();

    // App for TUI
    let app = tui::App::new(mind.config.model.clone(), shared_context, shared_active_tools);
    let ui_agent = mind.agent.clone();
    let turn_watch = mind.turn_watch();

    // Spawn Mind event loop
    tokio::spawn(async move {
        mind.run(mind_rx, turn_rx).await;
    });
    crate::user::event_loop::run(
        app, ui_agent, shared_mind, turn_watch, mind_tx, ui_tx, ui_rx, observe_input_rx,
        channel_tx, channel_rx, notify_rx, idle_state,
    ).await
}