# motosan-agent-loop

> Standalone ReAct agent loop for Rust. LLM-agnostic iterative reasoning and tool execution — bring your own backend, register tools, let AgentLoop handle the rest.

- Version: 0.8.2
- GitHub: https://github.com/motosan-dev/motosan-agent-loop
- crates.io: https://crates.io/crates/motosan-agent-loop
- docs.rs: https://docs.rs/motosan-agent-loop

## Install

```toml
[dependencies]
motosan-agent-loop = "0.8"
motosan-agent-tool = "0.3"
async-trait = "0.1"
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
```

### Feature Flags

| Feature | Default | Description |
|---------|---------|-------------|
| `motosan-ai` | off | `LlmClient` bridge for motosan-ai SDK |
| `mcp-client` | off | MCP server support via `rmcp` |
| `cancellation` | off | `CancellationToken` for graceful loop cancellation |

---

## Core Concepts

AgentLoop implements the ReAct (Reasoning + Acting) pattern:

1. Send messages + tool definitions to the LLM
2. If the LLM returns tool calls → execute tools → append results → go to 1
3. If the LLM returns text → return final answer

---

## LlmClient Trait

Implement this trait for your LLM backend:

```rust
use async_trait::async_trait;
use motosan_agent_loop::{ChatOutput, LlmClient, LlmResponse, Message, Result};
use motosan_agent_tool::ToolDef;

struct MyLlm;

#[async_trait]
impl LlmClient for MyLlm {
    async fn chat(&self, messages: &[Message], tools: &[ToolDef]) -> Result<ChatOutput> {
        // Call your LLM API here
        Ok(ChatOutput::new(LlmResponse::Message("Hello!".into())))
    }
}
```

### Required Methods

- `chat(&self, messages: &[Message], tools: &[ToolDef]) -> Result<ChatOutput>` — single turn

### Default Methods (override optional)

- `simple_call(&self, system: &str, prompt: &str) -> Result<String>` — simplified system+prompt interface for chain integration. Returns `Err` if the model produces tool calls.
- `chat_stream(messages, tools) -> Pin<Box<dyn Stream<Item = Result<StreamChunk>>>>` — streaming variant; default falls back to `chat()`

---

## AgentLoop

### Builder

```rust
use std::sync::Arc;
use std::time::Duration;
use motosan_agent_loop::AgentLoop;

let agent = AgentLoop::builder()
    .tool(my_tool)                          // register a tool (Arc<dyn Tool>)
    .tools(vec![tool_a, tool_b])            // register multiple tools
    .system_prompt("You are helpful.")      // static system prompt
    .context(my_context_provider)           // dynamic context injection
    .max_iterations(10)                     // default: 10
    .tool_timeout(Duration::from_secs(30))  // per-tool call timeout
    .tool_context(my_tool_context)          // custom ToolContext for all tools
    .with_ask_user()                        // enable built-in ask_user tool (30s timeout)
    .with_ask_user_timeout(Duration::from_secs(60)) // custom timeout
    .channel_config(config)                 // bounded queue config for AgentSession
    .build();
```

### run() — basic execution

```rust
let result = agent
    .run(&my_llm, vec![Message::user("Hi!")], |event| match &event {
        AgentEvent::ToolStarted { name } => println!("tool: {name}"),
        AgentEvent::ToolCompleted { name, .. } => println!("done: {name}"),
        AgentEvent::TextChunk(t) => print!("{t}"),
        _ => {}
    })
    .await?;

println!("{}", result.answer);        // final text answer
println!("{}", result.iterations);     // number of LLM round-trips
println!("{:?}", result.usage);        // TokenUsage { input_tokens, output_tokens }
println!("{:?}", result.messages);     // full conversation history
println!("{:?}", result.tool_calls);   // Vec<(tool_name, args)>
```

### run_streaming() — true streaming

```rust
let result = agent
    .run_streaming(&llm, messages, |event| match &event {
        AgentEvent::TextChunk(delta) => print!("{delta}"),
        AgentEvent::TextDone(full) => println!("\n---\n{full}"),
        _ => {}
    })
    .await?;
```

