fix(agent,lua,storage,ui): harden agent spawn and session recursion

changelog.d/bounded-tool-batching.fixed.md

@@ -0,0 +1 @@
+Bounded model-emitted tool calls so excess work queues instead of starting every command at once. Cheap reads use a wider lane, while process-backed tools share an eight-call limit and nested agents share a four-call limit.

n00n-agent/src/agent/run.rs

@@ -19,15 +19,16 @@ use crate::cancel::{CancelMap, CancelToken, PreDispatchGate};
 use crate::mcp::McpSession;
 use crate::permissions::{PermissionAnswer, PermissionManager};
 use crate::tools::{
-    ActiveTools, Deadline, FileReadTracker, LocalTools, ToolAudience, ToolContext, ToolFilter,
-    ToolRegistry,
+    ActiveTools, Deadline, FileReadTracker, LocalTools, SessionIdentity, ToolAudience, ToolContext,
+    ToolFilter, ToolRegistry,
 };
 use crate::{
     AgentConfig, AgentError, AgentEvent, AgentInput, AgentMode, EventSender, ExtractedCommand,
     FusionContinuation, FusionPhase, FusionRequestDecision, FusionState, InterruptPoint,
     InterruptSource, ToolDoneEvent, TurnCompleteEvent,
 };
 use n00n_config::{ToolKey, ToolOutputLines};
