Refactor session ownership, turn cancellation, and tool execution

crates/agent_core/src/dialect.rs

@@ -11,7 +11,7 @@
 use crate::types::ToolRequest;
 use crate::ui::{AgentUi, HiddenTools, StreamProcessorTrait};
 use anyhow::Result;
-use llm::{LLMResponse, Message};
+use llm::LLMResponse;
 use std::sync::Arc;
 use tools_core::ToolRegistry;
 
@@ -65,8 +65,4 @@ pub trait ToolDialect: Send + Sync {
         registry: &ToolRegistry,
         capability: &str,
     ) -> Option<String>;
-
-    /// Whether an already stored message contains a tool invocation in this
-    /// dialect (used to normalize the history when loading a session).
-    fn message_contains_invocation(&self, message: &Message, registry: &ToolRegistry) -> bool;
 }

crates/agent_core/src/execution.rs

@@ -0,0 +1,135 @@
+//! The tool journal: every call's outcome record in request order, and the
+//! loop's own self-describing entries for calls without a concrete tool
+//! result. Successful and functionally failed tool outputs keep the codec
+//! of their tool.
+
+use crate::types::ToolExecution;
+use serde::{Deserialize, Serialize};
+use std::collections::BTreeSet;
+use tools_core::{Render, ResourcesTracker, ToolResult};
+
+pub(crate) const RUNTIME_OUTPUT_CODEC: &str = "__agent_runtime_outcome_v1";
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ExecutionState {
+    /// The call was rejected or cancelled before anything ran.
+    NotStarted,
+    /// Persisted BEFORE invoking a tool. After interruption we cannot tell
+    /// whether its effects happened, including the save/invoke crash window.
+    Started,
+    /// The invocation itself failed (as opposed to a tool reporting an error
+    /// through its own output type).
+    Failed,
+}
+
+/// The loop's own outcome record for a call. Never a success: successful
+/// tools journal their real output.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RuntimeToolOutput {
+    pub state: ExecutionState,
+    pub message: String,
+}
+
+impl RuntimeToolOutput {
+    pub fn not_started(reason: impl AsRef<str>) -> Self {
+        Self {
+            state: ExecutionState::NotStarted,
+            message: format!("Tool execution has not started. {}", reason.as_ref()),
+        }
+    }
+
+    pub fn started() -> Self {
+        Self {
+            state: ExecutionState::Started,
+            message: "Tool execution may have started, but its outcome is unknown. Verify the state before retrying any side effects.".into(),
+        }
+    }
+
+    pub fn failed(message: impl Into<String>) -> Self {
+        Self {
+            state: ExecutionState::Failed,
+            message: message.into(),
+        }
+    }
+}
+
+impl Render for RuntimeToolOutput {
+    fn status(&self) -> String {
+        match self.state {
+            ExecutionState::NotStarted => "Not started",
+            ExecutionState::Started => "Outcome unknown",
+            ExecutionState::Failed => "Error",
+        }
+        .into()
+    }
+
+    fn render(&self, _: &mut ResourcesTracker) -> String {
+        self.message.clone()
+    }
+}
+
+impl ToolResult for RuntimeToolOutput {
+    fn is_success(&self) -> bool {
+        false
+    }
+}
+
+/// The run's record of tool calls in request order, plus which entries
+/// changed since the last checkpoint.
+#[derive(Default)]
+pub struct ToolJournal {
+    entries: Vec<ToolExecution>,
+    changed: BTreeSet<String>,
+}
+
+impl ToolJournal {
+    /// Restore the journal from persisted entries; nothing counts as changed.
+    pub fn restore(entries: Vec<ToolExecution>) -> Self {
+        Self {
+            entries,
+            changed: BTreeSet::new(),
+        }
+    }
+
+    pub fn entries(&self) -> &[ToolExecution] {
+        &self.entries
+    }
+
+    pub fn find(&self, id: &str) -> Option<&ToolExecution> {
+        self.entries
+            .iter()
+            .find(|entry| entry.tool_request.id == id)
+    }
+
+    pub fn contains(&self, id: &str) -> bool {
+        self.find(id).is_some()
+    }
+
+    /// Record the entry for a call, replacing an earlier entry with the same
+    /// id in place so request order is preserved.
+    pub fn record(&mut self, execution: ToolExecution) {
+        let id = execution.tool_request.id.clone();
+        match self
+            .entries
+            .iter()
+            .position(|entry| entry.tool_request.id == id)
+        {
+            Some(index) => self.entries[index] = execution,
+            None => self.entries.push(execution),
+        }
+        self.changed.insert(id);
+    }
+
+    /// Entries recorded or updated since the last checkpoint, in journal order.
+    pub fn changed(&self) -> impl Iterator<Item = &ToolExecution> + '_ {
+        self.entries
+            .iter()
+            .filter(|entry| self.changed.contains(&entry.tool_request.id))
+    }
+
+    /// Forget the change marks after a successful checkpoint.
+    pub fn mark_checkpointed(&mut self) {
+        self.changed.clear();
+    }
+}