Override `LlmClient::chat_stream()` for true streaming; default falls back to `chat()`.

### run_with_ops() — interactive control

```rust
use tokio::sync::mpsc;
use motosan_agent_loop::{AgentOp, AgentEvent};

let (ops_tx, ops_rx) = mpsc::channel(32);
let run = agent.run_with_ops(&llm, messages, Some(ops_rx), |event| {
    if let AgentEvent::AskUser { call_id, question, .. } = event {
        println!("Agent asks: {question}");
        let tx = ops_tx.clone();
        tokio::spawn(async move {
            let _ = tx.send(AgentOp::AskUserAnswer {
                call_id: Some(call_id),
                answer: "Option A".into(),
            }).await;
        });
    }
});

// From another task:
let _ = ops_tx.send(AgentOp::InjectHint("Be concise".into())).await;
let _ = ops_tx.send(AgentOp::Interrupt).await;
let result = run.await?;
```

### run_with_cancel() — cancellation support

Requires `cancellation` feature.

```rust
use motosan_agent_loop::CancellationToken;

let token = CancellationToken::new();
let t = token.clone();
tokio::spawn(async move {
    tokio::time::sleep(Duration::from_secs(5)).await;
    t.cancel();
});

let result = agent.run_with_cancel(&llm, messages, token, |_| {}).await;
// Returns Err(AgentError::Cancelled) if cancelled mid-loop.
```

---

## AgentSession (Long-Lived Background Session)

```rust
use std::sync::Arc;
use motosan_agent_loop::{AgentSession, MemorySessionStore, FileSessionStore};

// Basic session
let (session, handle) = AgentSession::new(agent, Arc::new(llm), |_| {});
session.send("Search for Taipei rentals").await;
session.inject("Prefer listings with elevators").await;
session.interrupt().await;

// Persistent session
let store = Arc::new(MemorySessionStore::new());
let (session, handle) = AgentSession::new_with_store(
    "chat-1", store.clone(), agent, Arc::new(llm), |_| {}
);

// Lifecycle
session.flush().await?;    // explicit flush to store
session.close().await?;    // interrupt + final flush + prevent further sends
session.reset().await;     // clear history, bump epoch, discard in-flight turns
let history = session.history().await; // Vec<Message>
```

### subscribe() — push notifications

```rust
let mut rx = session.subscribe(); // watch::Receiver<usize>
tokio::spawn(async move {
    while rx.changed().await.is_ok() {
        let history_len = *rx.borrow();
        println!("History now has {history_len} messages");
    }
});
```

Notifies on every history change (user message added, turn completed). Eliminates polling.

### reply() — answer ask_user requests

```rust
session.reply(Some("call_id"), "my answer").await;
```

---

## ContextProvider

```rust
use async_trait::async_trait;
use motosan_agent_loop::{ContextProvider, Result};

struct RagProvider;

#[async_trait]
impl ContextProvider for RagProvider {
    async fn build(&self, query: &str) -> Result<String> {
        // Fetch relevant documents based on user query
        Ok(format!("Relevant docs for: {query}"))
    }
}
```

Returned string is injected as a System message. Empty string = skipped.
Multiple providers run in parallel via `try_join_all`.

---

## MCP Server Support

Requires `mcp-client` feature.

```toml
motosan-agent-loop = { version = "0.8", features = ["mcp-client"] }
```

```rust
use motosan_agent_loop::mcp::{McpServerStdio, McpServerHttp};

let agent = AgentLoop::builder()
    .mcp_server(McpServerStdio::new("npx", ["-y", "@modelcontextprotocol/server-filesystem"]))
    .mcp_server(McpServerHttp::new("https://mcp.example.com/sse"))
    .build();

// MCP servers auto-connect on run() and disconnect after.
let result = agent.run(&llm, messages, |_| {}).await?;
```

---

## motosan-ai Bridge

Requires `motosan-ai` feature.

```toml
motosan-agent-loop = { version = "0.8", features = ["motosan-ai"] }
motosan-ai = { version = "0.5", features = ["anthropic"] }
```

