refactor: deliver messages directly to agent mailbox, remove TUI inbox routing (#84)

TUI previously buffered messages in a local Inbox and used agent_idle as a
routing decision: when agent was busy, messages were queued locally and an
interrupt signal was used to wake the agent to drain the queue. This was
TUI concept leakage into the agent runtime — agent internal state was
being directly mutated by the display layer, and interrupt was overloaded
with "cancel work" AND "new message arrived" semantics.

The worst symptom: ephemeral agents exited immediately on user input
because the yield_now idle check fired before the TUI's delayed inbox
forward could complete the IPC round-trip.

Architecture principles applied:
- Agent owns its internal state (idle/busy); external actors interact only
  via mailbox delivery + explicit interrupt
- TUI state (observable.status, is_idle) is derived solely from agent
  events, never set directly
- InterruptSignal carries only cancel/shutdown semantics

Key changes:
- Remove agent_idle field; derive via AgentViewState::is_idle() from status
- Remove Inbox struct entirely (dead after routing bypass)
- TUI calls append_user_display() + route_message() directly — no buffering
- Ephemeral idle check: drop yield_now(), drain_pending() is reliable now
- drain_pending() returns Vec<AgentInput> so Control commands are not
  silently dropped
- Merge InputFromClient into AgentInput (Interrupt variant was vestigial)
- Status transition rolls back on emit failure for consistency
- Upgrade observation-channel try_send drops from debug to warn for prod
  visibility

49/49 tests pass, zero clippy warnings, rustfmt clean.

Author: yishuiliunian

crates/loopal-agent-hub/src/agent_io.rs

@@ -48,7 +48,7 @@ pub async fn agent_io_loop(
                     }
                     let h = hub.lock().await;
                     if h.registry.event_sender().try_send(event).is_err() {
-                        tracing::debug!(agent = %agent_name, "event dropped (channel full)");
+                        tracing::warn!(agent = %agent_name, "event dropped (channel full)");
                     }
                 }
             }

crates/loopal-agent-hub/src/agent_registry/completion.rs

@@ -27,7 +27,9 @@ impl AgentRegistry {
         let pending_delivery = self.prepare_parent_delivery(name, &text);
 
         let event = AgentEvent::named(name, AgentEventPayload::Finished);
-        let _ = self.event_tx.try_send(event);
+        if self.event_tx.try_send(event).is_err() {
+            tracing::warn!(agent = %name, "Finished event dropped (channel full)");
+        }
 
         if let Some(tx) = self.completions.remove(name) {
             let _ = tx.send(Some(text));

crates/loopal-agent-hub/src/routing.rs

@@ -26,6 +26,11 @@ pub async fn route_to_agent(
         target: envelope.target.clone(),
         content_preview: envelope.content_preview().to_string(),
     });
-    let _ = observation_tx.try_send(event);
+    if observation_tx.try_send(event).is_err() {
+        tracing::warn!(
+            target = %envelope.target,
+            "observation event dropped (channel full)"
+        );
+    }
     Ok(())
 }

crates/loopal-agent-hub/src/spawn_manager.rs

@@ -130,7 +130,7 @@ pub async fn register_agent_connection(
             session_id: session_id.map(String::from),
         });
         if h.registry.event_sender().try_send(event).is_err() {
-            tracing::debug!(agent = %name, "SubAgentSpawned event dropped");
+            tracing::warn!(agent = %name, "SubAgentSpawned event dropped (channel full)");
         }
     }
     agent_id

crates/loopal-agent-server/src/hub_frontend.rs

@@ -14,18 +14,18 @@ use tracing::{debug, info};
 
 use loopal_error::{LoopalError, Result};
 use loopal_ipc::protocol::methods;
-use loopal_protocol::{AgentEvent, AgentEventPayload, Envelope, Question, UserQuestionResponse};
+use loopal_protocol::{AgentEvent, AgentEventPayload, Question, UserQuestionResponse};
 use loopal_runtime::agent_input::AgentInput;
 use loopal_runtime::frontend::traits::{AgentFrontend, EventEmitter};
 use loopal_tool_api::PermissionDecision;
 
 use crate::hub_emitter::HubEventEmitter;