crates/agent_core/src/hooks.rs

@@ -8,20 +8,19 @@
 //! same dyn-Any approach `ToolContext` uses.
 
 use crate::dialect::ToolDialect;
-use crate::tree::{ConversationPath, MessageNode, NodeId};
-use crate::types::{ToolExecution, ToolRequest};
+use crate::tree::Conversation;
+use crate::types::ToolRequest;
 use anyhow::Result;
 use llm::Message;
 use std::any::Any;
-use std::collections::BTreeMap;
 use std::time::Duration;
-use tools_core::ToolRegistry;
+use tools_core::{AnyOutput, ToolRegistry};
 
 /// View of the agent state that hooks may read and act on.
 pub struct LoopCtx<'a> {
-    pub tool_executions: &'a mut Vec<ToolExecution>,
-    pub message_nodes: &'a mut BTreeMap<NodeId, MessageNode>,
-    pub active_path: &'a ConversationPath,
+    /// The conversation tree. Edits through it are part of the next
+    /// checkpoint.
+    pub conversation: &'a mut Conversation,
     /// The session this agent runs, `None` while no session is assigned yet.
     /// Lets shared hook state (built once per process) be keyed per session —
     /// same role `PromptCtx::session_id` plays for system-prompt providers.
@@ -36,13 +35,20 @@ pub struct LoopCtx<'a> {
 /// Intercepts tool requests that the application handles itself instead of
 /// dispatching them to the registry, and observes successful executions.
 pub trait ToolInterceptor: Send + Sync {
-    /// Returns `Some(result)` when the request was handled here. Intercepted
-    /// tools do not appear in the UI.
-    fn try_intercept(&self, _request: &ToolRequest, _ctx: &mut LoopCtx) -> Option<Result<bool>> {
+    /// Handles the request in the application instead of the registry and
+    /// returns the output the loop journals for it. Scope and permission
+    /// checks always precede this hook, for parallel batches as well.
+    /// Intercepted tools do not appear in the UI.
+    fn try_intercept(
+        &self,
+        _request: &ToolRequest,
+        _ctx: &mut LoopCtx,
+    ) -> Option<Result<Box<dyn AnyOutput>>> {
         None
     }
 
-    /// Invoked after any tool executed successfully (standard path included).
+    /// Invoked on the state owner after any successful tool (including
+    /// intercepted and parallel calls), with its final, possibly rewritten input.
     fn after_tool_success(&self, _request: &ToolRequest, _ctx: &mut LoopCtx) {}
 }
 
@@ -74,7 +80,9 @@ pub trait IterationHook: Send + Sync {
 
 /// Decides which tool requests of a turn may execute concurrently.
 pub trait ToolDispatchPolicy: Send + Sync {
-    /// Indices of the requests that may execute concurrently with each other.
+    /// Indices of requests that support detached services and may overlap.
+    /// Only adjacent selected requests overlap: unselected calls are ordering
+    /// barriers. Authorization and completion hooks still run on the state owner.
     fn parallel_indices(&self, requests: &[ToolRequest]) -> Vec<usize>;
 }
 

crates/agent_core/src/lib.rs

@@ -3,14 +3,15 @@
 //! Applications embed [`runtime::AgentRuntime`] and bring their own tools
 //! (via a `tools_core::ToolRegistry`), their own behavior plugins (the hook
 //! traits in [`hooks`]), their own UI adapter ([`ui::AgentUi`]), their own
-//! persistence ([`persistence::SnapshotPersistence`]), and — optionally —
+//! persistence ([`persistence::CheckpointPersistence`]), and — optionally —
 //! their own tool invocation format ([`dialect::ToolDialect`]; the built-in
 //! default is native tool calling, [`native::NativeDialect`]).
 //!
 //! Application state rides on the loop type-erased (`extensions` slots, a
 //! dyn-Any approach) — no generics infect the embedding application.
 
 pub mod dialect;
+pub mod execution;
 pub mod hooks;
 pub mod native;
 pub mod persistence;
@@ -20,9 +21,9 @@ pub mod types;
 pub mod ui;
 
 pub use dialect::ToolDialect;
-pub use persistence::{AgentSnapshot, SnapshotPersistence};
+pub use persistence::{AgentCheckpoint, CheckpointPersistence};
 pub use runtime::{AgentRuntime, AgentRuntimeComponents};
-pub use tree::{ConversationPath, MessageNode, NodeId};
+pub use tree::{Conversation, ConversationPath, MessageNode, NodeId};
 pub use types::{
     ParseError, PromptTooLongError, SerializedToolExecution, ToolExecution, ToolRequest,
     text_summary_from_blocks, to_tool_definition, to_tool_definitions,

crates/agent_core/src/native/json_processor.rs

@@ -420,9 +420,9 @@ impl JsonStreamProcessor {
                         self.emit_fragment(DisplayFragment::ToolEnd { id: tool_id })?;
                         self.state.json_parsing_state = JsonParsingState::ExpectOpenBrace; // Reset for next potential JSON object
                         self.state.current_key = None;
-                        self.state.buffer.clear(); // Object done, clear buffer of this object. This might be too aggressive if there's trailing content.
-                    // Let's refine: only clear if this was the *only* content, or handle trailing chars.
-                    // For now, `drain` handles consumed chars.
+                        // The loop below consumes this closing brace. Clearing
+                        // here would both discard trailing input and make that
+                        // drain panic for an empty object ({}).
                     } else if char_to_process == ',' {
                         // This is for cases like {"a":"b",} -> expecting a key next.
                         // If we see `,,,` this will just loop. Assuming valid JSON structure mostly.

crates/agent_core/src/native/mod.rs

@@ -11,7 +11,7 @@ use crate::dialect::ToolDialect;
 use crate::types::ToolRequest;
 use crate::ui::{AgentUi, HiddenTools, StreamProcessorTrait};
 use anyhow::Result;
-use llm::{ContentBlock, LLMResponse, Message, MessageContent};
+use llm::{ContentBlock, LLMResponse};
 use std::sync::Arc;
 use tools_core::ToolRegistry;
 
@@ -82,14 +82,4 @@ impl ToolDialect for NativeDialect {
         // Native mode uses API-provided tool definitions, no custom documentation needed
         None
     }
-
-    fn message_contains_invocation(&self, message: &Message, _registry: &ToolRegistry) -> bool {
-        if let MessageContent::Structured(blocks) = &message.content {
-            blocks
-                .iter()
-                .any(|block| matches!(block, ContentBlock::ToolUse { .. }))
-        } else {
-            false
-        }
-    }
 }

crates/agent_core/src/persistence.rs

@@ -1,28 +1,36 @@
-//! Core-shaped persistence: the loop saves what it owns — the conversation
-//! tree, the linearized history, the tool executions, and the id counters.
-//! Application-level fields travel separately through the extension state
-//! and are assembled into the application's storage format by its adapter.
+//! Core-shaped persistence: after every change the loop hands its
+//! persistence the delta since the previous checkpoint — the nodes and
+//! journal entries that changed, plus the small always-current fields.
+//! Prompt-only repairs and context-recovery projections are never part of
+//! it. Application-level fields travel separately through the extension
+//! state and are assembled into the application's storage format by its
+//! adapter.
 
-use crate::tree::{ConversationPath, MessageNode, NodeId};
+use crate::tree::{MessageNode, NodeId};
 use crate::types::ToolExecution;
 use anyhow::Result;
 use std::any::Any;
-use std::collections::BTreeMap;
 
-/// What the agent loop itself knows about and persists.
-pub struct AgentSnapshot {
-    pub session_id: Option<String>,
-    pub message_nodes: BTreeMap<NodeId, MessageNode>,
-    pub active_path: ConversationPath,
+/// What changed since the previous checkpoint of this run.
+pub struct AgentCheckpoint<'a> {
+    pub session_id: &'a str,
+    /// Nodes appended or edited since the previous checkpoint.
+    pub changed_nodes: Vec<&'a MessageNode>,
+    pub active_path: &'a [NodeId],
     pub next_node_id: NodeId,
-    /// Linearized message history (derived from `active_path`).
-    pub messages: Vec<llm::Message>,
-    pub tool_executions: Vec<ToolExecution>,
+    /// Journal entries recorded or updated since the previous checkpoint.
+    pub changed_executions: Vec<&'a ToolExecution>,
     pub next_request_id: u64,
 }
 
-/// Persistence used by the agent loop: it saves the loop's snapshot, with
-/// the application fields supplied by the extension state.
-pub trait SnapshotPersistence: Send + Sync {
-    fn save(&mut self, snapshot: AgentSnapshot, extensions: &(dyn Any + Send)) -> Result<()>;
+/// Persistence used by the agent loop.
+pub trait CheckpointPersistence: Send + Sync {
+    /// Merge the checkpoint into the stored session. A call is atomic: on
+    /// `Err` nothing of it is stored, and the loop keeps the changes marked
+    /// for its next attempt.
+    fn commit(
+        &mut self,
+        checkpoint: &AgentCheckpoint<'_>,
+        extensions: &(dyn Any + Send),
+    ) -> Result<()>;
 }