M2 TUI: EngineHandle, ratatui app, markdown renderer, permission modal, session picker, snapshot tests

harness-app: - EngineHandle: non-blocking multi-turn API (prompt/abort/permission_reply/ list_sessions/session_messages/message_parts), persistent SQLite store, no auto-approve — TUI handles permission asks via real oneshot path - App refactored to wrap EngineHandle; headless run -p keeps auto-approve harness-tui: - Terminal guard (raw mode, alternate screen, panic hook, Drop restore) - Event loop: tokio::select! over crossterm events, bus events, 33ms render tick - AppState with MessageView/PartView (cached Vec<Line> field), ModalState - pulldown-cmark → ratatui markdown renderer (headings, bold, italic, code blocks, lists, blockquotes, links, manual word-wrap) - Layout: header, chat viewport, input (tui-textarea), status bar - Permission modal (y/a/n) wired to EngineHandle::permission_reply - Session picker (Ctrl+S) with resume - Abort (Esc), Ctrl+C×2 quit, slash commands (/new /model /agent /sessions) - 7 TestBackend snapshot tests (empty, messages, tool cards, modals)
2026-07-08 23:35:40 +02:00
parent b6e94c67c7
commit 0812f6d94c
18 changed files with 2848 additions and 79 deletions
@@ -2,18 +2,25 @@
 //! config, store, event bus, permission service, tool registry, provider registry — and
 //! exposes a small headless API (`run_prompt`) used by the `harness run -p` debug command.

+use std::collections::hash_map::DefaultHasher;
+use std::collections::HashMap;
+use std::hash::Hasher;
+use std::path::Path;
 use std::path::PathBuf;
 use std::sync::{Arc, Mutex};
 use std::time::{SystemTime, UNIX_EPOCH};

 use harness_core::config::{self, Config, ConfigError};
 use harness_core::engine::{run_session, RunConfig, StepContext};
-use harness_core::event::{EventBus, RunOutcome};
-use harness_core::permission::{spawn_auto_approve, PermissionService};
+use harness_core::event::{AppEvent, EventBus, RunOutcome};
+use harness_core::permission::{PermissionReply, PermissionService};
 use harness_core::store::{Store, StoreError};
 use harness_core::tool::ToolRegistry;
-use harness_core::types::{Message, ModelRef, Part, PartBody, PartId, Session, SessionId};
+use harness_core::types::{
+    Message, MessageId, ModelRef, Part, PartBody, PartId, Session, SessionId,
+};
 use harness_providers::{AnthropicProvider, ProviderRegistry};
+use tokio::task::JoinHandle;
 use tokio_util::sync::CancellationToken;

 /// Placeholder until M4's markdown agent registry lands (`assets/agents/orchestrator.md`).