-use crate::session_hub::{InputFromClient, SharedSession};
+use crate::session_hub::SharedSession;
 
 /// Frontend that multiplexes across all clients in a shared session.
 pub struct HubFrontend {
     session: tokio::sync::RwLock<Arc<SharedSession>>,
-    input_rx: tokio::sync::Mutex<tokio::sync::mpsc::Receiver<InputFromClient>>,
+    input_rx: tokio::sync::Mutex<tokio::sync::mpsc::Receiver<AgentInput>>,
     agent_name: Option<String>,
     /// Watch channel for interrupt detection in recv_input.
     interrupt_rx: tokio::sync::Mutex<tokio::sync::watch::Receiver<u64>>,
@@ -34,7 +34,7 @@ pub struct HubFrontend {
 impl HubFrontend {
     pub fn new(
         session: Arc<SharedSession>,
-        input_rx: tokio::sync::mpsc::Receiver<InputFromClient>,
+        input_rx: tokio::sync::mpsc::Receiver<AgentInput>,
         agent_name: Option<String>,
         interrupt_rx: tokio::sync::watch::Receiver<u64>,
     ) -> Self {
@@ -95,18 +95,12 @@ impl AgentFrontend for HubFrontend {
         // has already been handled by TurnCancel. Without this, changed()
         // fires immediately on the old value and exits the agent loop.
         interrupt_rx.borrow_and_update();
-        loop {
-            tokio::select! {
-                msg = rx.recv() => {
-                    match msg? {
-                        InputFromClient::Message(env) => return Some(AgentInput::Message(env)),
-                        InputFromClient::Control(cmd) => return Some(AgentInput::Control(cmd)),
-                        InputFromClient::Interrupt => continue,
-                    }
-                }
-                _ = interrupt_rx.changed() => {
-                    return None; // Interrupted — exit agent loop
-                }
+        tokio::select! {
+            msg = rx.recv() => {
+                return msg;
+            }
+            _ = interrupt_rx.changed() => {
+                return None; // Interrupted — exit agent loop
             }
         }
     }
@@ -199,14 +193,12 @@ impl AgentFrontend for HubFrontend {
         true
     }
 
-    async fn drain_pending(&self) -> Vec<Envelope> {
+    async fn drain_pending(&self) -> Vec<AgentInput> {
         let mut rx = self.input_rx.lock().await;
-        let mut envelopes = Vec::new();
+        let mut inputs = Vec::new();
         while let Ok(msg) = rx.try_recv() {
-            if let InputFromClient::Message(env) = msg {
-                envelopes.push(env);
-            }
+            inputs.push(msg);
         }
-        envelopes
+        inputs
     }
 }

crates/loopal-agent-server/src/lib.rs

@@ -48,7 +48,7 @@ pub fn ipc_frontend_for_test(
 #[doc(hidden)]
 pub fn hub_frontend_for_test(
     session: std::sync::Arc<session_hub::SharedSession>,
-    input_rx: tokio::sync::mpsc::Receiver<session_hub::InputFromClient>,
+    input_rx: tokio::sync::mpsc::Receiver<loopal_runtime::agent_input::AgentInput>,
     interrupt_rx: tokio::sync::watch::Receiver<u64>,
 ) -> std::sync::Arc<dyn loopal_runtime::frontend::traits::AgentFrontend> {
     std::sync::Arc::new(hub_frontend::HubFrontend::new(
@@ -64,5 +64,6 @@ pub fn hub_frontend_for_test(
 pub mod testing {
     pub use crate::agent_setup::build_with_frontend;
     pub use crate::params::{StartParams, build_kernel_with_provider};
-    pub use crate::session_hub::{InputFromClient, SharedSession};
+    pub use crate::session_hub::SharedSession;
+    pub use loopal_runtime::agent_input::AgentInput;
 }

crates/loopal-agent-server/src/session_forward.rs

@@ -10,8 +10,8 @@ use loopal_ipc::connection::{Connection, Incoming};
 use loopal_ipc::jsonrpc;
 use loopal_ipc::protocol::methods;
 use loopal_protocol::{ControlCommand, Envelope};
+use loopal_runtime::agent_input::AgentInput;
 
-use crate::session_hub::InputFromClient;
 use crate::session_start::SessionHandle;
 
 /// Result of forward_loop — tells dispatch_loop what happened.
@@ -74,7 +74,7 @@ pub(crate) async fn forward_loop(
                         } else if method == methods::AGENT_MESSAGE.name {
                             // Hub-injected message (e.g. sub-agent completion notification).
                             if let Ok(env) = serde_json::from_value::<Envelope>(params) {
-                                let _ = session.input_tx.send(InputFromClient::Message(env)).await;
+                                let _ = session.input_tx.send(AgentInput::Message(env)).await;
                             }
                         }
                     }
@@ -98,7 +98,7 @@ async fn route_request(
     match method {
         m if m == methods::AGENT_MESSAGE.name => match serde_json::from_value::<Envelope>(params) {
             Ok(env) => {
-                let _ = session.input_tx.send(InputFromClient::Message(env)).await;
+                let _ = session.input_tx.send(AgentInput::Message(env)).await;
                 let _ = connection
                     .respond(id, serde_json::json!({"ok": true}))
                     .await;
@@ -112,7 +112,7 @@ async fn route_request(
         m if m == methods::AGENT_CONTROL.name => {
             match serde_json::from_value::<ControlCommand>(params) {
                 Ok(cmd) => {
-                    let _ = session.input_tx.send(InputFromClient::Control(cmd)).await;
+                    let _ = session.input_tx.send(AgentInput::Control(cmd)).await;
                     let _ = connection
                         .respond(id, serde_json::json!({"ok": true}))
                         .await;

crates/loopal-agent-server/src/session_hub.rs

@@ -8,6 +8,7 @@ use tokio::sync::Mutex;
 
 use loopal_ipc::connection::Connection;
 use loopal_protocol::InterruptSignal;
+use loopal_runtime::agent_input::AgentInput;
 
 /// A connected client handle within a shared session.
 pub struct ClientHandle {
@@ -22,19 +23,12 @@ pub struct SharedSession {
     pub session_id: String,
     pub clients: Mutex<Vec<ClientHandle>>,
     /// Channel to send input into the agent loop.
-    pub input_tx: tokio::sync::mpsc::Sender<InputFromClient>,
+    pub input_tx: tokio::sync::mpsc::Sender<AgentInput>,
     /// Interrupt signal shared with the agent loop.
     pub interrupt: InterruptSignal,
     pub interrupt_tx: Arc<tokio::sync::watch::Sender<u64>>,
 }
 
-/// Input forwarded from a client connection to the agent loop.
-pub enum InputFromClient {
-    Message(loopal_protocol::Envelope),
-    Control(loopal_protocol::ControlCommand),
-    Interrupt,
-}
-
 /// Server-wide session registry.
 #[derive(Default)]
 pub struct SessionHub {
@@ -108,7 +102,7 @@ impl SessionHub {
 impl SharedSession {
     /// Create a placeholder session (for bootstrapping before session_id is known).
     pub fn placeholder(
-        input_tx: tokio::sync::mpsc::Sender<InputFromClient>,
+        input_tx: tokio::sync::mpsc::Sender<AgentInput>,
         interrupt: InterruptSignal,
         interrupt_tx: Arc<tokio::sync::watch::Sender<u64>>,
     ) -> Self {

crates/loopal-agent-server/src/session_start.rs

@@ -10,12 +10,13 @@ use loopal_config::load_config;
 use loopal_error::AgentOutput;
 use loopal_ipc::connection::Connection;
 use loopal_protocol::InterruptSignal;
+use loopal_runtime::agent_input::AgentInput;
 use loopal_runtime::agent_loop;
 
 use crate::agent_setup;
 use crate::hub_frontend::HubFrontend;
 use crate::params::StartParams;
-use crate::session_hub::{InputFromClient, SessionHub, SharedSession};
+use crate::session_hub::{SessionHub, SharedSession};
 
 /// Handle returned to the dispatch loop after starting a session.
 pub(crate) struct SessionHandle {
@@ -77,7 +78,7 @@ pub(crate) async fn start_session(
     };
 
     // Create session infrastructure
-    let (input_tx, input_rx) = tokio::sync::mpsc::channel::<InputFromClient>(16);
+    let (input_tx, input_rx) = tokio::sync::mpsc::channel::<AgentInput>(16);
     let interrupt = InterruptSignal::new();
     let (watch_tx, watch_rx) = tokio::sync::watch::channel(0u64);
     let interrupt_tx = Arc::new(watch_tx);

crates/loopal-agent-server/tests/suite.rs

@@ -6,6 +6,8 @@ mod bridge_edge_test;
 mod bridge_helpers;
 #[path = "suite/dispatch_loop_test.rs"]
 mod dispatch_loop_test;
+#[path = "suite/hub_drain_test.rs"]
+mod hub_drain_test;
 #[path = "suite/hub_frontend_edge_test.rs"]
 mod hub_frontend_edge_test;
 #[path = "suite/hub_frontend_test.rs"]