```rust
use motosan_ai::{Client, Provider};

let client = Client::builder()
    .provider(Provider::Anthropic)
    .api_key(std::env::var("ANTHROPIC_API_KEY")?)
    .build()?;

// Client implements LlmClient — use it directly.
let result = agent.run(&client, messages, |_| {}).await?;
```

---

## Core Types

| Type | Description |
|------|-------------|
| `LlmClient` | Trait — implement `chat()` for your LLM backend |
| `ChatOutput` | Response wrapper: `LlmResponse` + optional `TokenUsage` |
| `LlmResponse` | `Message(String)` or `ToolCalls(Vec<ToolCallItem>)` |
| `StreamChunk` | `TextDelta(String)`, `Done(LlmResponse)`, or `Usage(TokenUsage)` |
| `TokenUsage` | `{ input_tokens: u64, output_tokens: u64 }` |
| `ToolCallItem` | `{ id: String, name: String, args: Value }` |
| `AgentLoop` | The core ReAct loop |
| `AgentLoopBuilder` | Builder pattern for AgentLoop configuration |
| `AgentResult` | `{ answer, tool_calls, iterations, usage, messages }` |
| `AgentEvent` | Callback events: tool, streaming, interactive, backpressure |
| `AgentOp` | Interactive commands: `Interrupt`, `InjectUserMessage`, `InjectHint`, `AskUserAnswer` |
| `AgentSession` | Long-lived background session with managed history |
| `SessionStore` | Persistence trait (`MemorySessionStore`, `FileSessionStore`) |
| `SessionMeta` | Session metadata for store operations |
| `ContextProvider` | Trait — inject dynamic context into conversations |
| `BackpressurePolicy` | `Block` (default), `DropOldest`, `Reject` |
| `ChannelConfig` | Bounded queue config: `input_capacity`, `ops_capacity`, `ops_backpressure` |
| `AgentError` | `Llm`, `Tool`, `Serde`, `Io`, `MaxIterations`, `Cancelled`, `Mcp` |
| `CancellationToken` | Re-exported from `tokio-util` (behind `cancellation` feature) |
| `McpServer` | Trait — MCP server abstraction (`McpServerStdio`, `McpServerHttp`) |
| `Message` | `system(s)`, `user(s)`, `assistant(s)`, `tool_result(id, content)` |

---

## AgentEvent Variants

```
IterationStarted(usize)           — start of each LLM round-trip
ToolStarted { name }              — before tool execution
ToolCompleted { name, result }    — after tool execution
TextChunk(String)                 — streaming text delta
TextDone(String)                  — full text after streaming completes
Interrupted                       — loop interrupted via AgentOp::Interrupt
AskUser { call_id, question, options } — agent requests user input
AskUserTimeout { call_id, question }   — ask_user timed out
OpsSaturated { capacity }         — ops channel full (emitted once)
OpDropped { reason }              — op dropped due to backpressure
OpRejected { reason }             — op rejected due to backpressure
```

### Event Ordering (per iteration)

1. `IterationStarted(n)`
2. `ToolStarted` (per tool)
3. `ToolCompleted` (per tool — may interleave in parallel batches)
4. `TextChunk` / `TextDone` (on final text response)

For `ask_user`: `AskUser` emitted between `ToolStarted` and `ToolCompleted`.

---

## Session Lifecycle Guarantees

- **Epoch-based reset**: `reset()` bumps epoch; in-flight turns with stale epoch are silently discarded.
- **No duplicate persistence**: User message stored once before turn; post-turn skips it.
- **Close semantics**: Sets flag, interrupts current turn, final flush. New sends dropped.
- **Drop behavior**: Best-effort flush. Use `flush()` or `close()` for strict durability.

---

## Dependency Tree

```
motosan-agent-loop
  +-- motosan-agent-tool   (Tool / ToolDef / ToolResult)
  +-- async-trait
  +-- futures
  +-- serde / serde_json
  +-- thiserror
  +-- tokio                (time, sync, rt, fs)
  +-- [motosan-ai]         (optional: motosan-ai feature)
  +-- [rmcp]               (optional: mcp-client feature)
  +-- [tokio-util]         (optional: cancellation feature)
```

## License

MIT
