feat(agent): implement agent execution engine and turn handling with background processing

apps/infrastructure/src/agent/mod.rs

@@ -0,0 +1,6 @@
+//! Agent execution engine — orchestrates LLM streaming, system prompt assembly,
+//! tool execution, and turn event emitting on background threads.
+
+pub mod runner;
+
+pub use runner::{spawn_agent_turn, AgentTurnParams};

apps/infrastructure/src/agent/runner.rs

@@ -0,0 +1,216 @@
+//! Agent turn engine runner — handles LLM API calls, system prompt assembly,
+//! and tool execution loops on background threads.
+
+use std::collections::VecDeque;
+use std::path::PathBuf;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex};
+
+use tracing::{debug, info, warn};
+
+use zesdex_domain::core::tool_call::sanitize_tool_arguments;
+use zesdex_domain::core::ChatMessage;
+use crate::llm::provider::LlmClient;
+use crate::tools::{all_tools, tool_defs, ToolCtx};
+use crate::TurnEvent;
+
+/// Owned parameters required to spawn and execute an agent turn on a background thread.
+pub struct AgentTurnParams {
+    pub messages: Vec<ChatMessage>,
+    pub session_dir: PathBuf,
+    pub workspace_roots: Vec<PathBuf>,
+    pub turn_events: Arc<Mutex<VecDeque<TurnEvent>>>,
+    pub in_flight: Arc<AtomicBool>,
+    pub abort: Arc<AtomicBool>,
+    pub api_key: String,
+    pub model: String,
+    pub api_base: Option<String>,
+}
+
+/// Spawns an agent turn on a background OS thread.
+#[tracing::instrument(skip(params))]
+pub fn spawn_agent_turn(mut params: AgentTurnParams) {
+    info!(
+        "spawning agent turn with {} messages (model: {})",
+        params.messages.len(),
+        params.model
+    );
+
+    std::thread::spawn(move || {
+        run_turn(&mut params);
+    });
+}
+
+/// The core agent turn loop — LLM call → tool execution → repeat.
+#[tracing::instrument(skip(params))]
+fn run_turn(params: &mut AgentTurnParams) {
+    let client = LlmClient::new(
+        params.api_key.clone(),
+        params.model.clone(),
+        params.api_base.clone(),
+    );
+
+    let tools = all_tools();
+    let defs = tool_defs(&tools);
+
+    let mut sys_prompt = "You are Zesdex, an AI coding assistant. You have access to various tools via native function calling to help the user. \
+         For complex problems, use `seq_think` to reason step-by-step. \
+         For any non-trivial tasks, you MUST prioritize creating a structured plan (using `plan_enter`) and a list of TODOs (using `todowrite`) BEFORE executing any other tools or modifying files. \
+         For large multi-step operations or delegating tasks, you MUST prioritize using `workflow_run` (to run a yaml workflow script) or `hive_mind` (to orchestrate multiple agents) to complete the task efficiently. \
+         Use other tools when necessary. If a tool returns an error, fix the issue before retrying. \
+         Respond conversationally, concisely, and helpfully.".to_string();
+
+    if let Some(root) = params.workspace_roots.first() {
+        let tree = crate::utils::build_workspace_tree(root, 800);
+        let rich_ctx = crate::utils::build_rich_context(root);
+
+        sys_prompt.push_str("\n\nWorkspace structure:\n```\n");
+        sys_prompt.push_str(&tree);
+        sys_prompt.push_str("\n```\n\n");
+
+        sys_prompt.push_str(&rich_ctx);
+    }
+
+    let sys_msg = ChatMessage::system(sys_prompt);
+
+    let tool_ctx = ToolCtx::builder()
+        .session_dir(params.session_dir.clone())
+        .workspaces(params.workspace_roots.clone())
+        .turn_events(Arc::clone(&params.turn_events))
+        .build();
+
+    for iteration in 0..50 {
+        if params.abort.load(Ordering::SeqCst) {
+            params.abort.store(false, Ordering::SeqCst);
+            push_event(
+                &params.turn_events,
+                TurnEvent::SystemNote {
+                    kind: "info".into(),
+                    message: "Turn aborted by user".into(),
+                },
+            );
+            break;
+        }
+
+        debug!("agent turn iteration {iteration}");
+
+        // Stream the LLM response
+        push_event(&params.turn_events, TurnEvent::StreamStart);
+
+        let mut req_messages = params.messages.clone();
+        req_messages.insert(0, sys_msg.clone());
+
+        let result = client.chat_with_tools_streaming(
+            &req_messages,
+            Some(defs.clone()),
+            Some(0.7),
+            Some(4096),
+            |event| {
+                if params.abort.load(Ordering::SeqCst) {
+                    return false;
+                }
+                match event {
+                    zesdex_domain::core::StreamEvent::Token(s) => {
+                        push_event(&params.turn_events, TurnEvent::StreamToken(s.clone()));
+                    }
+                    zesdex_domain::core::StreamEvent::Reasoning(s) => {
+                        push_event(&params.turn_events, TurnEvent::StreamReasoning(s.clone()));
+                    }
+                    _ => {}
+                }
+                true
+            },
+            Some(&params.abort),
+        );
+
+        match result {
+            Ok((assistant_msg, usage)) => {
+                let content = assistant_msg.content.clone().unwrap_or_default();
+                let tool_calls = assistant_msg.tool_calls.clone().unwrap_or_default();
+
+                push_event(
+                    &params.turn_events,
+                    TurnEvent::StreamDone(assistant_msg.clone()),
+                );
+
+                if let Some((tokens_in, tokens_out)) = usage {
+                    push_event(
+                        &params.turn_events,
+                        TurnEvent::Usage {
+                            tokens_in,
+                            tokens_out,
+                        },
+                    );
+                }
+
+                if tool_calls.is_empty() {
+                    params.messages.push(ChatMessage::assistant(Some(content)));
+                    break;
+                }
+
+                params.messages.push(assistant_msg);
+
+                for tc in &tool_calls {
+                    let name = &tc.function.name;
+                    let args = sanitize_tool_arguments(&tc.function.arguments);
+
+                    debug!("executing tool: {name}");
+
+                    let output = if let Some(tool) = tools.iter().find(|t| t.name() == name) {
+                        match tool.run(&tool_ctx, &args) {
+                            Ok(o) => o,
+                            Err(e) => format!("Error: {e}"),
+                        }
+                    } else {
+                        format!("Unknown tool: {name}")
+                    };
+
+                    let is_error = output.starts_with("Error:");
+
+                    push_event(
+                        &params.turn_events,
+                        TurnEvent::ToolResult {
+                            tool_call_id: tc.id.clone(),
+                            tool_name: name.clone(),
+                            output: output.clone(),
+                            is_error,
+                            path: None,
+                        },
+                    );
+
+                    params
+                        .messages
+                        .push(ChatMessage::tool(tc.id.clone(), output.clone()));
+                }
+            }
+            Err(e) => {
+                warn!("LLM call failed: {e}");
+                push_event(
+                    &params.turn_events,
+                    TurnEvent::Error(format!("LLM error: {e}")),
+                );
+                break;
+            }
+        }
+    }
+
+    // Propagate accumulated messages back to caller so the next turn starts
+    // with full history (assistant replies + tool results).
+    push_event(
+        &params.turn_events,
+        TurnEvent::Compacted(params.messages.clone()),
+    );
+    push_event(&params.turn_events, TurnEvent::Done);
+    mark_done(&params.in_flight);
+}
+
+fn push_event(queue: &Arc<Mutex<VecDeque<TurnEvent>>>, event: TurnEvent) {
+    if let Ok(mut q) = queue.lock() {
+        q.push_back(event);
+    }
+}
+
+/// Mark the turn as done using lock-free atomic store.
+fn mark_done(flag: &Arc<AtomicBool>) {
+    flag.store(false, Ordering::SeqCst);
+}

