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

codewhale_core/engine/
mod.rs