+#[cfg(test)]
 use n00n_storage::id::SessionRef;
 
 use crate::tokenize::{
@@ -167,7 +168,7 @@ pub struct AgentParams {
     pub config: Arc<AgentConfig>,
     pub tool_output_lines: ToolOutputLines,
     pub permissions: Arc<PermissionManager>,
-    pub session_id: Option<SessionRef>,
+    pub identity: Option<SessionIdentity>,
     pub timeouts: n00n_providers::Timeouts,
     pub openai_options: OpenAiOptions,
     pub file_tracker: Arc<FileReadTracker>,
@@ -214,13 +215,14 @@ pub struct Agent<'h> {
     thinking_empty_retried: bool,
     permissions: Arc<PermissionManager>,
     opts: RequestOptions,
-    session_id: Option<SessionRef>,
+    identity: Option<SessionIdentity>,
     timeouts: n00n_providers::Timeouts,
     openai_options: OpenAiOptions,
     file_tracker: Arc<FileReadTracker>,
     prompt_slots: Arc<crate::prompt::ResolvedSlots>,
     subagent_cancels: Arc<crate::cancel::CancelMap<String>>,
     registry: Arc<crate::tools::ToolRegistry>,
+    admission_scope: Arc<str>,
     audience: ToolAudience,
     workflow: bool,
     local_tools: LocalTools,
@@ -237,6 +239,12 @@ impl<'h> Agent<'h> {
     pub fn new(params: AgentParams, run: AgentRunParams<'h>) -> Self {
         let supports_tool_examples = params.model.supports_tool_examples();
         let fusion_enabled = params.config.fusion.enabled;
+        let admission_scope = params
+            .identity
+            .as_ref()
+            .map_or_else(crate::tools::ToolAdmission::new_scope, |identity| {
+                Arc::<str>::from(identity.session_id().to_string())
+            });
         let fusion_state = if fusion_enabled {
             Some(FusionState::new())
         } else {
@@ -273,11 +281,12 @@ impl<'h> Agent<'h> {
             post_tool_empty_retried: false,
             thinking_empty_retried: false,
             opts: RequestOptions::default(),
-            session_id: params.session_id,
+            identity: params.identity,
             file_tracker: params.file_tracker,
             prompt_slots: params.prompt_slots,
             subagent_cancels: params.subagent_cancels,
             registry: params.registry,
+            admission_scope,
             audience: params.audience,
             workflow: false,
             local_tools: LocalTools::default(),
@@ -359,6 +368,23 @@ impl<'h> Agent<'h> {
         self.total_cost
     }
 
+    pub async fn run_tool(&self, id: String, name: &str, input: &Value) -> ToolDoneEvent {
+        let ctx = self.tool_context();
+        let done = tool_dispatch::run(
+            &self.registry,
+            self.mcp.as_ref(),
+            id,
+            name,
+            input,
+            &ctx,
+            tool_dispatch::Emit::Notify,
+        )
+        .await;
+        self.event_tx
+            .try_send(AgentEvent::ToolDone(Box::new(done.clone())));
+        done
+    }
+
     /// Runs the agent loop with the given input.
     ///
     /// # Errors
@@ -502,7 +528,7 @@ impl<'h> Agent<'h> {
             event_tx: &self.event_tx,
             cancel: &self.cancel,
             opts,
-            session_id: self.session_id.as_ref(),
+            session_id: self.identity.as_ref().map(SessionIdentity::session_id),
         })
         .await
     }
@@ -886,7 +912,9 @@ impl<'h> Agent<'h> {
             prompt_slots: Arc::clone(&self.prompt_slots),
             opts: self.opts.clone(),
             subagent_cancels: Arc::clone(&self.subagent_cancels),
+            identity: self.identity.clone(),
             registry: Arc::clone(&self.registry),
+            admission_scope: Arc::clone(&self.admission_scope),
             workflow: self.workflow,
             audience: self.audience,
             tool_filter: self.effective_tool_filter(),
@@ -1144,7 +1172,7 @@ impl<'h> Agent<'h> {
             &self.event_tx,
             &self.cancel,
             CompactionTrigger::Auto,
-            self.session_id.as_ref(),
+            self.identity.as_ref().map(SessionIdentity::session_id),
             &cwd,
             None,
         )
@@ -2141,7 +2169,7 @@ mod tests {
                     },
                     std::path::PathBuf::from("/tmp"),
                 )),
-                session_id: None,
+                identity: None,
                 timeouts: n00n_providers::Timeouts::default(),
                 openai_options: OpenAiOptions::default(),
                 file_tracker: FileReadTracker::fresh(),
@@ -2161,6 +2189,18 @@ mod tests {
         (agent, event_rx)
     }
 
+    #[test]
+    fn tool_context_preserves_agent_session_identity() {
+        let mut history = History::new(Vec::new());
+        let (mut agent, _event_rx) = make_agent(MockProvider::new(Vec::new()), &mut history);
+        let identity = SessionIdentity::root(SessionRef::generate());
+        agent.identity = Some(identity.clone());
+
+        let ctx = agent.tool_context();
+
+        assert_eq!(ctx.identity, Some(identity));
+    }
+
     fn make_agent_with_config(
         provider: MockProvider,
         history: &mut History,
@@ -3354,7 +3394,7 @@ mod tests {
                         },
                         std::path::PathBuf::from("/tmp"),
                     )),
-                    session_id: None,
+                    identity: None,
                     timeouts: n00n_providers::Timeouts::default(),
                     openai_options: OpenAiOptions::default(),
                     file_tracker: FileReadTracker::fresh(),

n00n-agent/src/agent/streaming.rs

@@ -8,6 +8,8 @@ use tracing::{info, warn};
 use crate::cancel::CancelToken;
 use crate::{AgentError, AgentEvent, EventSender};
 
+const PROVIDER_EVENT_QUEUE_CAPACITY: usize = 256;
+
 pub(crate) struct StreamContext<'a> {
     pub provider: &'a dyn Provider,
     pub model: &'a Model,
@@ -63,11 +65,23 @@ pub(crate) async fn stream_with_retry(
     let messages = &*messages;
     let mut retry = RetryState::new();
     loop {
-        let (ptx, prx) = flume::unbounded();
+        let (ptx, prx) = flume::bounded(PROVIDER_EVENT_QUEUE_CAPACITY);
         let forwarder = smol::spawn({
             let event_tx = ctx.event_tx.clone();
             async move { forward_provider_events(prx, &event_tx).await }
         });
+        let Ok(permit) = ctx
+            .cancel
+            .race(
+                n00n_providers::admission::ProviderAdmission::global()
+                    .acquire(ctx.model.provider.as_ref()),
+            )
+            .await
+        else {
+            drop(ptx);
+            let _ = forwarder.await;
+            return Err(AgentError::Cancelled);
+        };
         let result = futures_lite::future::race(
             ctx.provider.stream_message(
                 ctx.model,
@@ -84,6 +98,7 @@ pub(crate) async fn stream_with_retry(
             },
         )
         .await;
+        drop(permit);
         drop(ptx);
         let emitted_output = forwarder.await;
         match result {

n00n-agent/src/agent/tool_dispatch.rs

@@ -13,7 +13,7 @@ use crate::permissions::PermissionCheckContext;
 use crate::skill_policy::SKILL_POLICY_DENIED_PREFIX;
 use crate::task_set::TaskSet;
 use crate::tools::registry::{ToolInvocation, ToolRegistry, ToolSource};
-use crate::tools::{LocalToolFn, ToolContext};
+use crate::tools::{LocalToolFn, ToolAdmissionClass, ToolContext};
 use crate::{AgentError, AgentEvent, ToolDoneEvent, ToolOutput, ToolStartEvent};
 use n00n_config::ToolKey;
 
@@ -400,6 +400,16 @@ async fn run_authorized(
         return tool_done_error(id.clone(), Arc::from(name), reason);
     }
     if let Some(local) = ctx.local_tools.get(name) {
+        let class = ToolAdmissionClass::for_tool(name, None);
+        let _admission = match ctx
+            .registry
+            .admission()
+            .acquire(&ctx.admission_scope, class, &ctx.cancel)
+            .await
+        {
+            Ok(guard) => guard,
+            Err(error) => return tool_done_error(id, Arc::from(name), error.to_string()),
+        };
         return run_local_tool(local, id, name, input, ctx, emit);
     }
     let entry = registry.get(name);
@@ -502,6 +512,19 @@ async fn run_authorized(
             return tool_done_error(id.clone(), Arc::clone(&tool_id), e);
         }
 
+        let _admission = match ctx
+            .registry
+            .admission()
+            .acquire(
+                &ctx.admission_scope,
+                entry.tool.admission_class(),
+                &ctx.cancel,
+            )
+            .await
+        {
+            Ok(guard) => guard,
+            Err(error) => return tool_done_error(id, Arc::clone(&tool_id), error.to_string()),
+        };
         let result = invocation.execute(ctx).await;
 
         let elapsed = started.elapsed();
@@ -516,7 +539,8 @@ async fn run_authorized(
                 let output = match result.telemetry {
                     Some(telemetry) => output.with_telemetry(Some(telemetry)),
                     None => output,
-                };
+                }
+                .bounded(ctx.config.max_output_lines, ctx.config.max_output_bytes);
                 ToolDoneEvent {
                     id,
                     tool: tool_id,
@@ -538,7 +562,11 @@ async fn run_authorized(
                     id,
                     tool: tool_id,
                     output: ToolOutput::Plain(crate::TextOutput {
-                        text: message,
+                        text: crate::tools::truncate_output(
+                            &message,
+                            ctx.config.max_output_lines,
+                            ctx.config.max_output_bytes,
+                        ),
                         instructions: None,
                         state: None,
                         telemetry: result.telemetry,
@@ -550,8 +578,30 @@ async fn run_authorized(
             }
         }
     } else if let Some(mcp) = mcp.filter(|_| name == TOOL_SEARCH_TOOL_NAME) {
+        let _admission = match ctx
+            .registry
+            .admission()
+            .acquire(&ctx.admission_scope, ToolAdmissionClass::Cheap, &ctx.cancel)
+            .await
+        {
+            Ok(guard) => guard,
+            Err(error) => return tool_done_error(id, tool_id, error.to_string()),
+        };
         run_tool_search(mcp, id, input, ctx, emit)
     } else if mcp.is_some_and(|m| m.has_tool(&mcp_lookup)) {
+        let _admission = match ctx
+            .registry
+            .admission()
+            .acquire(
+                &ctx.admission_scope,
+                ToolAdmissionClass::Standard,
+                &ctx.cancel,
+            )
+            .await
+        {
+            Ok(guard) => guard,
+            Err(error) => return tool_done_error(id, tool_id, error.to_string()),
+        };
         execute_mcp_tool(ctx, &id, tool_id, &mcp_lookup, input, emit).await
     } else {
         let msg = format!("{UNKNOWN_TOOL_PREFIX}: {mcp_lookup}");
@@ -599,8 +649,22 @@ fn run_tool_search(
     let query = input["query"].as_str().unwrap_or_else(Default::default);
     emit_raw_start(ctx, emit, &id, &tool_id, query.to_owned(), input);
     let (output, is_error) = match mcp.search_tools(query) {
-        Ok(out) => (out, false),
-        Err(e) => (e, true),
+        Ok(out) => (
+            crate::tools::truncate_output(
+                &out,
+                ctx.config.max_output_lines,
+                ctx.config.max_output_bytes,
+            ),
+            false,
+        ),
+        Err(e) => (
+            crate::tools::truncate_output(
+                &e,
+                ctx.config.max_output_lines,
+                ctx.config.max_output_bytes,
+            ),
+            true,
+        ),
     };
     ToolDoneEvent {
         id,
@@ -623,10 +687,24 @@ fn run_local_tool(
     let tool_id: Arc<str> = Arc::from(name);
     emit_raw_start(ctx, emit, &id, &tool_id, name.to_owned(), input);
     let (output, is_error) = match local(input) {
-        Ok(output) => (output, false),
+        Ok(output) => (
+            crate::tools::truncate_output(
+                &output,
+                ctx.config.max_output_lines,
+                ctx.config.max_output_bytes,
+            ),
+            false,
+        ),
         Err(e) => {
             warn!(tool = %name, error = %e, "local tool failed");
-            (e, true)
+            (
+                crate::tools::truncate_output(
+                    &e,
+                    ctx.config.max_output_lines,
+                    ctx.config.max_output_bytes,
+                ),
+                true,
+            )
         }
     };
     ToolDoneEvent {
@@ -787,8 +865,24 @@ async fn execute_mcp_tool(
     // definition joins the next request; a denied call must not load anything.
     mcp.mark_loaded(tool_name);
     match mcp.call_tool(tool_name, input).await {
-        Ok(text) => tool_done_plain(id.to_owned(), tool_id, text),
-        Err(e) => tool_done_error(id.to_owned(), tool_id, e.to_string()),
+        Ok(text) => tool_done_plain(
+            id.to_owned(),
+            tool_id,
+            crate::tools::truncate_output(
+                &text,
+                ctx.config.max_output_lines,
+                ctx.config.max_output_bytes,
+            ),
+        ),
+        Err(e) => tool_done_error(
+            id.to_owned(),
+            tool_id,
+            crate::tools::truncate_output(
+                &e.to_string(),
+                ctx.config.max_output_lines,
+                ctx.config.max_output_bytes,
+            ),
+        ),
     }
 }