Users, file manager, image generation, budgets and agent workflow

- User management: local accounts (Argon2) and OpenID Connect, admin/user
  roles, per-user folders plus admin-shared folders, read-only session
  sharing, event filtering per user
- File manager with CodeMirror editor and git client (status, diff,
  stage, commit, branches, fetch/pull/push, log, clone/init)
- Image generation via fal.ai: generate_image tool and image page
- Budgets: monthly limits per provider, provider balances/costs
  (OpenRouter, Anthropic admin cost report, fal.ai), price discovery
- Subagents on other models, background tasks, concurrency limits per
  host/model/subagent
- MCP tools loaded on demand (mcp_search/mcp_call), read-only hints
- Permission mode "auto", safe read-only commands always allowed
- Persistent message queue with attachments, deletable
- Live reasoning incl. <think> tags, activity animations, CSS themes,
  mobile layout, statistics per user, upload size handling
- Docker: named volume for state, localhost-bound port

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Richard ZinkandClaude Opus 5.5 committed 2026-11-08 16:54:08 +01:00
1 parent f3a2fe3816
commit 49262c612c
70 files changed
+8495 -714

No files matched your search

+299 -66
View File
@@ -20,30 +20,55 @@ use tokio_util::sync::CancellationToken;
const MAX_STEPS: usize = 250;
const MAX_SUBAGENT_DEPTH: usize = 3;
/// Start processing a user prompt. If the session is busy the prompt is queued
/// and injected at the next step.
/// What happened to a prompt.
pub enum Submitted {
Started,
Queued(crate::store::QueueItem),
}
fn emit_queue(app: &App, session_id: &str) {
let items = app.store.queue_list(session_id).unwrap_or_default();
app.emit(Event::QueueUpdated { session_id: session_id.into(), items: serde_json::json!(items) });
}
/// Take all queued messages (one user message's worth of parts).
fn take_queue(app: &App, session_id: &str) -> Vec<Part> {
let items = app.store.queue_take(session_id).unwrap_or_default();
if !items.is_empty() {
emit_queue(app, session_id);
}
items.into_iter().flat_map(|i| i.parts).collect()
}
/// Start processing a user prompt. If the session is busy, the prompt (with
/// attachments) is queued and delivered at the agent's next step.
pub fn start_prompt(app: Arc<App>, session_id: String, parts: Vec<Part>) -> Result<()> {
submit(app, session_id, parts, "user").map(|_| ())
}
pub fn submit(app: Arc<App>, session_id: String, parts: Vec<Part>, source: &str) -> Result<Submitted> {
// Queueing and the end-of-run check run under the same runtime lock, so a
// message is never lost or delivered twice.
let cancel = CancellationToken::new();
let queued = app.with_runtime(&session_id, |r| -> Result<Option<crate::store::QueueItem>> {
if r.cancel.is_some() {
let item = crate::store::QueueItem { id: new_id(), parts: parts.clone(), source: source.into(), created_at: now_ms() };
app.store.queue_add(&session_id, &item)?;
Ok(Some(item))
} else {
r.cancel = Some(cancel.clone());
Ok(None)
}
})?;
if let Some(item) = queued {
emit_queue(&app, &session_id);
return Ok(Submitted::Queued(item));
}
let text_only: String = parts
.iter()
.filter_map(|p| if let Part::Text { text } = p { Some(text.as_str()) } else { None })
.collect::<Vec<_>>()
.join("\n");
let queued = app.with_runtime(&session_id, |r| {
if r.cancel.is_some() && parts.iter().all(|p| matches!(p, Part::Text { .. })) {
r.queue.push(text_only.clone());
true
} else {
false
}
});
if queued {
return Ok(());
}
if app.is_busy(&session_id) {
bail!("session is busy");
}
let cancel = CancellationToken::new();
app.with_runtime(&session_id, |r| r.cancel = Some(cancel.clone()));
app.emit(Event::SessionStatus { session_id: session_id.clone(), busy: true });
tokio::spawn(async move {
@@ -59,17 +84,31 @@ pub fn start_prompt(app: Arc<App>, session_id: String, parts: Vec<Part>) -> Resu
tracing::warn!("session {session_id}: {e:#}");
app.emit(Event::Error { session_id: session_id.clone(), message: format!("{e:#}") });
}
// Prompts queued after the last step start a new turn.
// Messages queued after the last step start a new turn – unless the user
// stopped the agent: then they stay queued until sent explicitly.
let leftover = app.with_runtime(&session_id, |r| {
r.cancel = None;
std::mem::take(&mut r.queue)
if cancel.is_cancelled() { Vec::new() } else { take_queue(&app, &session_id) }
});
app.emit(Event::SessionStatus { session_id: session_id.clone(), busy: false });
if !leftover.is_empty() && !cancel.is_cancelled() {
let _ = start_prompt(app, session_id, vec![Part::Text { text: leftover.join("\n\n") }]);
if !leftover.is_empty() {
let _ = start_prompt(app, session_id, leftover);
}
});
Ok(())
Ok(Submitted::Started)
}
/// Send queued messages now (after a stop or a server restart).
pub fn flush_queue(app: Arc<App>, session_id: &str) -> Result<bool> {
if app.is_busy(session_id) {
return Ok(false);
}
let parts = take_queue(&app, session_id);
if parts.is_empty() {
return Ok(false);
}
start_prompt(app, session_id.to_string(), parts)?;
Ok(true)
}
pub fn abort(app: &App, session_id: &str) {
@@ -77,17 +116,72 @@ pub fn abort(app: &App, session_id: &str) {
if let Some(c) = &r.cancel {
c.cancel();
}
r.queue.clear();
});
}
/// Messages since the last compaction summary.
/// Messages since the last compaction summary. Tool calls left without a
/// result (run interrupted, e.g. by a server restart) get a synthetic result:
/// providers reject histories with unanswered tool calls.
fn active_history(all: Vec<Message>) -> Vec<Message> {
let start = all
.iter()
.rposition(|m| m.parts.iter().any(|p| matches!(p, Part::Summary { .. })))
.unwrap_or(0);
all.into_iter().skip(start).collect()
let msgs: Vec<Message> = all.into_iter().skip(start).collect();
let mut out: Vec<Message> = Vec::with_capacity(msgs.len());
for (i, m) in msgs.iter().enumerate() {
out.push(m.clone());
if m.role != Role::Assistant {
continue;
}
let answered: std::collections::HashSet<&str> = msgs
.get(i + 1)
.filter(|n| n.role == Role::User)
.map(|n| {
n.parts
.iter()
.filter_map(|p| if let Part::ToolResult { call_id, .. } = p { Some(call_id.as_str()) } else { None })
.collect()
})
.unwrap_or_default();
let missing: Vec<Part> = m
.parts
.iter()
.filter_map(|p| match p {
Part::ToolCall { id, name, .. } if !answered.contains(id.as_str()) => {
Some(tool_result(id, name, "Not executed: the run was interrupted (e.g. server restart).".into(), true, None))
}
_ => None,
})
.collect();
if missing.is_empty() {
continue;
}
// Results must directly follow the call: merge into the next user message or insert one.
match msgs.get(i + 1) {
Some(n) if n.role == Role::User => {
// The next message is pushed in the next iteration; prepend there.
let mut next = n.clone();
let mut parts = missing;
parts.extend(next.parts);
next.parts = parts;
// Replace by pushing now and skipping the original.
out.push(next);
}
_ => out.push(Message::new(Role::User, missing)),
}
}
// Remove duplicates created by the merge above (same id pushed twice: keep the merged one).
let mut dedup: Vec<Message> = Vec::with_capacity(out.len());
for m in out {
if let Some(last) = dedup.last() {
if last.id == m.id {
continue;
}
}
dedup.push(m);
}
dedup
}
fn normalize_tool_name(s: &str) -> String {
@@ -95,8 +189,17 @@ fn normalize_tool_name(s: &str) -> String {
}
pub fn available_tools(app: &App, agent: &Agent) -> Vec<Arc<dyn Tool>> {
let cfg = app.config();
let mut all = tools::builtin_tools();
all.extend(app.mcp.tools());
if !cfg.images.enabled {
all.retain(|t| t.name() != "generate_image");
}
let (direct, deferred) = app.mcp.tools(&cfg);
all.extend(direct);
if !deferred.is_empty() {
all.push(Arc::new(crate::mcp::McpSearch));
all.push(Arc::new(crate::mcp::McpCall));
}
all.into_iter()
.filter(|t| !agent.read_only || t.read_only())
// Craft tools only where they are useful.
@@ -107,7 +210,7 @@ pub fn available_tools(app: &App, agent: &Agent) -> Vec<Arc<dyn Tool>> {
let n = normalize_tool_name(t.name());
list.iter().any(|a| {
let a = normalize_tool_name(a);
a == n || (t.name().starts_with("mcp__") && a == "mcp")
a == n || (t.name().starts_with("mcp") && a == "mcp")
})
}
})
@@ -119,6 +222,17 @@ struct Step {
stop_reason: String,
}
/// Boxed with an explicit `Send` bound: subagents recurse into the agent loop,
/// and the declared type breaks the cycle for the compiler.
fn run_session_boxed(
app: Arc<App>,
session_id: String,
cancel: CancellationToken,
depth: usize,
) -> futures::future::BoxFuture<'static, Result<String>> {
Box::pin(async move { run_session(app, &session_id, cancel, depth).await })
}
/// Run the loop for a session until the model stops. Returns the final text.
pub async fn run_session(app: Arc<App>, session_id: &str, cancel: CancellationToken, depth: usize) -> Result<String> {
let mut final_text = String::new();
@@ -128,7 +242,9 @@ pub async fn run_session(app: Arc<App>, session_id: &str, cancel: CancellationTo
}
let session = app.session(session_id)?;
let cwd = PathBuf::from(&session.cwd);
let agent = prompts::find_agent(&app, &cwd, &session.agent);
let roots = app.user_roots(session.owner_id.as_deref());
let root = app.root_for(session.owner_id.as_deref(), &cwd);
let agent = prompts::find_agent(&app, &root, &cwd, &session.agent);
let model_spec = agent.model.clone().unwrap_or_else(|| session.model.clone());
let (provider, model) = app.registry().resolve(&model_spec)?;
@@ -144,7 +260,7 @@ pub async fn run_session(app: Arc<App>, session_id: &str, cancel: CancellationTo
let tools = available_tools(&app, &agent);
let history = active_history(app.store.messages(session_id)?);
let req = ChatRequest {
system: prompts::system_prompt(&app, &agent, &cwd, &model_spec),
system: prompts::system_prompt(&app, &agent, &root, &cwd, &model_spec, &session.permission_mode),
messages: history,
tools: tools.iter().map(|t| t.def()).collect(),
max_tokens: model.max_output.unwrap_or(16_000),
@@ -203,6 +319,7 @@ pub async fn run_session(app: Arc<App>, session_id: &str, cancel: CancellationTo
cwd: cwd.clone(),
cancel: cancel.clone(),
agent: agent.name.clone(),
roots: roots.clone(),
};
let by_name = |n: &str| tools.iter().find(|t| t.name() == n).cloned();
let all_read_only = calls.iter().all(|(_, n, _)| by_name(n).is_some_and(|t| t.read_only()));
@@ -217,11 +334,8 @@ pub async fn run_session(app: Arc<App>, session_id: &str, cancel: CancellationTo
results.push(execute_call(&app, &session, &base, by_name(name), id, name, input.clone(), depth).await);
}
}
// Inject prompts the user sent while we were working.
let queued = app.with_runtime(session_id, |r| std::mem::take(&mut r.queue));
for q in queued {
results.push(Part::Text { text: q });
}
// Deliver messages the user queued while we were working (with attachments).
results.extend(take_queue(&app, session_id));
save_user_parts(&app, session_id, results)?;
}
Ok(final_text)
@@ -265,12 +379,6 @@ async fn execute_call(
return tool_result(call_id, name, "Maximum subagent depth reached.".into(), true, None);
}
// Permission check. Approvals are tracked on the top-level session.
let root = app.root_session_id(&session.id);
let cfg = app.config();
let decision = app.with_runtime(&root, |r| {
Evaluator { cfg: &cfg.permissions, mode: &session.permission_mode, approved: &r.approved }.evaluate(tool.as_ref(), &input)
});
let ctx = ToolCtx {
app: app.clone(),
session_id: base.session_id.clone(),
@@ -278,10 +386,33 @@ async fn execute_call(
cwd: base.cwd.clone(),
cancel: base.cancel.clone(),
agent: base.agent.clone(),
roots: base.roots.clone(),
};
// mcp_call is judged (and approved) as the MCP tool it calls.
let call_name = name;
let (tool, input, name): (Arc<dyn Tool>, Value, String) = if name == "mcp_call" {
match crate::mcp::resolve_call_target(&ctx, &input) {
Ok(target) => {
let args = input.get("arguments").cloned().unwrap_or(serde_json::json!({}));
let n = target.full_name.clone();
(target, args, n)
}
Err(e) => return tool_result(call_id, call_name, format!("Error: {e:#}"), true, None),
}
} else {
(tool, input, name.to_string())
};
let name = name.as_str();
// Permission check. Approvals are tracked on the top-level session.
let root = app.root_session_id(&session.id);
let cfg = app.config();
let decision = app.with_runtime(&root, |r| {
Evaluator { cfg: &cfg.permissions, mode: &session.permission_mode, approved: &r.approved }.evaluate(tool.as_ref(), &input)
});
match decision {
Decision::Deny => {
return tool_result(call_id, name, "Denied by the permission rules in the configuration.".into(), true, None);
return tool_result(call_id, call_name, "Denied by the permission rules in the configuration.".into(), true, None);
}
Decision::Ask => {
let diff = tool.preview(&input, &ctx).await;
@@ -294,7 +425,7 @@ async fn execute_call(
if let Some(m) = message.filter(|m| !m.trim().is_empty()) {
text.push_str(&format!(" User feedback: {m}"));
}
return tool_result(call_id, name, text, true, None);
return tool_result(call_id, call_name, text, true, None);
}
PermissionReply::Always => app.with_runtime(&root, |r| {
if name == "bash" {
@@ -321,40 +452,76 @@ async fn execute_call(
match res {
Ok(out) => Part::ToolResult {
call_id: call_id.into(),
name: name.into(),
name: call_name.into(),
content: out.content,
images: out.images,
is_error: out.is_error,
meta: out.meta,
},
Err(e) => tool_result(call_id, name, format!("Error: {e:#}"), true, None),
Err(e) => tool_result(call_id, call_name, format!("Error: {e:#}"), true, None),
}
}
async fn run_task(app: &Arc<App>, ctx: &ToolCtx, input: Value, depth: usize) -> Result<tools::ToolOutput> {
let prompt = tools::arg_str(&input, "prompt")?.to_string();
let description = input["description"].as_str().unwrap_or("subagent").to_string();
let agent = input["subagent_type"].as_str().unwrap_or("general").to_string();
let (child, text) = spawn_subagent(app.clone(), ctx, &agent, &description, prompt, depth + 1).await?;
Ok(tools::ToolOutput::text(text).with_meta(serde_json::json!({"session_id": child, "agent": agent})))
let agent_name = input["subagent_type"].as_str().unwrap_or("general").to_string();
let background = input["background"].as_bool().unwrap_or(false);
let (child, agent) = create_child(app, ctx, &agent_name, &description, prompt)?;
let meta = serde_json::json!({
"session_id": child.id, "agent": agent.name, "model": child.model, "background": background
});
if background {
let app = app.clone();
let parent_id = ctx.session_id.clone();
let child_id = child.id.clone();
let agent_name = agent.name.clone();
let desc = description.clone();
// Background tasks outlive the current turn; they are stopped via their own session.
let cancel = CancellationToken::new();
tokio::spawn(async move {
let res = run_child(app.clone(), &child, &agent, cancel, depth + 1).await;
let body = match res {
Ok(text) => text,
Err(e) => format!("The background task failed: {e:#}"),
};
let text = format!(
"<background-task-result agent=\"{agent_name}\" session=\"{child_id}\" description=\"{desc}\">\n{body}\n</background-task-result>"
);
// Injected into the running turn, or starts a new turn when the parent is idle.
if app.store.get_session(&parent_id).ok().flatten().is_some() {
if let Err(e) = submit(app.clone(), parent_id.clone(), vec![Part::Text { text }], "background") {
tracing::warn!("could not deliver background result: {e:#}");
}
}
});
return Ok(tools::ToolOutput::text(format!(
"Started background task '{description}' with subagent '{}' (model {}). Its result will arrive \
automatically as a message when it finishes – continue with other work and do not poll.",
meta["agent"].as_str().unwrap_or(""),
meta["model"].as_str().unwrap_or("")
))
.with_meta(meta));
}
let text = run_child(app.clone(), &child, &agent, ctx.cancel.child_token(), depth + 1).await?;
Ok(tools::ToolOutput::text(text).with_meta(meta))
}
/// Entry point used by the task tool when called outside the runner.
pub async fn run_subagent(app: Arc<App>, ctx: &ToolCtx, agent: &str, description: &str, prompt: String) -> Result<(String, String)> {
spawn_subagent(app, ctx, agent, description, prompt, 1).await
let (child, agent) = create_child(&app, ctx, agent, description, prompt)?;
let text = run_child(app, &child, &agent, ctx.cancel.child_token(), 1).await?;
Ok((child.id, text))
}
async fn spawn_subagent(
app: Arc<App>,
ctx: &ToolCtx,
agent_name: &str,
description: &str,
prompt: String,
depth: usize,
) -> Result<(String, String)> {
/// Create the subagent session with its first message.
fn create_child(app: &Arc<App>, ctx: &ToolCtx, agent_name: &str, description: &str, prompt: String) -> Result<(Session, Agent)> {
let parent = app.session(&ctx.session_id)?;
let parent_agent = prompts::find_agent(&app, &ctx.cwd, &parent.agent);
let agents = prompts::agents(&app, &ctx.cwd);
let root = ctx.root();
let parent_agent = prompts::find_agent(app, &root, &ctx.cwd, &parent.agent);
let agents = prompts::agents(app, &root, &ctx.cwd);
let mut agent = agents
.iter()
.find(|a| a.name == agent_name && a.mode != "primary")
@@ -367,6 +534,9 @@ async fn spawn_subagent(
if parent_agent.read_only && !agent.read_only {
agent = agents.into_iter().find(|a| a.name == "explore").unwrap_or(agent);
}
if let Some(m) = &agent.model {
app.registry().resolve(m).map_err(|e| anyhow!("subagent '{}' has an unusable model: {e:#}", agent.name))?;
}
let now = now_ms();
let child = Session {
id: new_id(),
@@ -376,6 +546,7 @@ async fn spawn_subagent(
agent: agent.name.clone(),
permission_mode: parent.permission_mode.clone(),
parent_id: Some(parent.id.clone()),
owner_id: parent.owner_id.clone(),
created_at: now,
updated_at: now,
todos: Value::Null,
@@ -385,11 +556,34 @@ async fn spawn_subagent(
let msg = Message::new(Role::User, vec![Part::Text { text: prompt }]);
app.store.add_message(&child.id, &msg)?;
app.emit(Event::MessageAdded { session_id: child.id.clone(), message: msg });
Ok((child, agent))
}
/// Run a subagent session to completion, honouring the subagent limits.
async fn run_child(app: Arc<App>, child: &Session, agent: &Agent, cancel: CancellationToken, depth: usize) -> Result<String> {
// Registered like a normal run so the user can stop it from its own session.
app.with_runtime(&child.id, |r| r.cancel = Some(cancel.clone()));
app.emit(Event::SessionStatus { session_id: child.id.clone(), busy: true });
let res = Box::pin(run_session(app.clone(), &child.id, ctx.cancel.child_token(), depth)).await;
// Only first-level subagents count against the limits: nested subagents would
// otherwise deadlock waiting for a slot their own parent holds.
let _slot = if depth <= 1 {
let (slot, waited) = app.subagent_slot(&agent.name, agent.max_parallel, &child.id, &cancel).await;
if waited {
app.emit(Event::Waiting { session_id: child.id.clone(), reason: None });
}
slot
} else {
None
};
let res = if cancel.is_cancelled() {
Err(anyhow!("aborted"))
} else {
run_session_boxed(app.clone(), child.id.clone(), cancel, depth).await
};
app.with_runtime(&child.id, |r| r.cancel = None);
app.emit(Event::SessionStatus { session_id: child.id.clone(), busy: false });
let text = res?;
Ok((child.id, if text.trim().is_empty() { "(subagent returned no text)".into() } else { text }))
Ok(if text.trim().is_empty() { "(subagent returned no text)".into() } else { text })
}
fn is_retryable(e: &anyhow::Error) -> bool {
@@ -415,7 +609,20 @@ async fn stream_step(
let p = provider.clone();
let r = req.clone();
let c = cancel.clone();
let handle = tokio::spawn(async move { p.stream(r, tx, c).await });
// Wait for a free slot on the model/host; not counted as time to first token.
let registry = app.registry();
if registry.is_saturated(model_spec) {
app.emit(Event::Waiting { session_id: session.id.clone(), reason: Some(format!("model:{model_spec}")) });
}
let slot = tokio::select! {
s = registry.acquire(model_spec) => s,
_ = cancel.cancelled() => bail!("aborted"),
};
app.emit(Event::Waiting { session_id: session.id.clone(), reason: None });
let handle = tokio::spawn(async move {
let _slot = slot;
p.stream(r, tx, c).await
});
let started = Instant::now();
let mut ttft: Option<u128> = None;
@@ -516,9 +723,13 @@ async fn stream_step(
}
/// Collect a complete, tool-free answer (titles, summaries).
async fn complete_text(provider: Arc<dyn Provider>, req: ChatRequest, cancel: CancellationToken) -> Result<(String, Usage)> {
async fn complete_text(app: &App, spec: &str, provider: Arc<dyn Provider>, req: ChatRequest, cancel: CancellationToken) -> Result<(String, Usage)> {
let slot = app.registry().acquire(spec).await;
let (tx, mut rx) = mpsc::channel(256);
let handle = tokio::spawn(async move { provider.stream(req, tx, cancel).await });
let handle = tokio::spawn(async move {
let _slot = slot;
provider.stream(req, tx, cancel).await
});
let mut text = String::new();
let mut usage = Usage::default();
while let Some(ev) = rx.recv().await {
@@ -579,7 +790,7 @@ fn maybe_generate_title(app: Arc<App>, session_id: &str, text: &str) {
reasoning: false,
model: model.clone(),
};
match complete_text(provider, req, CancellationToken::new()).await {
match complete_text(&app, &spec, provider, req, CancellationToken::new()).await {
Ok((title, usage)) => {
record_side_usage(&app, &session, &spec, &model, &usage, "title");
let title = title.lines().find(|l| !l.trim().is_empty()).unwrap_or("").trim().trim_matches('"').to_string();
@@ -625,7 +836,7 @@ pub async fn compact(app: Arc<App>, session_id: &str, cancel: CancellationToken)
reasoning: false,
model: model.clone(),
};
let (summary, usage) = complete_text(provider, req, cancel).await?;
let (summary, usage) = complete_text(&app, &spec, provider, req, cancel).await?;
record_side_usage(&app, &session, &spec, &model, &usage, "compaction");
if summary.trim().is_empty() {
bail!("compaction produced an empty summary");
@@ -638,3 +849,25 @@ pub async fn compact(app: Arc<App>, session_id: &str, cancel: CancellationToken)
app.save_session(s)?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn call(id: &str) -> Part {
Part::ToolCall { id: id.into(), name: "bash".into(), input: serde_json::json!({}), extra: None }
}
#[test]
fn dangling_tool_calls_get_results() {
let a = Message::new(Role::Assistant, vec![call("1"), call("2")]);
let r = Message::new(Role::User, vec![tool_result("1", "bash", "ok".into(), false, None)]);
let a2 = Message::new(Role::Assistant, vec![call("3")]);
let h = active_history(vec![Message::new(Role::User, vec![Part::Text { text: "hi".into() }]), a, r, a2]);
// user, assistant(1,2), user(result 2 + result 1), assistant(3), user(result 3)
assert_eq!(h.len(), 5);
let ids = |m: &Message| m.parts.iter().filter_map(|p| if let Part::ToolResult { call_id, .. } = p { Some(call_id.clone()) } else { None }).collect::<Vec<_>>();
assert_eq!(ids(&h[2]), vec!["2", "1"]);
assert_eq!(ids(&h[4]), vec!["3"]);
}
}