apps/infrastructure/src/ipc/client.rs

@@ -12,7 +12,7 @@ pub struct IpcClient {
 impl IpcClient {
     pub fn connect_unix(path: &str) -> anyhow::Result<Self> {
         let stream = UnixStream::connect(path)?;
-        let conn = crate::ipc::conn::Connection::new(stream);
+        let conn = crate::ipc::conn::Connection::new(stream)?;
         Ok(Self {
             conn: Mutex::new(conn),
         })
@@ -22,7 +22,7 @@ impl IpcClient {
         let mut guard = self
             .conn
             .lock()
-            .expect("IpcClient mutex poisoned");
+            .map_err(|e| anyhow::anyhow!("IpcClient mutex poisoned: {e}"))?;
         guard.send(msg)
     }
 
@@ -30,7 +30,7 @@ impl IpcClient {
         let mut guard = self
             .conn
             .lock()
-            .expect("IpcClient mutex poisoned");
+            .map_err(|e| anyhow::anyhow!("IpcClient mutex poisoned: {e}"))?;
         guard.receive()
     }
 }

apps/infrastructure/src/ipc/conn.rs

@@ -1,6 +1,7 @@
 //! Connection wrapper around a Unix socket stream,
 //! pairing a buffered reader with a raw writer.
 
+use anyhow::Context;
 use std::io::BufReader;
 use std::os::unix::net::UnixStream;
 
@@ -11,14 +12,14 @@ pub struct Connection {
 }
 
 impl Connection {
-    pub fn new(stream: UnixStream) -> Self {
+    pub fn new(stream: UnixStream) -> anyhow::Result<Self> {
         let reader = BufReader::new(
             stream
                 .try_clone()
-                .expect("UnixStream::try_clone should never fail on Linux"),
+                .context("failed to clone Unix stream for IPC reader")?,
         );
         let writer = stream;
-        Self { reader, writer }
+        Ok(Self { reader, writer })
     }
 
     pub fn send<T: serde::Serialize>(&mut self, msg: &T) -> anyhow::Result<()> {

apps/infrastructure/src/ipc/server.rs

@@ -21,6 +21,6 @@ impl IpcServer {
 
     pub fn accept(&self) -> anyhow::Result<crate::ipc::conn::Connection> {
         let (stream, _addr) = self.listener.accept()?;
-        Ok(crate::ipc::conn::Connection::new(stream))
+        crate::ipc::conn::Connection::new(stream)
     }
 }

apps/infrastructure/src/lib.rs

@@ -26,6 +26,7 @@
 //! └── middleware/          — Axum HTTP middleware (auth, cors, rate-limit)
 //! ```
 
+pub mod agent;
 pub mod auth;
 pub mod bgbash;
 pub mod guard;

apps/infrastructure/src/llm/provider.rs

@@ -324,7 +324,12 @@ impl LlmClient {
                         name,
                         arguments_delta,
                     } => {
-                        while self.tool_calls.len() <= *index {
+                        // Cap the index to prevent memory exhaustion from
+                        // maliciously large indices.
+                        const MAX_TOOL_CALLS: usize = 64;
+                        let index = usize::min(*index, MAX_TOOL_CALLS.saturating_sub(1));
+
+                        while self.tool_calls.len() <= index {
                             self.tool_calls.push(zesdex_domain::core::ToolCall {
                                 id: String::new(),
                                 type_: "function".to_string(),
@@ -334,8 +339,8 @@ impl LlmClient {
                                 },
                             });
                         }
-                        
-                        let tc = &mut self.tool_calls[*index];
+
+                        let tc = &mut self.tool_calls[index];
                         
                         if let Some(ref id_val) = id {
                             tc.id = id_val.clone();

apps/infrastructure/src/persistence/iam/session_lock_repo.rs

@@ -75,17 +75,36 @@ impl SessionLockRepository for FileSystemSessionLockRepository {
             Ok(p) => p,
             Err(_) => return false,
         };
+
+        // Resolve our own executable path once.
+        let self_exe = match std::fs::read_link("/proc/self/exe") {
+            Ok(exe) => exe,
+            Err(_) => return false,
+        };
+
+        let proc_exe = std::path::PathBuf::from(format!("/proc/{pid}/exe"));
+
+        // Phase 1: read /proc/<pid>/exe and compare with self_exe.
+        let target = match std::fs::read_link(&proc_exe) {
+            Ok(t) => t,
+            Err(_) => return false,
+        };
+        if target != self_exe {
+            return false;
+        }
+
+        // Phase 2: verify the process is still alive.
+        // SAFETY: `libc::kill(pid, 0)` does not send a signal; it only checks
+        // whether the process exists and the caller has permission to signal it.
         if unsafe { libc::kill(pid_signed, 0) != 0 } {
             return false;
         }
-        let proc_exe = std::path::PathBuf::from(format!("/proc/{pid}/exe"));
-        if let Ok(target) = std::fs::read_link(&proc_exe) {
-            if let Ok(exe) = std::env::current_exe() {
-                if target != exe {
-                    return false;
-                }
-            }
+
+        // Phase 3: re-check /proc/<pid>/exe to detect PID reuse between
+        // Phase 1 and Phase 2.
+        match std::fs::read_link(&proc_exe) {
+            Ok(recheck) if recheck == self_exe => true,
+            _ => false,
         }
-        true
     }
 }