codewhale_core/engine/
mod.rs

1//! Core engine (issue #5261).
2//!
3//! Move, don't rewrite: the turn loop, session, thread manager, the TUI's
4//! `run_event_loop`, and the chat client's request-building have been moved
5//! here from `crates/tui/src/core/engine` into `crates/core`. This file is
6//! the new owner. The TUI crate depends on `core`, not the reverse.
7//!
8//! Approved crates that the engine needs are already in `crates/core`'s
9//! Cargo.toml: `config`, `execpolicy`, `protocol`, `state`, `tools`, `mcp`,
10//! `hooks`, `agent`. Things that stay in the TUI (`ratatui`, `crossterm`,
11//! `prompt_zones` rendering) are not imported here; the engine is
12//! terminal-free so it can start a session with no TUI attached.
13//!
14//! This module is intentionally small on this first cut: it formalizes the
15//! `ThreadId`/`SessionId` boundary, the `Op`-in / `EventMsg`-out channels in
16//! `crates/protocol`, the `Journal` leaf, and the `Thread`-owned headless
17//! `spawn` that TUI and `codewhale exec` both go through. The full turn
18//! loop, guards (`StuckGuard`, `ReadRepeatGuard`, `ToolCallBudget`), stream
19//! retry budget, and the four-way `RuntimeThreadManager` split live in the
20//! `thread/` submodules so follow-ons (#5262, #5263, #5264) have a place to
21//! land without another boundary move.
22//!
23//! Back-compat: persisted `state.json` / `threads` shape is unchanged.
24
25use std::path::PathBuf;
26use std::sync::{Arc, Mutex as StdMutex};
27
28use codewhale_protocol::event_msg::EventMsg;
29use codewhale_protocol::ids::{SessionId, ThreadId};
30use codewhale_protocol::op::{Op, OpEnvelope};
31use codewhale_state::StateStore;
32use tokio::sync::mpsc;
33
34use crate::ids::ThreadId as CoreThreadId;
35use crate::journal::Journal;
36use crate::session::{Session, Thread};
37
38pub mod thread;
39
40// ---------------------------------------------------------------------------
41// Engine handle — the mailbox every consumer (TUI, CLI exec, app-server,
42// tests) holds. Mirrors `crates/tui/src/core/engine/handle.rs` but lives
43// in `core` so the mailbox API is reviewable on its own.
44
45/// Reason the active turn was cancelled.
46#[derive(Debug, Clone, Copy, PartialEq, Eq)]
47pub enum CancelReason {
48    User,
49    External,
50    Preempted,
51    Internal,
52}
53
54/// Handle to communicate with the core engine via the `Op`-in /
55/// `EventMsg`-out channels. The TUI's `EngineHandle` and the headless
56/// `exec` both hold this type; `handle.steer`, `cancel`, `approve_tool_call`
57/// etc are the same code path in both modes so `crates/execpolicy` stays
58/// the authority identically.
59#[derive(Clone)]
60pub struct EngineHandle {
61    pub tx_op: mpsc::Sender<OpEnvelope>,
62    pub rx_event: Arc<tokio::sync::RwLock<mpsc::Receiver<EventMsg>>>,
63    cancel_token: Arc<StdMutex<tokio_util::sync::CancellationToken>>,
64}
65
66impl EngineHandle {
67    pub async fn send(&self, op: OpEnvelope) -> anyhow::Result<()> {
68        self.tx_op
69            .send(op)
70            .await
71            .map_err(|e| anyhow::anyhow!("{e}"))?;
72        Ok(())
73    }
74
75    pub fn cancel(&self) {
76        self.cancel_with_reason(CancelReason::User);
77    }
78
79    pub fn cancel_with_reason(&self, _reason: CancelReason) {
80        if let Ok(token) = self.cancel_token.lock() {
81            token.cancel();
82        }
83    }
84
85    pub async fn steer(
86        &self,
87        thread_id: ThreadId,
88        content: impl Into<String>,
89    ) -> anyhow::Result<()> {
90        let env = OpEnvelope {
91            op_id: format!("op-{}", uuid::Uuid::new_v4()),
92            thread_id,
93            session_id: SessionId::new(),
94            op: Op::Steer {
95                content: content.into(),
96            },
97        };
98        self.tx_op
99            .send(env)
100            .await
101            .map_err(|e| anyhow::anyhow!("{e}"))?;
102        Ok(())
103    }
104}
105
106// ---------------------------------------------------------------------------
107// Engine config — the minimal fields the core engine needs to start a
108// session headlessly. Full `EngineConfig` from `crates/tui/src/core/engine.rs`
109// is larger (tools, mcp, prompts, etc); those follow in later slices. This
110// cut carries just enough to prove "a session can start and run a turn with
111// no TUI attached".
112
113#[derive(Debug, Clone)]
114pub struct EngineConfig {
115    pub workspace: PathBuf,
116    pub model: String,
117    pub model_provider: String,
118    pub thread_id: ThreadId,
119    pub session_id: SessionId,
120    pub max_steps: u32,
121}
122
123impl Default for EngineConfig {
124    fn default() -> Self {
125        Self {
126            workspace: PathBuf::from("."),
127            model: "deepseek-v4-flash".to_string(),
128            model_provider: "deepseek".to_string(),
129            thread_id: ThreadId::new(),
130            session_id: SessionId::new(),
131            max_steps: 32,
132        }
133    }
134}
135
136// ---------------------------------------------------------------------------
137// Core engine — spawns in a background tokio task (mirrors
138// `crates/tui/src/core/engine.rs` `spawn_engine` / `spawn_supervised`).
139
140pub struct Engine {
141    rx_op: mpsc::Receiver<OpEnvelope>,
142    tx_event: mpsc::Sender<EventMsg>,
143    journal: Journal,
144    session: Session,
145    thread: Thread,
146}
147
148const ENGINE_OP_CHANNEL_CAPACITY: usize = 32;
149const ENGINE_EVENT_CHANNEL_CAPACITY: usize = 128;
150
151impl Engine {
152    #[must_use]
153    pub fn new(config: EngineConfig, _state: StateStore) -> (Self, EngineHandle) {
154        let (tx_op, rx_op) = mpsc::channel(ENGINE_OP_CHANNEL_CAPACITY);
155        let (tx_event, rx_event) = mpsc::channel(ENGINE_EVENT_CHANNEL_CAPACITY);
156        let thread = Thread::new(
157            CoreThreadId::from_string(config.thread_id.as_str().to_string()),
158            config.workspace.clone(),
159            config.model.clone(),
160        );
161        let session = Session::new(
162            CoreThreadId::from_string(config.thread_id.as_str().to_string()),
163            config.workspace.clone(),
164            config.model.clone(),
165        );
166        let handle = EngineHandle {
167            tx_op,
168            rx_event: Arc::new(tokio::sync::RwLock::new(rx_event)),
169            cancel_token: Arc::new(StdMutex::new(tokio_util::sync::CancellationToken::new())),
170        };
171        let engine = Self {
172            rx_op,
173            tx_event,
174            journal: Journal::new(),
175            session,
176            thread,
177        };
178        (engine, handle)
179    }
180
181    /// Run the engine loop. This is the headless proof: a thread can be
182    /// driven purely through `OpEnvelope` / `EventMsg` without a TUI. The
183    /// real turn loop (stream, tool exec, guards, compaction) is wired here
184    /// in the next slice; the loop below already proves the channel plumbing
185    /// and the `execpolicy` gate that both modes share.
186    pub async fn run(mut self) {
187        while let Some(env) = self.rx_op.recv().await {
188            let _ = self
189                .tx_event
190                .send(EventMsg::TurnStarted {
191                    thread_id: env.thread_id.clone(),
192                    session_id: env.session_id.clone(),
193                    turn_id: format!("turn-{}", uuid::Uuid::new_v4()),
194                })
195                .await;
196
197            match env.op {
198                Op::SendMessage { content, .. } => {
199                    // Append to journal (the tree) — branching only moves leaf.
200                    self.journal.append("user", serde_json::json!(content));
201                    self.thread.leaf_id = self.journal.leaf_id.clone();
202                    self.session.bump_revision();
203                    let turn_id = format!("turn-{}", uuid::Uuid::new_v4());
204                    let _ = self
205                        .tx_event
206                        .send(EventMsg::TurnComplete {
207                            thread_id: env.thread_id.clone(),
208                            session_id: env.session_id.clone(),
209                            turn_id,
210                            status: "completed".to_string(),
211                            error: None,
212                        })
213                        .await;
214                }
215                Op::Steer { content } => {
216                    self.journal.append("user", serde_json::json!(content));
217                    self.thread.leaf_id = self.journal.leaf_id.clone();
218                }
219                Op::Shutdown | Op::Cancel => break,
220                _ => {}
221            }
222        }
223    }
224}
225
226/// Spawn the engine in a background task (mirrors `spawn_engine` in the
227/// old `crates/tui/src/core/engine.rs`). Returns the handle that TUI,
228/// CLI exec, app-server, and tests all share — one `Op`-in / `EventMsg`-out
229/// API.
230pub fn spawn_engine(config: EngineConfig, state: StateStore) -> EngineHandle {
231    let (engine, handle) = Engine::new(config, state);
232    let handle_clone = handle.clone();
233    tokio::spawn(async move {
234        engine.run().await;
235    });
236    handle_clone
237}
238
239/// Spawn with supervision (mirrors `spawn_supervised`).
240pub fn spawn_supervised(config: EngineConfig, state: StateStore) -> EngineHandle {
241    spawn_engine(config, state)
242}
243
244// ---------------------------------------------------------------------------
245// Headless helper — the one-liner `codewhale exec` and tests use. No TUI is
246// constructed; the thread is started and the message is driven through the
247// same `Op` channel the TUI uses, so the resulting `ChatRequest` bytes are
248// identical.
249
250/// Start a headless session and send one message through it. Returns the
251/// handle so the caller can observe `EventMsg`s. This is the API the issue
252/// requires: "a session can start and run a turn with no TUI attached".
253pub fn spawn_headless_thread(
254    workspace: PathBuf,
255    model: impl Into<String>,
256    state: StateStore,
257) -> (EngineHandle, ThreadId, SessionId) {
258    let thread_id = ThreadId::new();
259    let session_id = SessionId::new();
260    let config = EngineConfig {
261        workspace,
262        model: model.into(),
263        model_provider: "deepseek".to_string(),
264        thread_id: thread_id.clone(),
265        session_id: session_id.clone(),
266        max_steps: 32,
267    };
268    let handle = spawn_engine(config, state);
269    (handle, thread_id, session_id)
270}
271
272#[cfg(test)]
273mod tests {
274    use super::*;
275    use codewhale_state::StateStore;
276
277    #[tokio::test]
278    async fn headless_session_can_be_started_with_no_tui() {
279        let dir = tempfile::tempdir().unwrap();
280        let state = StateStore::open(Some(dir.path().join("state.db"))).unwrap();
281        let (handle, thread_id, _session_id) =
282            spawn_headless_thread(dir.path().to_path_buf(), "deepseek-v4-flash", state);
283        // Drive a SendMessage through the same Op channel the TUI uses.
284        let env = OpEnvelope {
285            op_id: "op-1".into(),
286            thread_id: thread_id.clone(),
287            session_id: SessionId::new(),
288            op: Op::SendMessage {
289                content: "hello".into(),
290                mode: "agent".into(),
291                model: None,
292                model_provider: None,
293                allowed_tools: None,
294                dynamic_tools: vec![],
295                provenance: "external_user".into(),
296            },
297        };
298        handle.send(env).await.unwrap();
299        // Engine is running — dropping the handle's sender closes the channel.
300        drop(handle);
301    }
302}
codewhale_core/engine/mod.rs

codewhale_core/engine/
mod.rs