@@ -34,6 +41,8 @@ pub enum AppError {
    InvalidModelRef(String),
    #[error("no provider registered for {0:?} (missing API key?)")]
    UnknownProvider(String),
+    #[error("session {0} already has an active run")]
+    SessionRunning(SessionId),
 }

 fn now_ms() -> i64 {
@@ -43,26 +52,64 @@ fn now_ms() -> i64 {
        .as_millis() as i64
 }

-pub struct App {
-    pub config: Config,
-    pub store: Store,
-    pub bus: EventBus,
-    pub permissions: Arc<PermissionService>,
-    pub tools: ToolRegistry,
-    pub providers: ProviderRegistry,
-    pub cwd: PathBuf,
-    data_dir: PathBuf,
+fn db_path(cwd: &Path) -> PathBuf {
+    let base = dirs::data_dir()
+        .unwrap_or_else(std::env::temp_dir)
+        .join("ai-harness")
+        .join("db");
+    let cwd_str = cwd.to_string_lossy();
+    let sanitized: String = cwd_str
+        .chars()
+        .map(|c| if c.is_alphanumeric() { c } else { '_' })
+        .collect();
+    let mut hasher = DefaultHasher::new();
+    hasher.write(cwd_str.as_bytes());
+    let hash = hasher.finish();
+    let hash_hex = format!("{:016x}", hash);
+    let slug = format!("{}_{}", sanitized, &hash_hex[..8]);
+    base.join(format!("{}.sqlite", slug))
 }

-impl App {
-    /// In-memory store, auto-approve permission frontend — the M1 headless configuration.
-    /// A persistent SQLite-backed store and a real TUI permission modal land in M2.
+/// Non-blocking, multi-turn engine API used by the TUI.
+#[derive(Clone)]
+pub struct EngineHandle {
+    inner: Arc<EngineInner>,
+}
+
+struct EngineInner {
+    config: Config,
+    store: Store,
+    bus: EventBus,
+    permissions: Arc<PermissionService>,
+    tools: ToolRegistry,
+    providers: ProviderRegistry,
+    cwd: PathBuf,
+    data_dir: PathBuf,
+    runs: Mutex<HashMap<SessionId, RunHandle>>,
+}
+
+struct RunHandle {
+    cancel: CancellationToken,
+}
+
+impl EngineHandle {
+    /// Persistent SQLite-backed store. Used by the TUI and the default headless `App`.
    pub fn init(cwd: PathBuf) -> Result<Self, AppError> {
+        let path = db_path(&cwd);
+        let store = Store::open(&path)?;
+        Self::new(cwd, store)
+    }
+
+    /// In-memory store — useful for tests and ephemeral sessions.
+    pub fn init_in_memory(cwd: PathBuf) -> Result<Self, AppError> {
+        let store = Store::open_in_memory()?;
+        Self::new(cwd, store)
+    }
+
+    fn new(cwd: PathBuf, store: Store) -> Result<Self, AppError> {
        let config = config::load(&cwd)?;
        let bus = EventBus::new();
-        let store = Store::open_in_memory()?;
        let permissions = Arc::new(PermissionService::new(bus.clone()));
-        spawn_auto_approve(bus.clone(), permissions.clone());

        let mut tools = ToolRegistry::new();
        harness_tools::register_builtins(&mut tools);
@@ -82,64 +129,106 @@ impl App {
            .join("tool-output");

        Ok(Self {
-            config,
-            store,
-            bus,
-            permissions,
-            tools,
-            providers,
-            cwd,
-            data_dir,
+            inner: Arc::new(EngineInner {
+                config,
+                store,
+                bus,
+                permissions,
+                tools,
+                providers,
+                cwd,
+                data_dir,
+                runs: Mutex::new(HashMap::new()),
+            }),
        })
    }

-    /// Runs a single headless turn: creates a root session, appends `prompt` as the user
-    /// message, and drives the engine loop to completion. Returns the outcome plus the
-    /// session id so the caller can fetch the transcript via `final_text`.
-    pub async fn run_prompt(
+    pub fn bus(&self) -> EventBus {
+        self.inner.bus.clone()
+    }
+
+    pub fn permissions(&self) -> Arc<PermissionService> {
+        self.inner.permissions.clone()
+    }
+
+    pub fn config(&self) -> Config {
+        self.inner.config.clone()
+    }
+
+    pub async fn new_session(&self, agent: &str, model_ref: &str) -> Result<SessionId, AppError> {
+        let (provider_id, model_id) = model_ref
+            .split_once('/')
+            .ok_or_else(|| AppError::InvalidModelRef(model_ref.to_string()))?;
+        let model = ModelRef::new(provider_id, model_id);
+        let now = now_ms();
+        let session = Session::new_root(agent, model, now);
+        let session_id = session.id.clone();
+        self.inner.store.upsert_session(session.clone()).await?;
+        self.inner.bus.publish(AppEvent::SessionCreated { session });
+        Ok(session_id)
+    }
+
+    pub async fn prompt(
        &self,
-        prompt: String,
+        session_id: SessionId,
+        text: String,
        model_ref: &str,
-    ) -> Result<(RunOutcome, SessionId), AppError> {
+    ) -> Result<(), AppError> {
        let (provider_id, model_id) = model_ref
            .split_once('/')
            .ok_or_else(|| AppError::InvalidModelRef(model_ref.to_string()))?;
        let provider = self
+            .inner
            .providers
            .get(provider_id)
            .ok_or_else(|| AppError::UnknownProvider(provider_id.to_string()))?;
        let model = ModelRef::new(provider_id, model_id);

-        let now = now_ms();
-        let session = Session::new_root("orchestrator", model.clone(), now);
-        let session_id = session.id.clone();
-        self.store.upsert_session(session).await?;
+        {
+            let runs = self.inner.runs.lock().unwrap();
+            if runs.contains_key(&session_id) {
+                return Err(AppError::SessionRunning(session_id));
+            }
+        }

+        let now = now_ms();
        let user_message = Message::new_user(session_id.clone(), now);
-        self.store.upsert_message(user_message.clone()).await?;
-        self.store
-            .upsert_part(Part {
-                id: PartId::new(),
-                message_id: user_message.id,
-                session_id: session_id.clone(),
-                idx: 0,
-                body: PartBody::Text {
-                    text: prompt,
-                    synthetic: false,
-                },
-            })
+        self.inner
+            .store
+            .upsert_message(user_message.clone())
            .await?;
+        // The store actor does not emit events, so publish the user message and its part
+        // ourselves — otherwise the TUI (which builds its live transcript purely from bus
+        // events) never shows the prompt the user just typed until the session is reloaded.
+        self.inner.bus.publish(AppEvent::MessageCreated {
+            message: user_message.clone(),
+        });
+
+        let user_part = Part {
+            id: PartId::new(),
+            message_id: user_message.id.clone(),
+            session_id: session_id.clone(),
+            idx: 0,
+            body: PartBody::Text {
+                text,
+                synthetic: false,
+            },
+        };
+        self.inner.store.upsert_part(user_part.clone()).await?;
+        self.inner
+            .bus
+            .publish(AppEvent::PartUpdated { part: user_part });

        let ctx = StepContext {
-            store: self.store.clone(),
-            bus: self.bus.clone(),
-            tools: self.tools.clone(),
-            permissions: self.permissions.clone(),
-            static_rules: self.config.permissions.clone(),
+            store: self.inner.store.clone(),
+            bus: self.inner.bus.clone(),
+            tools: self.inner.tools.clone(),
+            permissions: self.inner.permissions.clone(),
+            static_rules: self.inner.config.permissions.clone(),
            extra_rules: Arc::new(Mutex::new(Vec::new())),
            session_id: session_id.clone(),
-            cwd: self.cwd.clone(),
-            data_dir: self.data_dir.join(session_id.to_string()),
+            cwd: self.inner.cwd.clone(),
+            data_dir: self.inner.data_dir.join(session_id.to_string()),
            cancel: CancellationToken::new(),
            now,
        };
@@ -149,21 +238,79 @@ impl App {
            model,
            temperature: None,
            max_steps: DEFAULT_MAX_STEPS,
-            instructions: self.config.instructions.clone(),
+            instructions: self.inner.config.instructions.clone(),
        };

-        let outcome = run_session(provider, ctx, &run_config, now_ms).await;
-        Ok((outcome, session_id))
+        // Reserve the run slot *before* spawning. If we inserted after spawning, a run that
+        // finished quickly could remove its (not-yet-inserted) entry first, and our later
+        // insert would then strand the session as permanently "running".
+        let cancel = ctx.cancel.clone();
+        {
+            let mut runs = self.inner.runs.lock().unwrap();
+            if runs.contains_key(&session_id) {
+                return Err(AppError::SessionRunning(session_id));
+            }
+            runs.insert(
+                session_id.clone(),
+                RunHandle {
+                    cancel: cancel.clone(),
+                },
+            );
+        }
+
+        self.inner.bus.publish(AppEvent::RunStarted {
+            session_id: session_id.clone(),
+        });
+
+        let inner = self.inner.clone();
+        let spawn_session_id = session_id.clone();
+        tokio::spawn(async move {
+            let outcome = run_session(provider, ctx, &run_config, now_ms).await;
+            inner.bus.publish(AppEvent::RunFinished {
+                session_id: spawn_session_id.clone(),
+                outcome: outcome.clone(),
+            });
+            let mut runs = inner.runs.lock().unwrap();
+            runs.remove(&spawn_session_id);
+        });
+
+        Ok(())
+    }
+
+    pub fn abort(&self, session_id: &SessionId) {
+        let runs = self.inner.runs.lock().unwrap();
+        if let Some(run) = runs.get(session_id) {
+            run.cancel.cancel();
+        }
+    }
+
+    pub fn permission_reply(&self, id: &str, reply: PermissionReply) -> bool {
+        self.inner.permissions.reply(id, reply)
+    }
+
+    pub async fn list_sessions(&self) -> Result<Vec<Session>, AppError> {
+        Ok(self.inner.store.sessions().await?)
+    }
+
+    pub async fn session_messages(&self, session_id: SessionId) -> Result<Vec<Message>, AppError> {
+        Ok(self.inner.store.messages(session_id).await?)
+    }
+
+    pub async fn message_parts(&self, message_id: MessageId) -> Result<Vec<Part>, AppError> {
+        Ok(self.inner.store.parts(message_id).await?)
+    }
+
+    pub fn is_running(&self, session_id: &SessionId) -> bool {
+        let runs = self.inner.runs.lock().unwrap();
+        runs.contains_key(session_id)
    }

-    /// Concatenates the `Text` parts of the last message in the session — the final
-    /// assistant reply for a `harness run` invocation to print.
    pub async fn final_text(&self, session_id: &SessionId) -> Result<String, AppError> {
-        let messages = self.store.messages(session_id.clone()).await?;
+        let messages = self.inner.store.messages(session_id.clone()).await?;
        let Some(last) = messages.last() else {
            return Ok(String::new());
        };
-        let parts = self.store.parts(last.id.clone()).await?;
+        let parts = self.inner.store.parts(last.id.clone()).await?;
        let text = parts
            .iter()
            .filter_map(|p| match &p.body {
@@ -176,6 +323,94 @@ impl App {
    }
 }

+pub struct App {
+    engine: EngineHandle,
+    // Keeps the auto-approve task alive for the lifetime of the App.
+    _auto_approve_handle: Option<JoinHandle<()>>,
+}
+
+impl App {
+    /// Persistent SQLite-backed store with an auto-approve permission frontend — the default
+    /// headless configuration used by `harness run -p`.
+    pub fn init(cwd: PathBuf) -> Result<Self, AppError> {
+        let engine = EngineHandle::init(cwd)?;
+        let _auto_approve_handle = Some(spawn_auto_approve_task(&engine));
+        Ok(Self {
+            engine,
+            _auto_approve_handle,
+        })
+    }
+
+    /// In-memory store with an auto-approve permission frontend — useful for tests.
+    pub fn init_in_memory(cwd: PathBuf) -> Result<Self, AppError> {
+        let engine = EngineHandle::init_in_memory(cwd)?;
+        let _auto_approve_handle = Some(spawn_auto_approve_task(&engine));
+        Ok(Self {
+            engine,
+            _auto_approve_handle,
+        })
+    }
+
+    pub fn engine(&self) -> &EngineHandle {
+        &self.engine
+    }
+
+    pub fn config(&self) -> Config {
+        self.engine.config()
+    }
+
+    /// Runs a single headless turn: creates a root session, appends `prompt` as the user
+    /// message, and drives the engine loop to completion. Returns the outcome plus the
+    /// session id so the caller can fetch the transcript via `final_text`.
+    pub async fn run_prompt(
+        &self,
+        prompt: String,
+        model_ref: &str,
+    ) -> Result<(RunOutcome, SessionId), AppError> {
+        let session_id = self.engine.new_session("orchestrator", model_ref).await?;
+        let mut rx = self.engine.bus().subscribe();
+        // Subscribe before starting the run so we cannot miss the RunFinished event
+        // even for an instant (mock) provider.
+        self.engine
+            .prompt(session_id.clone(), prompt, model_ref)
+            .await?;
+
+        while let Ok(event) = rx.recv().await {
+            if let AppEvent::RunFinished {
+                session_id: sid,
+                outcome,
+            } = event
+            {
+                if sid == session_id {
+                    return Ok((outcome, session_id));
+                }
+            }
+        }
+
+        // Bus closed unexpectedly; report a stopped run so the caller can continue.
+        Ok((RunOutcome::Stopped, session_id))
+    }
+
+    /// Concatenates the `Text` parts of the last message in the session — the final
+    /// assistant reply for a `harness run` invocation to print.
+    pub async fn final_text(&self, session_id: &SessionId) -> Result<String, AppError> {
+        self.engine.final_text(session_id).await
+    }
+}
+
+fn spawn_auto_approve_task(engine: &EngineHandle) -> JoinHandle<()> {
+    let bus = engine.bus();
+    let permissions = engine.permissions();
+    tokio::spawn(async move {
+        let mut rx = bus.subscribe();
+        while let Ok(event) = rx.recv().await {
+            if let AppEvent::PermissionAsked { request } = event {
+                permissions.reply(&request.id, PermissionReply::Once);
+            }
+        }
+    })
+}
+
 #[cfg(test)]
 mod tests {
    use super::*;
@@ -184,16 +419,16 @@ mod tests {
    async fn init_loads_defaults_with_no_providers_configured() {
        let dir = tempfile::tempdir().unwrap();
        std::env::remove_var("ANTHROPIC_API_KEY");
-        let app = App::init(dir.path().to_path_buf()).unwrap();
-        assert!(app.providers.get("anthropic").is_none());
-        assert_eq!(app.tools.all().len(), 6);
+        let app = App::init_in_memory(dir.path().to_path_buf()).unwrap();
+        assert!(app.engine.inner.providers.get("anthropic").is_none());
+        assert_eq!(app.engine.inner.tools.all().len(), 6);
    }

    #[tokio::test]
    async fn run_prompt_errors_on_unregistered_provider() {
        let dir = tempfile::tempdir().unwrap();
        std::env::remove_var("ANTHROPIC_API_KEY");
-        let app = App::init(dir.path().to_path_buf()).unwrap();
+        let app = App::init_in_memory(dir.path().to_path_buf()).unwrap();
        let err = app
            .run_prompt("hi".into(), "anthropic/claude-sonnet-4-5")
            .await
@@ -204,7 +439,7 @@ mod tests {
    #[tokio::test]
    async fn run_prompt_errors_on_malformed_model_ref() {
        let dir = tempfile::tempdir().unwrap();
-        let app = App::init(dir.path().to_path_buf()).unwrap();
+        let app = App::init_in_memory(dir.path().to_path_buf()).unwrap();
        let err = app.run_prompt("hi".into(), "no-slash").await.unwrap_err();
        assert!(matches!(err, AppError::InvalidModelRef(_)));
    }
@@ -212,8 +447,83 @@ mod tests {
    #[tokio::test]
    async fn final_text_is_empty_for_unknown_session() {
        let dir = tempfile::tempdir().unwrap();
-        let app = App::init(dir.path().to_path_buf()).unwrap();
+        let app = App::init_in_memory(dir.path().to_path_buf()).unwrap();
        let text = app.final_text(&SessionId::new()).await.unwrap();
        assert_eq!(text, "");
    }
+
+    #[tokio::test]
+    async fn engine_init_in_memory_creates_handle() {
+        let dir = tempfile::tempdir().unwrap();
+        let engine = EngineHandle::init_in_memory(dir.path().to_path_buf()).unwrap();
+        assert!(!engine.config().providers.contains_key("anthropic"));
+    }
+
+    #[tokio::test]
+    async fn engine_new_session_persists_session() {
+        let dir = tempfile::tempdir().unwrap();
+        let engine = EngineHandle::init_in_memory(dir.path().to_path_buf()).unwrap();
+        let id = engine
+            .new_session("orchestrator", "anthropic/claude")
+            .await
+            .unwrap();
+        let sessions = engine.list_sessions().await.unwrap();
+        assert_eq!(sessions.len(), 1);
+        assert_eq!(sessions[0].id, id);
+        assert_eq!(sessions[0].agent, "orchestrator");
+    }
+
+    #[tokio::test]
+    async fn engine_list_sessions_returns_persisted_sessions() {
+        let dir = tempfile::tempdir().unwrap();
+        let engine = EngineHandle::init_in_memory(dir.path().to_path_buf()).unwrap();
+        let id1 = engine.new_session("a", "anthropic/claude").await.unwrap();
+        let id2 = engine.new_session("b", "anthropic/claude").await.unwrap();
+        let sessions = engine.list_sessions().await.unwrap();
+        assert_eq!(sessions.len(), 2);
+        assert!(sessions.iter().any(|s| s.id == id1));
+        assert!(sessions.iter().any(|s| s.id == id2));
+    }
+
+    #[tokio::test]
+    async fn engine_prompt_errors_on_unregistered_provider() {
+        let dir = tempfile::tempdir().unwrap();
+        std::env::remove_var("ANTHROPIC_API_KEY");
+        let engine = EngineHandle::init_in_memory(dir.path().to_path_buf()).unwrap();
+        let id = engine
+            .new_session("orchestrator", "anthropic/claude")
+            .await
+            .unwrap();
+        let err = engine
+            .prompt(id, "hi".into(), "anthropic/claude")
+            .await
+            .unwrap_err();
+        assert!(matches!(err, AppError::UnknownProvider(_)));
+    }
+
+    #[tokio::test]
+    async fn engine_prompt_errors_on_malformed_model_ref() {
+        let dir = tempfile::tempdir().unwrap();
+        let engine = EngineHandle::init_in_memory(dir.path().to_path_buf()).unwrap();
+        let id = engine
+            .new_session("orchestrator", "anthropic/claude")
+            .await
+            .unwrap();
+        let err = engine
+            .prompt(id, "hi".into(), "no-slash")
+            .await
+            .unwrap_err();
+        assert!(matches!(err, AppError::InvalidModelRef(_)));
+    }
+
+    #[tokio::test]
+    async fn engine_is_running_false_for_inactive_session() {
+        let dir = tempfile::tempdir().unwrap();
+        let engine = EngineHandle::init_in_memory(dir.path().to_path_buf()).unwrap();
+        let id = engine
+            .new_session("orchestrator", "anthropic/claude")
+            .await
+            .unwrap();
+        assert!(!engine.is_running(&id));
+    }
 }