starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-10-28 00:02:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: tool approval and a question tool for ai agent steps (#11826)

Browse Source 
* feat(ai-agent): built-in approval tool that parks the agent

An AI agent step can now be given a built-in "Request approval" tool. The model
calls `request_approval` with a message; the agent stores its conversation in
the new `ai_agent_checkpoint` table, releases its worker by parking its own
queue row like a workflow-as-code parent, and resumes with the answer (approved,
approver, comment) or a timeout once someone answers from the run page.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(ai-agent): tie approval answers to their request and hold the turn

- A turn that asks for approval runs none of its other calls; they are refused
  so the model repeats them after the answer.
- The run page sends the request's call id. An answer naming another request is
  ignored and the agent parks again for the time left; `suspend` is zeroed when
  the answer is read, so answers sent while the agent runs are refused.
- The checkpoint is read whatever the current roster holds, the answer's
  outcome is written by position, and a cancel that beats the park completes
  the job as cancelled.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(ai-agent): say why calls beside an invalid approval request were refused

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(ai-agent): one approval form per request, none on a finished run

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(ai-agent): answer approval requests from a flow's chat

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(ai-agent): show chat approvals for agents that do not stream

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(ai-agent): approval as a per-tool setting, built-in tool asks a question

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(ai-agent): say when a nested agent leaves out tools that require approval

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(ai-agent): show chat approvals and questions on the tool's own card

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(ai-agent): question choices, and approval buttons shared with approval steps

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(ai-agent): show a waiting agent as suspended, and refused calls as not run

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(ai-agent): each added chat card shows its own call, answered marker keyed by job

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test(ai-agent): approve and reject a held call through resume_suspended

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(ai-agent): require approval in a tool settings tab, plain descriptions on built-in tool panes

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(ai-agent): sentence case for the web search tool label

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(ai-agent): built-in tool drawers explain themselves in the header subtitle

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(ai-agent): clearer question tool subtitle

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(ai-agent): badge the tool settings tab like run settings

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* chore(ai-agent): regenerate system prompts and MCP session files

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
hugocasa
2026-10-26 19:22:45 +01:00
committed by GitHub
co-authored by Claude Opus 5.5
parent a2b91fd17e
commit a1bc9ec0c2
60 changed files with 2193 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/migrations/20261016162101_ai_agent_checkpoint.down.sql
+1
View File
@@ -0,0 +1 @@
DROP TABLE IF EXISTS ai_agent_checkpoint;
backend/migrations/20261016162101_ai_agent_checkpoint.up.sql
+9
View File
@@ -0,0 +1,9 @@
-- The conversation of an AI agent step parked on its approval tool, read back when the job is
-- pulled again. No foreign key to v2_job_queue: one would lock that table on every write here.
-- The monitor deletes rows whose job left the queue.
CREATE TABLE IF NOT EXISTS ai_agent_checkpoint (
job_id UUID PRIMARY KEY,
workspace_id VARCHAR(50) NOT NULL,
checkpoint JSONB NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
backend/src/monitor.rs
+19
View File
@@ -4172,6 +4172,9 @@ pub async fn monitor_db(
if let Err(e) = cleanup_flow_iterator_data_orphaned_jobs(&db).await {
tracing::error!("Error cleaning up flow_iterator_data: {:?}", e);
}
if let Err(e) = cleanup_ai_agent_checkpoint_orphaned_jobs(&db).await {
tracing::error!("Error cleaning up ai_agent_checkpoint: {:?}", e);
}
}
}
};
@@ -7729,6 +7732,22 @@ async fn cleanup_job_result_stream_orphaned_jobs(db: &DB) -> error::Result<()> {
Ok(())
}
/// A parked agent cancelled from outside, or a worker that died mid-segment, never deletes its
/// checkpoint itself.
async fn cleanup_ai_agent_checkpoint_orphaned_jobs(db: &DB) -> error::Result<()> {
let deleted = sqlx::query(
"DELETE FROM ai_agent_checkpoint c
WHERE NOT EXISTS (SELECT 1 FROM v2_job_queue q WHERE q.id = c.job_id)",
)
.execute(db)
.await?
.rows_affected();
if deleted > 0 {
tracing::info!("Cleaned up {deleted} orphaned ai_agent_checkpoint rows");
}
Ok(())
}
async fn cleanup_flow_iterator_data_orphaned_jobs(db: &DB) -> error::Result<()> {
let result = sqlx::query!(
"
backend/summarized_schema.txt
+1
View File
@@ -37,6 +37,7 @@ _sqlx_migrations: version(bigint), description(text), installed_on(ts), success(
account: workspace_id(char), id(int), expires_at(ts), refresh_token(char), client(char), refresh_error(text), grant_type(char), cc_client_id(char), cc_client_secret(char), cc_token_url(char), mcp_server_url(text), scopes(text[])
FK: (workspace_id) -> workspace(id)
agent_token_blacklist: token(char), expires_at(ts), blacklisted_at(ts), blacklisted_by(char)
ai_agent_checkpoint: job_id(uuid), workspace_id(char), checkpoint(jsonb), created_at(ts)
ai_agent_memory: workspace_id(char), conversation_id(uuid), step_id(char), messages(jsonb), created_at(ts), updated_at(ts)
ai_free_token_daily_usage: day(date), cost_nanos(bigint), updated_at(ts)
ai_free_token_usage: email(char), cost_nanos(bigint), updated_at(ts)
backend/tests/ai_agent_approval.rs
+154
View File
@@ -0,0 +1,154 @@
use axum::{routing::post, Json, Router};
use serde_json::{json, Value};
use sqlx::{types::Uuid, Pool, Postgres};
use std::time::Duration;
use windmill_common::{flows::FlowValue, jobs::JobPayload};
use windmill_test_utils::{
completed_job, in_test_worker, listen_for_completed_jobs, ApiServer, RunJob, StreamFind,
};
async fn model(Json(body): Json<Value>) -> ([(&'static str, &'static str); 1], String) {
let answered = body["messages"]
.as_array()
.unwrap()
.iter()
.any(|m| m["role"] == "tool");
let delta = if answered {
json!({"role":"assistant", "content":"done"})
} else {
json!({"role":"assistant", "tool_calls":[{
"index":0, "id":"call_0", "type":"function",
"function":{"name":"delete_database", "arguments":"{}"}
}]})
};
let event = json!({"choices":[{"index":0, "delta":delta, "finish_reason":null}]});
let end = json!({"choices":[{"index":0, "delta":{}, "finish_reason":if answered {"stop"} else {"tool_calls"}}]});
(
[("content-type", "text/event-stream")],
format!("data: {event}\n\ndata: {end}\n\ndata: [DONE]\n\n"),
)
}
/// The agent parks on a call that requires approval, is answered through the endpoint the run
/// page and the chat use, and resumes: an approved call runs, a rejected one never does.
async fn answer(db: Pool<Postgres>, approved: bool) -> anyhow::Result<Value> {
std::env::set_var("ALLOW_PRIVATE_AI_BASE_URLS", "true");
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
let base = format!("http://{}", listener.local_addr()?);
let stub = tokio::spawn(async move {
axum::serve(
listener,
Router::new().route("/v1/chat/completions", post(model)),
)
.await
.unwrap();
});
let flow: FlowValue = serde_json::from_value(json!({"modules":[{"id":"agent","value":{
"type":"aiagent",
"tools":[{"id":"t0", "summary":"delete_database", "requires_approval":true, "value":{
"type":"rawscript", "language":"bun", "input_transforms":{},
"content":"export async function main() { return 'ran' }"
}}],
"input_transforms":{
"provider":{"type":"static","value":{"kind":"customai","model":"approval-test","resource":{"base_url":format!("{base}/v1")}}},
"user_message":{"type":"static","value":"delete it"},
"max_iterations":{"type":"static","value":3}
}
}}]}))?;
let id = RunJob::from(JobPayload::RawFlow {
value: flow,
path: Some("u/test/agent_approval".into()),
restarted_from: None,
})
.push(&db)
.await;
let notifications = listen_for_completed_jobs(&db).await;
let db_ = db.clone();
let answered = async move {
let agent = loop {
let parked: Option<Uuid> = sqlx::query_scalar(
"SELECT q.id FROM v2_job_queue q JOIN v2_job j USING (id)
JOIN ai_agent_checkpoint c ON c.job_id = q.id
WHERE j.parent_job = $1 AND q.suspend > 0",
)
.bind(id)
.fetch_optional(&db_)
.await
.unwrap();
if let Some(agent) = parked {
break agent;
}
tokio::time::sleep(Duration::from_millis(200)).await;
};
let token = windmill_common::auth::create_token_for_owner(
&db_,
"test-workspace",
"u/test-user",
"approval-test",
100,
"test@windmill.dev",
&Uuid::nil(),
None,
None,
)
.await
.unwrap();
let response = reqwest::Client::new()
.post(format!(
"http://localhost:{port}/api/w/test-workspace/jobs_u/flow/resume_suspended/{agent}"
))
.bearer_auth(token)
.json(&json!({"approved":approved, "payload":{"call_id":"call_0", "comment":"not today"}}))
.send()
.await
.unwrap();
assert!(
response.status().is_success(),
"{}",
response.text().await.unwrap()
);
notifications.find(&id).await;
};
tokio::time::timeout(Duration::from_secs(60), in_test_worker(&db, answered, port)).await?;
let result = completed_job(id, &db).await;
assert!(result.success, "{:?}", result.result);
let tool_jobs: i64 = sqlx::query_scalar(
"SELECT count(*) FROM v2_job WHERE parent_job IN (SELECT id FROM v2_job WHERE parent_job = $1)",
)
.bind(id)
.fetch_one(&db)
.await?;
assert_eq!(tool_jobs, approved as i64, "only an approved call runs");
stub.abort();
server.close().await?;
let result = result.json_result().expect("agent result");
Ok(result["messages"]
.as_array()
.unwrap()
.iter()
.find(|m| m["role"] == "tool")
.expect("the call's answer")["content"]
.clone())
}
#[sqlx::test(fixtures("base"))]
#[serial_test::serial]
async fn approved_call_runs_after_the_agent_resumes(db: Pool<Postgres>) -> anyhow::Result<()> {
assert_eq!(answer(db, true).await?, "\"ran\"");
Ok(())
}
#[sqlx::test(fixtures("base"))]
#[serial_test::serial]
async fn rejected_call_reaches_the_model_as_not_run(db: Pool<Postgres>) -> anyhow::Result<()> {
let content = answer(db, false).await?;
let content = content.as_str().unwrap();
assert!(
content.starts_with("Not run") && content.contains("not today"),
"{content}"
);
Ok(())
}
backend/windmill-ai/src/types.rs
+5
View File
@@ -60,6 +60,11 @@ pub struct Tool {
pub module: Option<FlowModule>,
pub def: ToolDef,
pub mcp_source: Option<McpToolSource>,
/// Each call waits for a person to approve it, without the agent holding a worker.
pub requires_approval: bool,
/// Set on the built-in question tool, which the worker answers by parking the agent until
/// someone answers or this many seconds pass.
pub question_timeout_secs: Option<u32>,
}
#[derive(Deserialize, Serialize, Debug, Clone, PartialEq)]
backend/windmill-api/src/mcp/session/gen/instructions.json
+1 -1
View File
File diff suppressed because one or more lines are too long
backend/windmill-types/src/flow_status.rs
+25
View File
@@ -218,6 +218,31 @@ pub enum AgentAction {
},
Message {},
WebSearch {},
/// The agent is parked until someone approves or rejects these calls of tools that require
/// it; it is answered once its `agent_actions_success` entry is written. `call_id` is the
/// first call's, and names the request.
Approval {
call_id: String,
calls: Vec<ApprovalCall>,
expires_at: chrono::DateTime<chrono::Utc>,
},
/// The agent is parked until someone answers its question, the same way.
Question {
call_id: String,
question: String,
/// Answers the model proposes; a free-text answer is always possible.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
choices: Vec<String>,
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
multi_select: bool,
expires_at: chrono::DateTime<chrono::Utc>,
},
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct ApprovalCall {
pub function_name: String,
pub arguments: serde_json::Value,
}
#[derive(Serialize, Debug, Clone)]
backend/windmill-types/src/flows.rs
+19 -4
View File
@@ -386,8 +386,9 @@ impl FlowValue {
for tool in tools {
match &tool.value {
ToolValue::FlowModule(module_value) => cb(module_value, &tool.id)?,
ToolValue::Mcp(_) => {}
ToolValue::Websearch(_) => {}
ToolValue::Mcp(_)
| ToolValue::Websearch(_)
| ToolValue::Question(_) => {}
}
}
}
@@ -943,6 +944,10 @@ pub struct AgentTool {
/// Caps the token of this tool's jobs, on top of the agent step's own restriction.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub job_token_scopes: Option<Vec<String>>,
/// A person approves each call before it runs; the agent waits without holding a worker.
/// Flow module and MCP tools only.
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
pub requires_approval: bool,
pub value: ToolValue,
}
@@ -972,8 +977,7 @@ impl From<&AgentTool> for Option<FlowModule> {
job_token_scopes: tool.job_token_scopes.clone(),
..Default::default()
}),
ToolValue::Mcp(_) => None,
ToolValue::Websearch(_) => None,
ToolValue::Mcp(_) | ToolValue::Websearch(_) | ToolValue::Question(_) => None,
}
}
}
@@ -984,6 +988,7 @@ pub enum ToolValue {
FlowModule(FlowModuleValue),
Mcp(McpToolValue),
Websearch(WebsearchToolValue),
Question(QuestionToolValue),
}
// Custom deserializer for backward compatibility with old flows
@@ -1002,6 +1007,7 @@ impl<'de> Deserialize<'de> for ToolValue {
FlowModule(FlowModuleValue),
Mcp(McpToolValue),
Websearch(WebsearchToolValue),
Question(QuestionToolValue),
}
if let Ok(tagged) = TaggedToolValue::deserialize(&content) {
@@ -1009,6 +1015,7 @@ impl<'de> Deserialize<'de> for ToolValue {
TaggedToolValue::FlowModule(v) => ToolValue::FlowModule(v),
TaggedToolValue::Mcp(v) => ToolValue::Mcp(v),
TaggedToolValue::Websearch(v) => ToolValue::Websearch(v),
TaggedToolValue::Question(v) => ToolValue::Question(v),
});
}
@@ -1034,6 +1041,14 @@ pub struct McpToolValue {
#[derive(Serialize, Deserialize, Debug, Clone, Default)]
pub struct WebsearchToolValue {}
/// Lets the agent stop and wait for a person to answer a question it asks.
#[derive(Serialize, Deserialize, Debug, Clone, Default)]
pub struct QuestionToolValue {
/// Seconds to wait for an answer before the question is reported to the model as timed out.
#[serde(skip_serializing_if = "Option::is_none")]
pub timeout: Option<u32>,
}
fn is_none_or_empty_vec<T>(expr: &Option<Vec<T>>) -> bool {
expr.is_none() || expr.as_ref().unwrap().is_empty()
}
backend/windmill-worker/src/ai/compaction.rs
+63
View File
@@ -44,11 +44,55 @@ pub(crate) struct AgentHistory {
last_request: Option<(usize, usize)>,
}
/// An `AgentHistory` as a parked agent stores it. `OpenAIMessage` never serializes its
/// `agent_action`, so each message carries it alongside: the result and the flow viewer index
/// actions by it.
#[derive(serde::Serialize, serde::Deserialize)]
pub(crate) struct HistoryCheckpoint {
context: Vec<TaggedMessage>,
result: Vec<TaggedMessage>,
}
#[derive(serde::Serialize, serde::Deserialize)]
struct TaggedMessage {
#[serde(flatten)]
message: OpenAIMessage,
#[serde(default, skip_serializing_if = "Option::is_none")]
agent_action: Option<windmill_common::flow_status::AgentAction>,
}
fn tag(messages: &[OpenAIMessage]) -> Vec<TaggedMessage> {
messages
.iter()
.map(|m| TaggedMessage { message: m.clone(), agent_action: m.agent_action.clone() })
.collect()
}
fn untag(messages: Vec<TaggedMessage>) -> Vec<OpenAIMessage> {
messages
.into_iter()
.map(|t| OpenAIMessage { agent_action: t.agent_action, ..t.message })
.collect()
}
impl AgentHistory {
pub(crate) fn new(messages: Vec<OpenAIMessage>) -> Self {
Self { result: messages.clone(), context: messages, last_request: None }
}
pub(crate) fn checkpoint(&self) -> HistoryCheckpoint {
HistoryCheckpoint { context: tag(&self.context), result: tag(&self.result) }
}
/// No usage measurement survives a park, so the next projection estimates the whole prompt.
pub(crate) fn from_checkpoint(checkpoint: HistoryCheckpoint) -> Self {
Self {
context: untag(checkpoint.context),
result: untag(checkpoint.result),
last_request: None,
}
}
pub(crate) fn context(&self) -> &[OpenAIMessage] {
&self.context
}
@@ -676,6 +720,25 @@ async fn summarize_prefix(
#[cfg(test)]
mod tests {
// `OpenAIMessage` never serializes its action, and a parked agent's result and flow viewer
// index actions by it.
#[test]
fn checkpoint_keeps_agent_actions() {
use windmill_common::flow_status::AgentAction;
let mut tagged = message("assistant", "answer");
tagged.agent_action = Some(AgentAction::Message {});
let history = AgentHistory::new(vec![message("user", "question"), tagged]);
let json = serde_json::to_string(&history.checkpoint()).unwrap();
let restored = AgentHistory::from_checkpoint(serde_json::from_str(&json).unwrap());
let actions: Vec<bool> = restored
.result()
.iter()
.chain(restored.context())
.map(|m| m.agent_action.is_some())
.collect();
assert_eq!(actions, [false, true, false, true]);
}
#[test]
fn unusable_summaries_leave_context_and_usage_untouched() {
let mut history = AgentHistory::new(vec![
backend/windmill-worker/src/ai/mod.rs
+1
View File
@@ -2,6 +2,7 @@
// This module will contain all AI-related execution logic
pub mod compaction;
pub mod pause;
pub mod stream_event_processor;
pub mod tools;
pub mod utils;
backend/windmill-worker/src/ai/pause.rs
+377
View File
@@ -0,0 +1,377 @@
//! A call of a tool that requires approval, or of the built-in question tool, parks the agent
//! instead of running. The agent stores its conversation, releases its worker the way a
//! workflow-as-code parent does (`suspend_wac_parent`), and is pulled again when `suspend` reaches
//! 0 (an answer, or a rejection) or `suspend_until` passes (the request timing out).
//! `resume_suspended` answers it like a WAC approval, since its queue row is the one suspended.
//!
//! The checkpoint row outlives the segment that reads it, so a job restarted after its worker died
//! resumes from it again. It is deleted by the monitor once the job leaves the queue.
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use windmill_ai::{
ai_types::OpenAIToolCall,
types::{OpenAIContent, OpenAIMessage, TokenUsage, Tool, ToolDef, ToolDefFunction},
};
use windmill_common::{
db::DB,
error::{self, Error},
flow_status::{AgentAction, ApprovalCall},
flows::QuestionToolValue,
worker::{to_raw_value, Connection},
};
use windmill_queue::MiniPulledJob;
use crate::ai::compaction::HistoryCheckpoint;
use crate::ai::utils::find_unique_tool_name;
use crate::wac_executor::{end_wac_segment, suspend_wac_parent, WacPark};
/// The name a run enables the question tool by, reserved like `WEBSEARCH_ENABLED_NAME`: the
/// model-facing name may be renamed away from a flow module tool that already holds it.
pub(crate) const QUESTION_ENABLED_NAME: &str = "__wm_question";
/// Matches a suspended flow step's default.
const DEFAULT_TIMEOUT_SECS: u32 = 1800;
pub(crate) fn question_timeout(value: &QuestionToolValue) -> u32 {
value.timeout.unwrap_or(DEFAULT_TIMEOUT_SECS).max(1)
}
pub(crate) fn question_tool(timeout_secs: u32, existing: &[ToolDef]) -> Tool {
let parameters = serde_json::json!({
"type": "object",
"properties": {
"question": {
"type": "string",
"description": "The question, as the person answering will read it."
},
"choices": {
"type": "array",
"items": { "type": "string" },
"description": "Answers to offer, when the likely answers are known. The person \
can still answer in their own words."
},
"multi_select": {
"type": "boolean",
"description": "Whether the person may pick several choices. The answer is then \
a list."
}
},
"required": ["question"],
});
Tool {
def: ToolDef {
r#type: "function".to_string(),
function: ToolDefFunction {
name: find_unique_tool_name("ask_question", Some(existing)),
description: Some(
"Ask a person a question you cannot answer yourself, such as a choice only \
they can make or information you do not have. This run stops until someone \
answers from the flow's run page or chat, or the question times out, then \
resumes with the answer as this tool's result."
.to_string(),
),
parameters: to_raw_value(&parameters),
},
},
module: None,
mcp_source: None,
requires_approval: false,
question_timeout_secs: Some(timeout_secs),
}
}
#[derive(Serialize, Deserialize, Clone, Debug)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub(crate) enum PendingRequest {
/// Calls of tools that require approval, run once approved.
Approval {
calls: Vec<OpenAIToolCall>,
},
Question {
function_name: String,
},
}
/// What the agent waits on, read back to answer it when it is pulled again.
#[derive(Serialize, Deserialize, Clone, Debug)]
pub(crate) struct PendingPause {
/// The question's call, or an approval's first call: the answer must name it.
pub call_id: String,
pub expires_at: DateTime<Utc>,
pub request: PendingRequest,
#[serde(skip)]
pub action: Option<AgentAction>,
}
fn expires_at(timeout_secs: u32) -> DateTime<Utc> {
Utc::now() + chrono::Duration::seconds(timeout_secs as i64)
}
/// The model's arguments turned into a park, or the error the model is answered with instead.
pub(crate) fn start_question(
timeout_secs: u32,
tool_call: &OpenAIToolCall,
) -> Result<PendingPause, String> {
let args: serde_json::Value = serde_json::from_str(&tool_call.function.arguments)
.map_err(|e| format!("Invalid arguments: {e}"))?;
let question = args
.get("question")
.and_then(|v| v.as_str())
.map(str::trim)
.filter(|s| !s.is_empty())
.ok_or("`question` is required")?
.to_string();
let choices = args
.get("choices")
.and_then(|v| v.as_array())
.into_iter()
.flatten()
.filter_map(|c| c.as_str().map(str::trim).filter(|s| !s.is_empty()))
.map(str::to_string)
.collect::<Vec<_>>();
let multi_select = args
.get("multi_select")
.and_then(|v| v.as_bool())
.unwrap_or(false);
let expires_at = expires_at(timeout_secs);
Ok(PendingPause {
call_id: tool_call.id.clone(),
expires_at,
request: PendingRequest::Question { function_name: tool_call.function.name.clone() },
action: Some(AgentAction::Question {
call_id: tool_call.id.clone(),
question,
choices,
multi_select,
expires_at,
}),
})
}
/// One request for all of a turn's calls that require approval: they are answered together.
pub(crate) fn start_approval(calls: Vec<&OpenAIToolCall>) -> PendingPause {
let call_id = calls[0].id.clone();
let expires_at = expires_at(DEFAULT_TIMEOUT_SECS);
let shown = calls
.iter()
.map(|call| ApprovalCall {
function_name: call.function.name.clone(),
arguments: serde_json::from_str(&call.function.arguments)
.unwrap_or_else(|_| serde_json::Value::String(call.function.arguments.clone())),
})
.collect();
PendingPause {
action: Some(AgentAction::Approval { call_id: call_id.clone(), calls: shown, expires_at }),
call_id,
expires_at,
request: PendingRequest::Approval { calls: calls.into_iter().cloned().collect() },
}
}
/// Everything `run_agent` holds across iterations that the next segment needs.
#[derive(Serialize, Deserialize)]
pub(crate) struct AgentCheckpoint {
pub history: HistoryCheckpoint,
pub actions: Vec<AgentAction>,
pub content: Option<OpenAIContent>,
pub usage: Option<TokenUsage>,
pub events: String,
pub reasoning: String,
pub next_iteration: usize,
pub pending: PendingPause,
/// `resume_job` rows never leave while the agent is queued, so each answer must skip the
/// ones earlier parks already read.
#[serde(default)]
pub consumed_resume_rows: Vec<Uuid>,
}
pub(crate) async fn load_checkpoint(
db: &DB,
job_id: &Uuid,
) -> error::Result<Option<AgentCheckpoint>> {
let raw: Option<sqlx::types::Json<Box<serde_json::value::RawValue>>> =
sqlx::query_scalar("SELECT checkpoint FROM ai_agent_checkpoint WHERE job_id = $1")
.bind(job_id)
.fetch_optional(db)
.await?;
raw.map(|r| {
serde_json::from_str(r.0.get())
.map_err(|e| Error::internal_err(format!("Invalid AI agent checkpoint: {e}")))
})
.transpose()
}
pub(crate) enum ParkOutcome {
Parked,
/// A cancel reached the queue row first, so nothing was written.
Cancelled,
}
/// Park the agent until its approval is answered or expires.
pub(crate) async fn park(
db: &DB,
conn: &Connection,
job: &MiniPulledJob,
checkpoint: &AgentCheckpoint,
) -> error::Result<ParkOutcome> {
let remaining_ms = (checkpoint.pending.expires_at - Utc::now()).num_milliseconds();
let suspend_secs = (remaining_ms as f64 / 1000.0).max(1.0);
let checkpoint = serde_json::to_value(checkpoint).map_err(|e| {
Error::internal_err(format!("Failed to serialize AI agent checkpoint: {e}"))
})?;
let mut tx = db.begin().await?;
// One event: an answer, or a rejection, which zeroes `suspend` outright.
let segment_ms =
match suspend_wac_parent(&mut tx, &job.id, &job.workspace_id, 1, suspend_secs).await? {
WacPark::Parked(ms) => ms,
WacPark::Cancelled(_) => return Ok(ParkOutcome::Cancelled),
};
sqlx::query(
"INSERT INTO ai_agent_checkpoint (job_id, workspace_id, checkpoint) VALUES ($1, $2, $3)
ON CONFLICT (job_id) DO UPDATE SET checkpoint = EXCLUDED.checkpoint",
)
.bind(&job.id)
.bind(&job.workspace_id)
.bind(&checkpoint)
.execute(&mut *tx)
.await?;
tx.commit().await?;
end_wac_segment(conn, job, segment_ms);
Ok(ParkOutcome::Parked)
}
pub(crate) enum Resumed {
/// The answer, and whether it counts as a success on the run page.
Answered(Answer, bool),
/// Woken by an answer to an earlier request; the parked one still waits.
StillPending,
}
pub(crate) enum Answer {
/// The tool message answering the question.
Question(OpenAIMessage),
/// The approved calls run.
Run,
/// Each call is refused with this message.
Refuse(String),
}
/// Read the answer to the parked request.
///
/// `suspend` is zeroed in the same transaction as the read, so `resume_suspended` refuses any
/// answer sent while the agent runs instead of storing a row the next request would take. An answer
/// that names another request (the page sends the call id it shows) is read and ignored: a stale
/// page must not answer a request its user never saw.
pub(crate) async fn answer_pause(
db: &DB,
job_id: &Uuid,
pending: &PendingPause,
actions: &[AgentAction],
consumed_resume_rows: &mut Vec<Uuid>,
) -> error::Result<Resumed> {
let mut tx = db.begin().await?;
sqlx::query("UPDATE v2_job_queue SET suspend = 0 WHERE id = $1")
.bind(job_id)
.execute(&mut *tx)
.await?;
let rows: Vec<(
Uuid,
sqlx::types::Json<serde_json::Value>,
Option<String>,
bool,
)> = sqlx::query_as(
"SELECT id, value, approver, approved FROM resume_job
WHERE job = $1 AND id <> ALL($2) ORDER BY created_at ASC",
)
.bind(job_id)
.bind(&*consumed_resume_rows)
.fetch_all(&mut *tx)
.await?;
tx.commit().await?;
consumed_resume_rows.extend(rows.iter().map(|(id, ..)| *id));
let answer = rows.into_iter().find(|(_, value, ..)| {
value
.0
.get("call_id")
.and_then(|v| v.as_str())
.map_or(true, |call_id| call_id == pending.call_id)
});
let Some((_, value, approver, approved)) = answer else {
if Utc::now() < pending.expires_at {
return Ok(Resumed::StillPending);
}
let answer = match pending.request {
PendingRequest::Approval { .. } => Answer::Refuse(
"Not run: nobody approved this call before the request timed out.".to_string(),
),
PendingRequest::Question { .. } => Answer::Question(question_message(
pending,
actions,
serde_json::json!({
"timed_out": true,
"message": "Nobody answered before the question timed out.",
}),
)),
};
return Ok(Resumed::Answered(answer, false));
};
let approver = approver.unwrap_or_else(|| "anonymous".to_string());
let text = |key: &str| {
value
.0
.get(key)
.and_then(|v| v.as_str())
.map(str::trim)
.filter(|s| !s.is_empty())
.map(str::to_string)
};
let answer = match (&pending.request, approved) {
(PendingRequest::Approval { .. }, true) => Answer::Run,
(PendingRequest::Approval { .. }, false) => Answer::Refuse(match text("comment") {
Some(comment) => {
format!("Not run: {approver} rejected this call. Their comment: {comment}")
}
None => format!("Not run: {approver} rejected this call."),
}),
(PendingRequest::Question { .. }, true) => Answer::Question(question_message(
pending,
actions,
serde_json::json!({
// A list when several choices were picked.
"answer": match value.0.get("answer") {
Some(serde_json::Value::Array(picked)) => serde_json::json!(picked),
_ => serde_json::json!(text("answer").unwrap_or_default()),
},
"answered_by": approver,
}),
)),
(PendingRequest::Question { .. }, false) => Answer::Question(question_message(
pending,
actions,
serde_json::json!({ "declined": true, "declined_by": approver }),
)),
};
Ok(Resumed::Answered(answer, approved))
}
fn question_message(
pending: &PendingPause,
actions: &[AgentAction],
result: serde_json::Value,
) -> OpenAIMessage {
let action = actions
.iter()
.rev()
.find(|a| matches!(a, AgentAction::Question { call_id, .. } if *call_id == pending.call_id))
.cloned();
OpenAIMessage {
role: "tool".to_string(),
content: Some(OpenAIContent::Text(result.to_string())),
tool_call_id: Some(pending.call_id.clone()),
agent_action: action,
..Default::default()
}
}
backend/windmill-worker/src/ai/stream_event_processor.rs
+5 -3
View File
@@ -26,14 +26,16 @@ impl Clone for StreamEventProcessor {
}
impl StreamEventProcessor {
/// Create a new StreamEventProcessor that persists events to the database.
pub fn new(conn: &Connection, job: &MiniPulledJob) -> Self {
/// Create a new StreamEventProcessor that persists events to the database, after the
/// `first_offset` events the job already streamed: a row written at an offset already taken
/// is appended to that row, out of order.
pub fn new(conn: &Connection, job: &MiniPulledJob, first_offset: i32) -> Self {
let (tx, mut rx) = tokio::sync::mpsc::channel::<String>(100);
let conn = conn.clone();
let job_id = job.id.clone();
let workspace_id = job.workspace_id.clone();
let handle = tokio::spawn(async move {
let mut offset = -1;
let mut offset = first_offset - 1;
while let Some(event) = rx.recv().await {
offset += 1;
match tokio::time::timeout(
backend/windmill-worker/src/ai/tools.rs
+225 -11
View File
@@ -1,3 +1,4 @@
use crate::ai::pause::{start_approval, start_question, PendingPause};
use crate::ai::stream_event_processor::StreamEventProcessor;
use crate::ai::utils::{
add_message_to_conversation, execute_mcp_tool, get_step_name_from_flow,
@@ -97,12 +98,35 @@ pub async fn execute_tool_calls(
actions: &mut Vec<AgentAction>,
final_events_str: &mut String,
structured_output_tool_name: &Option<String>,
) -> Result<(Vec<OpenAIMessage>, Option<OpenAIContent>, bool), Error> {
gate: ApprovalGate<'_>,
) -> Result<ToolCallsOutcome, Error> {
let mut messages = Vec::new();
let mut used_structured_output_tool = false;
let mut final_content = None;
let mut calls = tool_calls.iter().peekable();
let CallPlan { to_run, refusals, deferred, pause } = plan_calls(
tool_calls,
tools,
&gate,
structured_output_tool_name.as_deref(),
);
// Answered calls were streamed as running when the agent parked.
let announced = matches!(gate, ApprovalGate::Ask);
for call in deferred.iter().chain(refusals.iter().map(|(call, _)| call)) {
if let Some(stream_event_processor) = ctx.stream_event_processor.filter(|_| announced) {
let event = StreamingEvent::ToolExecution {
call_id: call.id.clone(),
function_name: call.function.name.clone(),
};
stream_event_processor.send(event, final_events_str).await?;
}
}
for (call, refusal) in refusals {
refuse_tool_call(&mut ctx, call, &refusal, &mut messages, final_events_str).await?;
}
let mut calls = to_run.into_iter().peekable();
while let Some(tool_call) = calls.next() {
// Stream tool call progress
if let Some(stream_event_processor) = ctx.stream_event_processor {
@@ -142,7 +166,6 @@ pub async fn execute_tool_calls(
.find(|t| t.def.function.name == tool_call.function.name);
if let Some(tool) = tool {
// Check if this is an MCP tool
if let Some(mcp_source) = &tool.mcp_source {
execute_mcp_tool_call(
&mut ctx,
@@ -155,7 +178,17 @@ pub async fn execute_tool_calls(
)
.await?;
} else if tool.module.is_some() && interpolating_args(tool_call) {
refuse_tool_call(&mut ctx, tool_call, &mut messages, final_events_str).await?;
let error_msg = "Tool call refused: its arguments could not be read, or reference \
variables, resources or reserved names ($var:, $jsonvar:, \
$res:, $encrypted:, $WM_)";
refuse_tool_call(
&mut ctx,
tool_call,
error_msg,
&mut messages,
final_events_str,
)
.await?;
} else if tool.module.is_some() {
let mut batch = vec![(tool_call, tool)];
while let Some(next_call) = calls.peek() {
@@ -192,7 +225,114 @@ pub async fn execute_tool_calls(
}
}
Ok((messages, final_content, used_structured_output_tool))
Ok(ToolCallsOutcome { messages, final_content, used_structured_output_tool, pause })
}
/// A turn's calls: run now, refused with a message, or held for a person's answer.
struct CallPlan<'a> {
to_run: Vec<&'a OpenAIToolCall>,
refusals: Vec<(&'a OpenAIToolCall, String)>,
deferred: Vec<&'a OpenAIToolCall>,
pause: Option<PendingPause>,
}
fn plan_calls<'a>(
tool_calls: &'a [OpenAIToolCall],
tools: &[Tool],
gate: &ApprovalGate<'_>,
structured_output_tool_name: Option<&str>,
) -> CallPlan<'a> {
let tool_of = |call: &OpenAIToolCall| {
tools
.iter()
.find(|t| t.def.function.name == call.function.name)
};
let mut plan = CallPlan { to_run: vec![], refusals: vec![], deferred: vec![], pause: None };
match gate {
ApprovalGate::Answered(Some(refusal)) => plan
.refusals
.extend(tool_calls.iter().map(|call| (call, refusal.to_string()))),
ApprovalGate::Answered(None) => {
for call in tool_calls {
match tool_of(call) {
Some(_) => plan.to_run.push(call),
None => plan
.refusals
.push((call, "Not run: this tool is no longer available.".into())),
}
}
}
// Calls that need a person wait for the answer; every other call runs now.
ApprovalGate::Ask => {
// A call the run would refuse anyway is refused now rather than shown to the approver.
let held = |call: &OpenAIToolCall| {
tool_of(call).is_some_and(|t| {
t.requires_approval && !(t.module.is_some() && interpolating_args(call))
})
};
let approval_calls: Vec<_> = tool_calls.iter().filter(|call| held(call)).collect();
for call in tool_calls {
match tool_of(call) {
_ if held(call) => {}
Some(Tool { question_timeout_secs: Some(timeout_secs), .. }) => {
let refusal = if !approval_calls.is_empty() {
"Not run: ask it again once the pending approval is answered."
} else if plan.pause.is_some() {
"Not run: ask one question at a time."
} else {
match start_question(*timeout_secs, call) {
Ok(question) => {
plan.pause = Some(question);
plan.deferred.push(call);
continue;
}
Err(e) => {
plan.refusals.push((call, e));
continue;
}
}
};
plan.refusals.push((call, refusal.to_string()));
}
_ => plan.to_run.push(call),
}
}
if !approval_calls.is_empty() {
plan.deferred.extend(approval_calls.iter().copied());
plan.pause = Some(start_approval(approval_calls));
}
// The structured answer ends the run, so it cannot come before the answer it waits on.
if plan.pause.is_some() {
if let Some(at) = plan
.to_run
.iter()
.position(|c| Some(c.function.name.as_str()) == structured_output_tool_name)
{
plan.refusals.push((
plan.to_run.remove(at),
"Not run: give your answer once the pending request is answered.".into(),
));
}
}
}
}
plan
}
pub struct ToolCallsOutcome {
pub messages: Vec<OpenAIMessage>,
pub final_content: Option<OpenAIContent>,
pub used_structured_output_tool: bool,
/// The turn asked a person something: the agent parks, those calls unanswered.
pub pause: Option<PendingPause>,
}
pub enum ApprovalGate<'a> {
/// Park on calls of tools that require approval, and on a question.
Ask,
/// A person answered an approval request for these calls: run them, or refuse each with
/// this message.
Answered(Option<&'a str>),
}
/// Whether the model wrote an argument the tool's job would interpolate with the agent's token,
@@ -207,15 +347,13 @@ fn interpolating_args(tool_call: &OpenAIToolCall) -> bool {
async fn refuse_tool_call(
ctx: &mut ToolExecutionContext<'_>,
tool_call: &OpenAIToolCall,
error_msg: &str,
messages: &mut Vec<OpenAIMessage>,
final_events_str: &mut String,
) -> Result<(), Error> {
let error_msg = "Tool call refused: its arguments could not be read, or reference variables, \
resources or reserved names ($var:, $jsonvar:, $res:, $encrypted:, $WM_)"
.to_string();
messages.push(OpenAIMessage {
role: "tool".to_string(),
content: Some(OpenAIContent::Text(error_msg.clone())),
content: Some(OpenAIContent::Text(error_msg.to_string())),
tool_call_id: Some(tool_call.id.clone()),
..Default::default()
});
@@ -223,12 +361,12 @@ async fn refuse_tool_call(
let event = StreamingEvent::ToolResult {
call_id: tool_call.id.clone(),
function_name: tool_call.function.name.clone(),
result: error_msg.clone(),
result: error_msg.to_string(),
success: false,
};
stream_event_processor.send(event, final_events_str).await?;
}
let (content, extras) = windmill_tool_row(tool_call, false, &error_msg);
let (content, extras) = windmill_tool_row(tool_call, false, error_msg);
let agent_job_id = ctx.job.id;
add_tool_message_to_chat(ctx, Some(agent_job_id), &content, false, Some(extras)).await;
Ok(())
@@ -1134,6 +1272,82 @@ mod tests {
);
}
#[test]
fn only_calls_that_need_a_person_wait_for_one() {
use super::{plan_calls, ApprovalGate};
use crate::ai::pause::PendingRequest;
use windmill_ai::types::{Tool, ToolDef, ToolDefFunction};
let tool = |name: &str, requires_approval: bool, question: bool| Tool {
module: None,
def: ToolDef {
r#type: "function".to_string(),
function: ToolDefFunction {
name: name.to_string(),
description: None,
parameters: RawValue::from_string("{}".to_string()).unwrap(),
},
},
mcp_source: None,
requires_approval,
question_timeout_secs: question.then_some(60),
};
let call = |id: &str, name: &str| OpenAIToolCall {
id: id.to_string(),
function: OpenAIFunction {
name: name.to_string(),
arguments: r#"{"question":"which one?"}"#.to_string(),
},
r#type: "function".to_string(),
extra_content: None,
};
let tools = [
tool("list", false, false),
tool("delete", true, false),
tool("ask", false, true),
];
let ids =
|calls: &[&OpenAIToolCall]| calls.iter().map(|c| c.id.clone()).collect::<Vec<_>>();
let calls = [
call("1", "list"),
call("2", "delete"),
call("3", "ask"),
call("4", "delete"),
call("5", "answer"),
];
let plan = plan_calls(&calls, &tools, &ApprovalGate::Ask, Some("answer"));
assert_eq!(ids(&plan.to_run), ["1"]);
assert_eq!(ids(&plan.deferred), ["2", "4"]);
assert_eq!(
plan.refusals
.iter()
.map(|(c, _)| c.id.as_str())
.collect::<Vec<_>>(),
["3", "5"]
);
let pause = plan.pause.unwrap();
assert_eq!(pause.call_id, "2");
assert!(matches!(pause.request, PendingRequest::Approval { calls } if calls.len() == 2));
let calls = [call("1", "ask"), call("2", "list"), call("3", "ask")];
let plan = plan_calls(&calls, &tools, &ApprovalGate::Ask, Some("answer"));
assert_eq!(ids(&plan.to_run), ["2"]);
assert_eq!(ids(&plan.deferred), ["1"]);
assert_eq!(plan.refusals.len(), 1);
let calls = [call("1", "delete"), call("2", "gone")];
let plan = plan_calls(&calls, &tools, &ApprovalGate::Answered(None), None);
assert_eq!(ids(&plan.to_run), ["1"]);
assert_eq!(plan.refusals[0].0.id, "2");
let plan = plan_calls(
&calls,
&tools,
&ApprovalGate::Answered(Some("rejected")),
None,
);
assert!(plan.to_run.is_empty() && plan.refusals.len() == 2);
}
#[test]
fn a_model_written_reference_is_caught_however_it_is_spelled() {
let call = |arguments: &str| OpenAIToolCall {
backend/windmill-worker/src/ai/utils.rs
+36
View File
@@ -307,6 +307,40 @@ pub async fn update_flow_status_module_with_actions_success(
Ok(())
}
/// Record the outcome of the last of `action_count` actions, dropping any recorded past it.
pub async fn set_flow_status_module_last_action_success(
db: &DB,
parent_job: &Uuid,
action_count: usize,
action_success: bool,
) -> Result<(), Error> {
if let Step::Step { idx: step, .. } = get_step_of_flow_status(db, parent_job.to_owned()).await?
{
sqlx::query(
"UPDATE v2_job_status SET
flow_status = jsonb_set(
flow_status,
array['modules', $2::TEXT, 'agent_actions_success'],
COALESCE((
SELECT jsonb_agg(e ORDER BY i)
FROM jsonb_array_elements(
flow_status->'modules'->$2->'agent_actions_success'
) WITH ORDINALITY AS t(e, i)
WHERE i < $3
), '[]'::jsonb) || to_jsonb(ARRAY[$4::bool])
)
WHERE id = $1",
)
.bind(parent_job)
.bind(step as i32)
.bind(action_count as i64)
.bind(action_success)
.execute(db)
.await?;
}
Ok(())
}
/// Get step name from the flow module (summary if exists, else id)
pub fn get_step_name_from_flow(
summary: Option<&str>,
@@ -405,6 +439,8 @@ fn convert_mcp_tools_to_windmill_tools(
tool_name: mcp_tool.name.to_string(),
resource_path: resource_path.to_string(),
}),
requires_approval: false,
question_timeout_secs: None,
})
})
.collect()
backend/windmill-worker/src/ai_executor.rs
+272 -30
View File
@@ -2,13 +2,19 @@ use crate::ai::compaction::{
memory_within_capacity, persisted_bytes, AgentHistory, CompactionRequest, CompactionTrigger,
Compactor,
};
use crate::ai::tools::{execute_tool_calls, ToolAbortHandles, ToolExecutionContext};
use crate::ai::pause::{
answer_pause, load_checkpoint, park, question_timeout, question_tool, AgentCheckpoint, Answer,
ParkOutcome, PendingRequest, Resumed, QUESTION_ENABLED_NAME,
};
use crate::ai::tools::{
execute_tool_calls, ApprovalGate, ToolAbortHandles, ToolCallsOutcome, ToolExecutionContext,
};
use crate::ai::utils::{
add_message_to_conversation, any_tool_needs_previous_result, cleanup_mcp_clients,
filter_schema_by_input_transforms, find_unique_tool_name, get_flow_context,
get_flow_job_runnable_and_raw_flow, get_step_name_from_flow, load_mcp_tools,
parse_raw_script_schema, update_flow_status_module_with_actions,
update_flow_status_module_with_actions_success,
parse_raw_script_schema, set_flow_status_module_last_action_success,
update_flow_status_module_with_actions, update_flow_status_module_with_actions_success,
};
use crate::memory_common::MAX_MEMORY_SIZE_BYTES;
use crate::memory_oss::{memory_storage_capacity_bytes, read_from_memory, write_to_memory};
@@ -17,7 +23,10 @@ use async_recursion::async_recursion;
use regex::Regex;
use serde_json::value::RawValue;
use sha2::Digest;
use std::{collections::HashMap, sync::Arc};
use std::{
collections::{HashMap, HashSet},
sync::Arc,
};
use uuid::Uuid;
#[cfg(feature = "bedrock")]
use windmill_ai::ai_bedrock::check_env_credentials;
@@ -42,7 +51,7 @@ use windmill_ai::{
proxy::{
common_outbound_headers, needs_unavailable_oauth_exchange, retain_effective_credentials,
},
query_builder::{BuildRequestArgs, ParsedResponse},
query_builder::{BuildRequestArgs, ParsedResponse, StreamEventSink},
retry::{as_transient, into_final_error, response_error, send_error, Backoff, MAX_RETRIES},
types::*,
utils::{pinned_ai_client_for, should_use_structured_output_tool},
@@ -585,6 +594,9 @@ fn flow_module_tool_name(summary: Option<&str>) -> Result<&str, Error> {
Some(name) if name == WEBSEARCH_ENABLED_NAME => Err(Error::internal_err(format!(
"Invalid tool name: {name:?} is reserved for enabling web search"
))),
Some(name) if name == QUESTION_ENABLED_NAME => Err(Error::internal_err(format!(
"Invalid tool name: {name:?} is reserved for the question tool"
))),
Some(name) if TOOL_NAME_REGEX.is_match(name) => Ok(name),
other => Err(Error::internal_err(format!("Invalid tool name: {other:?}"))),
}
@@ -601,6 +613,7 @@ fn tool_enabled_name(tool: &AgentTool) -> Option<&str> {
match &tool.value {
ToolValue::Mcp(mcp) => Some(mcp.resource_path.trim_start_matches("$res:")),
ToolValue::Websearch(_) => Some(WEBSEARCH_ENABLED_NAME),
ToolValue::Question(_) => Some(QUESTION_ENABLED_NAME),
_ => tool.summary.as_deref(),
}
}
@@ -898,6 +911,23 @@ pub async fn handle_ai_agent_job(
// constrain a shared resource); don't advertise them — invoking one would only fail the depth
// check as a third level.
let tools = if direct_parent_job_kind == JobKind::AIAgent {
// A nested agent writes no flow status, so nothing would show a question or an approval
// request, and its parent agent holds a worker while it waits: it cannot park either. A
// tool that requires approval is dropped rather than run unapproved.
let (dropped, tools): (Vec<_>, Vec<_>) =
tools.into_iter().partition(|t| t.requires_approval);
if !dropped.is_empty() {
let names: Vec<&str> = dropped
.iter()
.map(|t| tool_enabled_name(t).unwrap_or(&t.id))
.collect();
let message = format!(
"Tools that require approval are left out of an agent used as a tool, which cannot \
wait for an answer: {}\n",
names.join(", ")
);
append_logs(&job.id, &job.workspace_id, message, conn).await;
}
tools
.into_iter()
.filter(|t| {
@@ -905,7 +935,7 @@ pub async fn handle_ai_agent_job(
&t.value,
ToolValue::FlowModule(
FlowModuleValue::AIAgent { .. } | FlowModuleValue::AIDecision { .. }
)
) | ToolValue::Question(_)
)
})
.collect()
@@ -935,8 +965,17 @@ pub async fn handle_ai_agent_job(
#[allow(unused_mut)]
let mut mcp_configs: Vec<crate::ai::utils::McpResourceConfig> = Vec::new();
let mut has_websearch = false;
let mut question_timeout_secs = None;
// Flow module tool ids and MCP resource paths whose calls each wait for approval.
let mut approval_required: HashSet<String> = HashSet::new();
for tool in tools {
if tool.requires_approval {
approval_required.insert(match &tool.value {
ToolValue::Mcp(mcp) => mcp.resource_path.trim_start_matches("$res:").to_string(),
_ => tool.id.clone(),
});
}
match &tool.value {
#[allow(unused_variables)]
ToolValue::Mcp(mcp_config) => {
@@ -981,6 +1020,9 @@ pub async fn handle_ai_agent_job(
tracing::debug!("WebSearch tool enabled");
has_websearch = true;
}
ToolValue::Question(question) => {
question_timeout_secs = Some(question_timeout(question));
}
}
}
@@ -990,6 +1032,7 @@ pub async fn handle_ai_agent_job(
let db = db;
let job = job;
let user_description = tool_descriptions.get(&t.id).cloned();
let requires_approval = approval_required.contains(&t.id);
async move {
let summary = flow_module_tool_name(t.summary.as_deref())?;
@@ -1124,8 +1167,10 @@ pub async fn handle_ai_agent_job(
}),
},
},
requires_approval,
module: Some(t),
mcp_source: None,
question_timeout_secs: None,
})
}
}))
@@ -1137,11 +1182,21 @@ pub async fn handle_ai_agent_job(
let mcp_clients = if !mcp_configs.is_empty() {
let (clients, mcp_tools) =
load_mcp_tools(db, &job.workspace_id, mcp_configs, client).await?;
tools.extend(mcp_tools);
tools.extend(mcp_tools.into_iter().map(|mut tool| {
tool.requires_approval = tool
.mcp_source
.as_ref()
.is_some_and(|s| approval_required.contains(&s.resource_path));
tool
}));
clients
} else {
HashMap::new()
};
if let Some(timeout_secs) = question_timeout_secs {
let defs: Vec<ToolDef> = tools.iter().map(|t| t.def.clone()).collect();
tools.push(question_tool(timeout_secs, &defs));
}
if let Some(enabled) = enabled_tools {
let matchable: Vec<&str> = roster_names.iter().map(|s| s.as_str()).collect();
@@ -1265,7 +1320,10 @@ pub async fn handle_ai_agent_job(
}
},
};
if result.is_err() {
if result
.as_ref()
.is_err_and(|e| !matches!(e, Error::WacSuspended(_)))
{
for handle in tool_abort_handles.lock().unwrap().drain(..) {
handle.abort();
}
@@ -1275,6 +1333,34 @@ pub async fn handle_ai_agent_job(
result
}
/// Park the agent on its pending approval, as the error `run_agent` returns.
async fn park_agent(
db: &DB,
conn: &Connection,
job: &MiniPulledJob,
checkpoint: &AgentCheckpoint,
cancel_rx: &tokio::sync::watch::Receiver<bool>,
) -> Error {
match park(db, conn, job, checkpoint).await {
Ok(ParkOutcome::Parked) => Error::WacSuspended(format!(
"AI agent {} parked on {}",
job.id, checkpoint.pending.call_id
)),
Ok(ParkOutcome::Cancelled) => {
// The cancel is on the queue row but not yet seen by the poller, which is what
// completes the job as cancelled rather than failed. Wait for it to notice.
let mut cancel_rx = cancel_rx.clone();
let _ = tokio::time::timeout(
std::time::Duration::from_secs(30),
cancel_rx.wait_for(|cancelled| *cancelled),
)
.await;
Error::ExecutionErr("Job cancelled".to_string())
}
Err(e) => e,
}
}
/// OpenAI rejects a `prompt_cache_key` over 64 characters
/// (`Invalid 'prompt_cache_key': string too long`), and a runnable path alone can pass
/// that. Fold an over-long key into a digest of itself: same step still yields the same
@@ -1380,6 +1466,27 @@ pub async fn run_agent(
// Fetch flow context for input transforms context, chat and memory
let mut flow_context = get_flow_context(db, job).await;
// Set when the job parked on an approval or a question and has been pulled again. Read
// whatever the roster says now: a linked agent can lose a tool while one of its runs is parked.
let resumed = match load_checkpoint(db, &job.id).await? {
Some(mut checkpoint) => {
let answer = answer_pause(
db,
&job.id,
&checkpoint.pending,
&checkpoint.actions,
&mut checkpoint.consumed_resume_rows,
)
.await?;
match answer {
Resumed::StillPending => {
return Err(park_agent(db, conn, job, &checkpoint, &cancel_rx).await)
}
Resumed::Answered(message, success) => Some((checkpoint, message, success)),
}
}
None => None,
};
// The run's memory id is also the chat conversation id, which a step's own memory id never
// replaces.
@@ -1437,7 +1544,7 @@ pub async fn run_agent(
return Err(Error::internal_err(missing.to_string()));
}
if matches!(output_type, OutputType::Text) {
if matches!(output_type, OutputType::Text) && resumed.is_none() {
match &history {
HistorySource::Messages(provided) => messages.extend(provided.iter().cloned()),
HistorySource::Managed { memory_id, bound } => {
@@ -1538,10 +1645,27 @@ pub async fn run_agent(
}
}
let mut messages = AgentHistory::new(messages);
let mut actions = vec![];
let mut content = None;
let mut final_usage: Option<TokenUsage> = None;
let mut final_events_str = String::new();
let mut final_reasoning = String::new();
let mut first_iteration = 0;
let mut consumed_resume_rows = vec![];
let mut resumed_answer = None;
let (mut messages, mut actions, mut content, mut final_usage) = match resumed {
Some((checkpoint, message, success)) => {
final_events_str = checkpoint.events;
final_reasoning = checkpoint.reasoning;
first_iteration = checkpoint.next_iteration;
consumed_resume_rows = checkpoint.consumed_resume_rows;
resumed_answer = Some((checkpoint.pending, message, success));
(
AgentHistory::from_checkpoint(checkpoint.history),
checkpoint.actions,
checkpoint.content,
checkpoint.usage,
)
}
None => (AgentHistory::new(messages), vec![], None, None),
};
// Check if this provider supports tools with the current output type
let supports_tools = query_builder.supports_tools_with_output_type(output_type);
@@ -1596,13 +1720,12 @@ pub async fn run_agent(
let user_wants_streaming = streaming_requested(args.streaming);
*has_stream = user_wants_streaming && is_text_output;
let mut final_events_str = String::new();
let mut final_reasoning = String::new();
// Always create a StreamEventProcessor for text output (use silent mode if user doesn't want streaming)
let stream_event_processor = if is_text_output {
let mut stream_event_processor = if is_text_output {
if user_wants_streaming {
Some(StreamEventProcessor::new(conn, job))
// Each streamed event is one line of the accumulated string.
let streamed = final_events_str.lines().count() as i32;
Some(StreamEventProcessor::new(conn, job, streamed))
} else {
Some(StreamEventProcessor::new_silent())
}
@@ -1670,8 +1793,81 @@ pub async fn run_agent(
windmill_ai::providers::anthropic::FederatedAccessToken,
> = None;
if let Some((pending, answer, success)) = resumed_answer {
if let Some(parent_job) = parent_job {
// Rewritten whole rather than appended: a job restarted after its worker died
// resumes from the same checkpoint and must not record the answer twice.
update_flow_status_module_with_actions(db, parent_job, &actions).await?;
set_flow_status_module_last_action_success(db, parent_job, actions.len(), success)
.await?;
}
match (pending.request, answer) {
(PendingRequest::Question { function_name }, Answer::Question(message)) => {
if let (Some(processor), Some(OpenAIContent::Text(result))) =
(stream_event_processor.as_ref(), message.content.as_ref())
{
let event = StreamingEvent::ToolResult {
call_id: pending.call_id.clone(),
function_name,
result: result.clone(),
success,
};
processor.send(event, &mut final_events_str).await?;
}
messages.push(message);
}
(PendingRequest::Approval { calls }, answer) => {
let refusal = match &answer {
Answer::Refuse(refusal) => Some(refusal.as_str()),
_ => None,
};
let tool_execution_ctx = ToolExecutionContext {
db,
conn,
job,
parent_job,
summary: &summary,
flow_step_id_override,
client,
worker_dir,
base_internal_url,
worker_name,
hostname,
occupancy_metrics,
killpill_rx,
stream_event_processor: stream_event_processor.as_ref(),
flow_context: &mut flow_context,
omit_output_from_conversation,
preserve_step_tags,
reasoning: None,
previous_result: &previous_result,
id_context: &id_context,
tool_abort_handles: tool_abort_handles.clone(),
job_completed_tx: job_completed_tx.clone(),
};
let outcome = execute_tool_calls(
tool_execution_ctx,
&calls,
&tools,
mcp_clients,
&mut actions,
&mut final_events_str,
&structured_output_tool_name,
ApprovalGate::Answered(refusal),
)
.await?;
messages.extend(outcome.messages);
}
(PendingRequest::Question { .. }, _) => {
return Err(Error::internal_err(
"A question was answered as an approval".to_string(),
))
}
}
}
// Main agent loop
for i in 0..max_iterations {
for i in first_iteration..max_iterations {
// Check if parent was canceled — stop iterating but let current tool calls finish
if *cancel_rx.borrow() {
return Err(Error::ExecutionErr("Job cancelled".to_string()));
@@ -2277,17 +2473,22 @@ pub async fn run_agent(
job_completed_tx: job_completed_tx.clone(),
};
let (tool_messages, tool_content, tool_used_structured_output) =
execute_tool_calls(
tool_execution_ctx,
&tool_calls,
&tools,
mcp_clients,
&mut actions,
&mut final_events_str,
&structured_output_tool_name,
)
.await?;
let ToolCallsOutcome {
messages: tool_messages,
final_content: tool_content,
used_structured_output_tool: tool_used_structured_output,
pause,
} = execute_tool_calls(
tool_execution_ctx,
&tool_calls,
&tools,
mcp_clients,
&mut actions,
&mut final_events_str,
&structured_output_tool_name,
ApprovalGate::Ask,
)
.await?;
messages.extend(tool_messages);
@@ -2331,6 +2532,42 @@ pub async fn run_agent(
}
used_structured_output_tool = tool_used_structured_output;
if let Some(mut pending) = pause {
actions.extend(pending.action.take());
// The next segment streams from where this one stops, so this one's events
// must be written first.
if let Some(handle) = stream_event_processor.take().and_then(|p| p.to_handle())
{
let _ = handle.await;
}
let checkpoint = AgentCheckpoint {
history: messages.checkpoint(),
actions: actions.clone(),
content: content.clone(),
usage: final_usage.clone(),
events: final_events_str,
reasoning: final_reasoning,
next_iteration: i + 1,
pending,
consumed_resume_rows,
};
let parked = park_agent(db, conn, job, &checkpoint, &cancel_rx).await;
// Written once parked: the run page offers an answer only for a request that
// `resume_suspended` can take. The park is committed, so a failure here only
// costs the page its banner.
if let (Error::WacSuspended(_), Some(parent_job)) = (&parked, parent_job) {
if let Err(e) =
update_flow_status_module_with_actions(db, parent_job, &actions).await
{
tracing::error!(
"Failed to show the pending request of {}: {e}",
job.id
);
}
}
return Err(parked);
}
// Check cancellation after tool calls complete to avoid a wasted LLM call
if *cancel_rx.borrow() {
return Err(Error::ExecutionErr("Job cancelled".to_string()));
@@ -3068,6 +3305,7 @@ mod tests {
its.insert(key.to_string(), js(expr));
AgentTool {
job_token_scopes: None,
requires_approval: false,
id: id.to_string(),
summary: None,
description: None,
@@ -3096,6 +3334,7 @@ mod tests {
script_tool("b", "y", "keep_me"),
AgentTool {
job_token_scopes: None,
requires_approval: false,
id: "m".to_string(),
summary: None,
description: None,
@@ -3144,6 +3383,7 @@ mod tests {
fn named(id: &str, summary: &str) -> AgentTool {
AgentTool {
job_token_scopes: None,
requires_approval: false,
id: id.to_string(),
summary: Some(summary.to_string()),
description: None,
@@ -3160,6 +3400,7 @@ mod tests {
fn mcp(id: &str, summary: &str, path: &str) -> AgentTool {
AgentTool {
job_token_scopes: None,
requires_approval: false,
id: id.to_string(),
summary: Some(summary.to_string()),
description: None,
@@ -3173,6 +3414,7 @@ mod tests {
fn websearch(id: &str, summary: Option<&str>) -> AgentTool {
AgentTool {
job_token_scopes: None,
requires_approval: false,
id: id.to_string(),
summary: summary.map(str::to_string),
description: None,
chat-sdk/README.md
+1
View File
@@ -244,6 +244,7 @@ interface ChatState {
history: 'server' | 'local' | 'none'
loadingMessages: boolean
hasMoreMessages: boolean
runningJobId?: string // the flow job the running turn follows
}
interface ChatMessage {
chat-sdk/src/chat.ts
+2
View File
@@ -305,6 +305,7 @@ class ChatImpl implements Chat {
async #followTurn(turn: Turn, isNew: boolean): Promise<void> {
let nextTurn: RunningTurn | undefined
this.#set({ runningJobId: turn.jobId })
try {
const stopPolling = this.#state.history === 'server' ? this.#startPolling(turn) : () => {}
let result: unknown
@@ -327,6 +328,7 @@ class ChatImpl implements Chat {
this.#failTurn(turn, e)
} finally {
if (this.#turn === turn) this.#turn = undefined
if (this.#state.runningJobId === turn.jobId) this.#set({ runningJobId: undefined })
}
if (nextTurn) await this.resumeTurn(nextTurn)
}
chat-sdk/src/types.ts
+2
View File
@@ -80,6 +80,8 @@ export interface ChatState {
history: HistoryMode
loadingMessages: boolean
hasMoreMessages: boolean
/** The flow job the running turn follows, once it is known. */
runningJobId?: string
}
export type TokenSource = string | (() => string | Promise<string>)
cli/src/commands/lint/flow_semantics.ts
+11 -5
View File
@@ -9,6 +9,12 @@
// frontend/src/lib/components/flows/agentToolUtils.ts applies the same rules.
const TOOL_NAME_REGEX = /^[a-zA-Z0-9_]+$/;
const WEBSEARCH_ENABLED_NAME = "__wm_web_search";
// The worker's `*_ENABLED_NAME`s: built-in tools, each named by its reserved name.
const BUILTIN_ENABLED_NAMES = new Set([
WEBSEARCH_ENABLED_NAME,
"__wm_question",
]);
const BUILTIN_TOOL_TYPES = new Set(["websearch", "question"]);
// `forbiddenIds` in frontend/src/lib/components/flows/idUtils.ts.
const RESERVED_IDS = new Set([
@@ -134,10 +140,10 @@ function checkAgentTools(
if (!Array.isArray(tools)) return;
const seen = new Map<string, number>();
tools.forEach((tool: any, i: number) => {
// An mcp entry exposes its server's own tool names and a websearch entry is
// named by `WEBSEARCH_ENABLED_NAME`; neither summary reaches the model.
// An mcp entry exposes its server's own tool names and a built-in entry is
// named by its reserved name; neither summary reaches the model.
const kind = tool?.value?.tool_type;
if (kind === "mcp" || kind === "websearch") return;
if (kind === "mcp" || BUILTIN_TOOL_TYPES.has(kind)) return;
const at = `${agent.pointer}/value/tools/${i}`;
const name = tool?.summary;
if (typeof name !== "string" || name.length === 0) {
@@ -152,9 +158,9 @@ function checkAgentTools(
);
return;
}
if (name === WEBSEARCH_ENABLED_NAME) {
if (BUILTIN_ENABLED_NAMES.has(name)) {
report.errors.push(
`${at} tool name '${name}' is reserved for enabling web search`,
`${at} tool name '${name}' is reserved for a built-in tool`,
);
return;
}
cli/src/guidance/skills.gen.ts
Generated
+8 -5
View File
File diff suppressed because one or more lines are too long
docs/reusable-ai-agents.md
+45
View File
@@ -156,6 +156,51 @@ older editor baked into `memory` stays a fallback behind the run's id until the
*Keep as memory id* or *Use the run's memory id*. In a chat flow it is dropped on save, since the
conversation id always took precedence there.
## Approval and questions
Two things park the agent rather than run: a call of a tool with `requires_approval` (flow module
and MCP tools), and a call of the built-in `question` tool, which the model sees as `ask_question`.
Both live in `windmill-worker/src/ai/pause.rs`.
- **Park.** In a turn that calls such tools, every other call runs right away. The calls that
require approval are gathered into one request, answered together; a question is its own request,
so a turn that also asks for approval has its question refused, as is a second question. The
worker stores the conversation, actions, usage, accumulated stream, next iteration and the
waiting calls in `ai_agent_checkpoint`, then parks the agent's own queue row with
`suspend_wac_parent`: `suspend = 1`, `suspend_until` at the timeout (1800s for an approval, the
question tool's `timeout` otherwise). The job is not completed, its worker is free, and the flow
stays on the step.
- **Answer.** `resume_suspended` with the agent job's id already treats a self-suspended job like a
WAC approval: it inserts the `resume_job` row and decrements, or zeroes on a rejection, the
agent's own `suspend`. The run page offers the form from the pending `approval` or `question`
agent action (Approve/Reject with the calls and their arguments, or the question with its
`choices` and a text answer, sent as a list when `multi_select`), and so does a chat flow's chat,
on the waiting call's tool card while the turn's top-level agent step waits (it polls the turn's
flow job only when a top-level agent step has such a tool or links a saved agent, read off the
flow rather than the stream, which a non-streaming agent leaves empty: the card is then added at
the end of the turn). Both send that request's `call_id` in the payload. There are no signed resume URLs, Slack or Teams messages,
approval page or approver conditions for it: whoever can answer a suspended step without
conditions can answer it.
- **Resume.** The suspended pull takes the row back and `run_agent` finds the checkpoint. In one
transaction it zeroes `suspend`, so `resume_suspended` refuses answers while the agent runs, and
reads the unread `resume_job` rows. The first one naming the parked request (or naming none)
answers it; one naming an earlier request, from a page left open, is read and ignored, and the
agent parks again for the time left. Approved calls then run as any call would; rejected ones,
and those still unanswered at the timeout, are answered as not run, with the approver's comment.
A call whose tool left the roster while parked is answered as no longer available. The agent
then skips memory and the prompt and continues the loop. The step's stream continues at the
offset the first segment reached. Resolution is live here too: a linked agent edited while
parked resumes with its new definition.
- **Cleanup.** The checkpoint row stays while the job is queued, so a job restarted after its
worker died resumes from it, and reads the same answer again: approved calls that ran before
the worker died run again. The monitor deletes rows whose job left the queue; the table has no
foreign key, since one would lock `v2_job_queue` on every write.
- **Limits.** A nested agent tool is never offered the question tool or a tool that requires
approval: it writes no flow status, and its parent holds a worker while it runs. The banner at the
top of the run page shows only a request from a top-level step, as does the chat; one from a step
inside a loop or branch is answered from its graph node. The chat transcript of a chat flow
records neither the request nor its answer.
## Drafts
The agent editor edits the resource through a **per-user resource draft** (`draft` table,
frontend/src/lib/components/AgentApprovalRequest.svelte
+122
View File
@@ -0,0 +1,122 @@
<script lang="ts">
import { answerAgentRequest, type AgentRequest } from './agentRequest'
import { Button } from './common'
import TextInput from './text_input/TextInput.svelte'
import ApprovalActions from './approvals/ApprovalActions.svelte'
import { displayDate } from '$lib/utils'
interface Props {
/** The workspace the job ran in, never the navigation workspace. */
workspaceId: string
/** The agent step's own job: it is the one parked, so it is the one answered. */
agentJobId: string
/** Its call id goes with the answer, so a page left open on an earlier request cannot
* answer the one the agent waits on now. */
request: AgentRequest
/** Whether the request still waits for an answer. */
pending: boolean
}
let { workspaceId, agentJobId, request, pending }: Props = $props()
let text = $state('')
let picked = $state<string[]>([])
let answered = $state<boolean | undefined>(undefined)
const choices = $derived(request.type === 'question' ? (request.choices ?? []) : [])
const multiSelect = $derived(request.type === 'question' && !!request.multi_select)
// A typed answer joins the picked choices when several can be picked, and replaces the pick
// otherwise.
const answer = $derived.by((): string | string[] => {
const typed = text.trim()
if (multiSelect) return typed ? [...picked, typed] : picked
return typed || picked[0] || ''
})
function pick(choice: string) {
if (multiSelect) {
picked = picked.includes(choice) ? picked.filter((c) => c !== choice) : [...picked, choice]
} else {
picked = [choice]
text = ''
}
}
async function onAnswer(approved: boolean) {
const value = request.type === 'question' ? answer : text.trim()
await answerAgentRequest(workspaceId, { agentJobId, action: request }, approved, value)
answered = approved
}
</script>
<div class="flex flex-col gap-2 w-full text-xs">
{#if request.type === 'question'}
<div class="text-emphasis font-semibold">The agent {pending ? 'asks' : 'asked'} a question</div>
<div class="text-primary whitespace-pre-wrap">{request.question}</div>
{:else}
<div class="text-emphasis font-semibold">
The agent {pending ? 'asks' : 'asked'} to run {request.calls.length === 1
? 'this call'
: 'these calls'}
</div>
{#each request.calls as call, i (i)}
<div class="flex flex-col gap-1">
<span class="font-mono text-primary">{call.function_name}</span>
<pre
class="text-2xs bg-surface-secondary rounded-md p-2 overflow-auto max-h-40 whitespace-pre-wrap break-all"
>{JSON.stringify(call.arguments, null, 2)}</pre
>
</div>
{/each}
{/if}
{#if answered !== undefined}
<div class="text-secondary">
{#if request.type === 'question'}
{answered ? 'Answered' : 'Declined'}, the agent resumes shortly.
{:else}
{answered ? 'Approved' : 'Rejected'}, the agent resumes shortly.
{/if}
</div>
{:else if pending}
<div class="text-hint">Expires {displayDate(request.expires_at)}</div>
{#if choices.length > 0}
<div class="flex flex-wrap gap-2">
{#each choices as choice (choice)}
<Button
variant="default"
unifiedSize="sm"
selected={picked.includes(choice)}
onClick={() => pick(choice)}
>
{choice}
</Button>
{/each}
</div>
{/if}
<TextInput
bind:value={text}
inputProps={{
placeholder:
request.type === 'question'
? choices.length > 0
? 'Or answer in your own words'
: 'Your answer'
: 'Reason for the agent, if you reject (optional)',
oninput: () => {
if (!multiSelect) picked = []
}
}}
size="sm"
/>
{#if request.type === 'question'}
<ApprovalActions
approveLabel="Send answer"
rejectTitle="Decline to answer"
approveDisabled={answer.length === 0}
{onAnswer}
/>
{:else}
<ApprovalActions approveLabel="Approve" rejectTitle="Reject" {onAnswer} />
{/if}
{/if}
</div>
frontend/src/lib/components/FlowStatusViewerInner.svelte
+26
View File
@@ -48,11 +48,14 @@
AI_TOOL_MESSAGE_PREFIX,
AI_MCP_TOOL_CALL_PREFIX,
AI_WEBSEARCH_PREFIX,
AI_APPROVAL_PREFIX,
AI_QUESTION_PREFIX,
getAgentActionStateId,
getToolCallId
} from './graph/renderers/nodes/AIToolNode.svelte'
import JobAssetsViewer from './assets/JobAssetsViewer.svelte'
import McpToolCallDetails from './McpToolCallDetails.svelte'
import AgentApprovalRequest from './AgentApprovalRequest.svelte'
import GfmMarkdown from './GfmMarkdown.svelte'
import JobOtelTraces from './JobOtelTraces.svelte'
import JobDetailHeader from './runs/JobDetailHeader.svelte'
@@ -780,6 +783,10 @@
job_id: action.job_id,
type: success != undefined ? (success ? 'Success' : 'Failure') : 'InProgress'
})
} else if (action.type == 'approval' || action.type == 'question') {
setModuleState(stateId, {
type: success != undefined ? (success ? 'Success' : 'Failure') : 'WaitingForEvents'
})
} else if (action.type == 'mcp_tool_call') {
setModuleState(stateId, {
type: success != undefined ? (success ? 'Success' : 'Failure') : 'InProgress'
@@ -2127,6 +2134,25 @@
{/if}
{:else if selectedNode?.startsWith(AI_WEBSEARCH_PREFIX)}
<Alert type="info" title="Web search was used in this step" />
{:else if selectedNode?.startsWith(AI_APPROVAL_PREFIX) || selectedNode?.startsWith(AI_QUESTION_PREFIX)}
{@const [, agentModuleId, actionIndex] = selectedNode.split('-')}
{@const agentNode = localModuleStates?.[agentModuleId]}
{@const approvalIndex = parseInt(actionIndex)}
{@const approval = agentNode?.agent_actions?.[approvalIndex]}
{#if (approval?.type === 'approval' || approval?.type === 'question') && agentNode?.job_id}
<div class="p-4">
<!-- One form per request: its answered state must not carry over to the next. -->
{#key `${agentNode.job_id}-${approval.call_id}`}
<AgentApprovalRequest
workspaceId={job.workspace_id ?? ''}
agentJobId={agentNode.job_id}
request={approval}
pending={job.type === 'QueuedJob' &&
agentNode?.agent_actions_success?.[approvalIndex] === undefined}
/>
{/key}
</div>
{/if}
{:else if selectedNode}
{@const node = localModuleStates[selectedNode]}
{#if selectedNode == 'end'}
frontend/src/lib/components/FlowStatusWaitingForEvents.svelte
+25 -55
View File
@@ -1,11 +1,10 @@
<script lang="ts">
import { mergeSchema } from '$lib/common'
import { type Job, JobService } from '$lib/gen'
import { sendUserToast } from '$lib/toast'
import { ExternalLink, X } from 'lucide-svelte'
import { ExternalLink } from 'lucide-svelte'
import DisplayResult from './DisplayResult.svelte'
import Tooltip from './Tooltip.svelte'
import { Button } from './common'
import ApprovalActions from './approvals/ApprovalActions.svelte'
import SchemaForm from './SchemaForm.svelte'
import { twMerge } from 'tailwind-merge'
import { untrack } from 'svelte'
@@ -54,7 +53,6 @@
default_payload = args
approvalPageUrl = job_result?.['approvalPage']
actionTaken = false
hide_cancel = job?.raw_flow?.modules?.[approvalStep]?.suspend?.hide_cancel ?? false
schema = mergeSchema(
job?.raw_flow?.modules?.[approvalStep]?.suspend?.resume_form?.schema ?? {},
@@ -62,31 +60,21 @@
)
}
let loading = $state(false)
let actionTaken = $state(false)
async function continu(approve: boolean) {
if (!workspaceId) {
sendUserToast('Cannot resume: the job has no workspace', true)
return
}
loading = true
try {
await JobService.resumeSuspended({
workspace: workspaceId,
jobId: job?.id ?? '',
requestBody: {
payload: approve ? (default_payload as any) : undefined,
approved: approve
}
})
actionTaken = true
onAction?.(approve)
} catch (e: any) {
sendUserToast(e?.body ?? e?.message ?? 'Failed', true)
} finally {
loading = false
throw new Error('Cannot resume: the job has no workspace')
}
await JobService.resumeSuspended({
workspace: workspaceId,
jobId: job?.id ?? '',
requestBody: {
payload: approve ? (default_payload as any) : undefined,
approved: approve
}
})
onAction?.(approve)
}
let approvalStep = $derived((job?.flow_status?.step ?? 1) - 1)
// Everything this panel shows (description, resume form enums, approval page) is
// fetched from the suspended job — a recording carries none of it — and Resume /
@@ -108,34 +96,16 @@
<DisplayResult {workspaceId} noControls result={description} language={job?.language} />
<div class="mt-2"></div>
{/if}
<div>
<div class={twMerge('flex gap-2 items-center', light ? 'flex-col' : 'flex-row ')}>
{#if !hide_cancel}
<div>
<Button
title="Cancel the step"
iconOnly
startIcon={{ icon: X }}
variant="default"
disabled={loading || actionTaken}
destructive
unifiedSize="md"
on:click={() => continu(false)}
/>
</div>
{/if}
<div>
<Button
variant="accent"
onClick={() => continu(true)}
disabled={loading || actionTaken}
unifiedSize="md"
>
Resume
<Tooltip class="text-white">Resume or approve this suspended step</Tooltip>
</Button>
</div>
<!-- A new approval step gets usable buttons again. -->
{#key job?.flow_status?.modules?.[approvalStep]?.job}
<ApprovalActions
approveLabel="Resume"
approveTooltip="Resume or approve this suspended step"
rejectTitle="Cancel the step"
hideReject={hide_cancel}
{light}
onAnswer={continu}
>
{#if approvalPageUrl}
<a
href={approvalPageUrl}
@@ -161,7 +131,7 @@
variable
</Tooltip>
{/if}
</div>
</div>
</ApprovalActions>
{/key}
</div>
{/if}
frontend/src/lib/components/JobStatus.svelte
+2 -1
View File
@@ -2,6 +2,7 @@
import { displayDate, msToReadableTime } from '$lib/utils'
import type { CompletedJob, QueuedJob } from '$lib/gen'
import Badge from './common/badge/Badge.svelte'
import { isSuspended } from './agentRequest'
import { forLater } from '$lib/forLater'
import { Circle } from 'lucide-svelte'
import NoWorkerWithTagWarning from './runs/NoWorkerWithTagWarning.svelte'
@@ -58,7 +59,7 @@
/>
{/if}
</Badge>
{:else if job && 'running' in job && job.running && job.suspend}
{:else if job && 'running' in job && job.running && isSuspended(job)}
<div>
<Badge {large} color="violet">
Suspended
frontend/src/lib/components/agentRequest.ts
+60
View File
@@ -0,0 +1,60 @@
import { JobService, type FlowStatus, type FlowStatusModule } from '$lib/gen'
export type AgentRequest = Extract<
NonNullable<FlowStatusModule['agent_actions']>[number],
{ type: 'approval' | 'question' }
>
/** The approval or question an agent step's actions end on, while nobody has answered it. */
export function waitingRequest(
mod: Pick<FlowStatusModule, 'agent_actions' | 'agent_actions_success'>
) {
const actions = mod.agent_actions
const last = actions?.[actions.length - 1]
if (
!last ||
(last.type !== 'approval' && last.type !== 'question') ||
mod.agent_actions_success?.[actions!.length - 1] !== undefined
) {
return undefined
}
return last
}
/** The approval or question a flow's current agent step is parked on: the step stays in
* progress while its own job waits. */
export function pendingAgentApproval(job: { flow_status?: FlowStatus } | undefined) {
const mod = job?.flow_status?.modules?.[job?.flow_status?.step]
if (mod?.type !== 'InProgress' || !mod.job) return undefined
const action = waitingRequest(mod)
return action && { agentJobId: mod.job, action }
}
/** Suspended, as the run's badge and icon say: an approval step, or an agent waiting on a person. */
export function isSuspended(job: { suspend?: number; flow_status?: FlowStatus }) {
return !!job.suspend || !!pendingAgentApproval(job)
}
export type PendingAgentRequest = NonNullable<ReturnType<typeof pendingAgentApproval>>
/** `answer` is the question's answer, a list when several choices were picked, or the reason given
* with a rejection. Its call id goes with the answer, so a page left open on an earlier request
* cannot answer the one waiting now. */
export async function answerAgentRequest(
workspace: string,
{ agentJobId, action }: PendingAgentRequest,
approved: boolean,
answer?: string | string[]
) {
await JobService.resumeSuspended({
workspace,
jobId: agentJobId,
requestBody: {
approved,
payload: {
call_id: action.call_id,
...(answer?.length ? { [action.type === 'question' ? 'answer' : 'comment']: answer } : {})
}
}
})
}
frontend/src/lib/components/agentTrace.ts
+9
View File
@@ -90,6 +90,15 @@ export function buildAgentTrace(messages: AgentMessage[]): AgentTraceEntry[] {
entries.push({ kind: 'search' })
continue
}
if (action?.type === 'question') {
entries.push({
kind: 'tool',
name: 'Ask a question',
args: message.tool_call_id ? argsByCallId.get(message.tool_call_id) : undefined,
result: contentText(message.content)
})
continue
}
// Every message this run produced is tagged, including the agent narrating
// its next move and its final answer. Untagged ones are the prompt, the
// question, or a previous turn replayed out of memory — history loses its
frontend/src/lib/components/approvals/ApprovalActions.svelte
+76
View File
@@ -0,0 +1,76 @@
<script lang="ts">
import type { Snippet } from 'svelte'
import { X } from 'lucide-svelte'
import { twMerge } from 'tailwind-merge'
import { sendUserToast } from '$lib/toast'
import { Button } from '../common'
import Tooltip from '../Tooltip.svelte'
interface Props {
approveLabel: string
approveTooltip?: string
rejectTitle: string
hideReject?: boolean
/** Stacks the buttons, for the graph's popover. */
light?: boolean
/** Holds the approve button back, e.g. until an answer is typed. */
approveDisabled?: boolean
/** Throws to report a failure; the buttons stay usable then. */
onAnswer: (approved: boolean) => Promise<void>
/** Shown after the buttons, on the same row. */
children?: Snippet
}
let {
approveLabel,
approveTooltip,
rejectTitle,
hideReject = false,
light = false,
approveDisabled = false,
onAnswer,
children
}: Props = $props()
let loading = $state(false)
let answered = $state(false)
async function answer(approved: boolean) {
loading = true
try {
await onAnswer(approved)
answered = true
} catch (e: any) {
sendUserToast(e?.body ?? e?.message ?? 'Failed', true)
} finally {
loading = false
}
}
</script>
<div class={twMerge('flex gap-2 items-center', light ? 'flex-col' : 'flex-row')}>
{#if !hideReject}
<Button
title={rejectTitle}
iconOnly
startIcon={{ icon: X }}
variant="default"
disabled={loading || answered}
destructive
unifiedSize="md"
onClick={() => answer(false)}
/>
{/if}
<Button
variant="accent"
onClick={() => answer(true)}
disabled={loading || answered || approveDisabled}
unifiedSize="md"
>
{approveLabel}
{#if approveTooltip}
<Tooltip class="text-white">{approveTooltip}</Tooltip>
{/if}
</Button>
{@render children?.()}
</div>
frontend/src/lib/components/copilot/chat/flow/openFlow.json
+1 -1
View File
File diff suppressed because one or more lines are too long
frontend/src/lib/components/copilot/chat/flow/openFlowZod.gen.ts
Generated
+4 -4
View File
File diff suppressed because one or more lines are too long
frontend/src/lib/components/flows/FlowProgressBar.svelte
+8 -1
View File
@@ -2,6 +2,7 @@
import { type Job } from '$lib/gen'
import ProgressBar from '../progressBar/ProgressBar.svelte'
import { forLater } from '$lib/forLater'
import { waitingRequest } from '../agentRequest'
import type { Snippet } from 'svelte'
interface Props {
@@ -79,7 +80,13 @@
if (modules.length > maxDone) {
const nextModule = modules[maxDone]
if (nextModule.type === 'InProgress') {
if (nextModule.type === 'InProgress' && waitingRequest(nextModule)) {
// An agent waiting on a person reads like an approval step.
newIsWaitingForEvents = true
if (nextModule.id) {
newCurrentStepId = nextModule.id
}
} else if (nextModule.type === 'InProgress') {
newNextInProgress = true
// Get the step ID if available
if (nextModule.id) {
frontend/src/lib/components/flows/agentDraft.svelte.ts
+4 -6
View File
@@ -24,7 +24,7 @@ import {
// Lives here rather than beside the other config helpers: `agentResourceUtils` is a leaf, and
// importing tool-name validation into it would cycle back through `flowInfers` and pull the whole
// editor into every module that reads a brain key.
import { getToolNameError } from './agentToolUtils'
import { getToolNameError, isNamedTool, type AgentTool } from './agentToolUtils'
/**
* Why a run of this config would fail before it started, if it would. Only the conditions the
@@ -49,12 +49,10 @@ function agentConfigRunError(args: Record<string, any> | undefined): string | un
return 'Tools must be a list. Fix it in the resource editor before deploying.'
}
// Only a flowmodule tool's summary is a callable name: the worker never reads an mcp or
// websearch summary, so a blank one there is not an error and must not count as a sibling.
// Same filter as `collectInvalidAgentToolNames`, which is the rule the graph enforces.
// built-in tool's summary, so a blank one there is not an error and must not count as a
// sibling. Same filter as `collectInvalidAgentToolNames`, which is the rule the graph enforces.
const tools = (args?.tools ?? []) as Record<string, any>[]
const named = tools.filter(
(t) => t?.value?.tool_type !== 'mcp' && t?.value?.tool_type !== 'websearch'
)
const named = tools.filter((t) => isNamedTool(t as AgentTool))
const siblingNames = named.map((t) => t?.summary ?? '')
for (const tool of named) {
const err = getToolNameError(tool?.summary ?? '', tool?.value?.tool_type, siblingNames)
frontend/src/lib/components/flows/agentToolTree.ts
+3 -4
View File
@@ -2,6 +2,7 @@ import type { FlowModule } from '$lib/gen'
import {
getToolNameError,
isFlowModuleTool,
isNamedTool,
type AgentTool,
type FlowModuleTool
} from './agentToolUtils'
@@ -195,10 +196,8 @@ export function collectInvalidAgentToolNames(modules: unknown): InvalidAgentTool
forEachAiAgentModule(modules, (mod, v) => {
const tools: AgentTool[] = Array.isArray(v.tools) ? v.tools : []
// Only a flowmodule tool's summary is a callable name: the worker never reads an mcp or
// websearch summary, so a blank one there must stay writable (both default to '').
const named = tools.filter(
(tool) => tool.value?.tool_type !== 'mcp' && tool.value?.tool_type !== 'websearch'
)
// built-in tool's summary, so a blank one there must stay writable.
const named = tools.filter(isNamedTool)
const siblingNames = named.map((tool) => tool.summary ?? '')
for (const tool of named) {
const name = tool.summary ?? ''
frontend/src/lib/components/flows/agentToolUtils.test.ts
+7
View File
@@ -49,4 +49,11 @@ describe('toolEnabledName', () => {
`'${WEBSEARCH_ENABLED_NAME}' is a reserved name`
)
})
it("names the question tool by the worker's reserved name", () => {
expect(
toolEnabledName({ id: 't', summary: 'label', value: { tool_type: 'question' } } as any)
).toBe('__wm_question')
expect(getToolNameError('__wm_question')).toBe(`'__wm_question' is a reserved name`)
})
})
frontend/src/lib/components/flows/agentToolUtils.ts
+35 -6
View File
@@ -10,6 +10,26 @@ import { forbiddenIds } from './idUtils'
* search. */
export const WEBSEARCH_ENABLED_NAME = '__wm_web_search'
/** Tools the worker answers itself rather than by running a module. None is named by its
* `summary`: each has a reserved `enabledName` (mirroring the `*_ENABLED_NAME`s in the worker),
* and the model sees a fixed name the worker picks. `question` parks the agent until someone
* answers (`windmill-worker/src/ai/pause.rs`). */
export const BUILTIN_TOOLS = {
websearch: { label: 'Web search', enabledName: WEBSEARCH_ENABLED_NAME },
question: { label: 'Ask a question', enabledName: '__wm_question' }
} as const
export type BuiltinToolType = keyof typeof BUILTIN_TOOLS
export function builtinToolType(tool: AgentTool | undefined): BuiltinToolType | undefined {
const type = tool?.value?.tool_type
return type && type in BUILTIN_TOOLS ? (type as BuiltinToolType) : undefined
}
/** Whether a tool is named by its `summary`, which only the model-facing kinds are. */
export function isNamedTool(tool: AgentTool | undefined): boolean {
return tool?.value?.tool_type !== 'mcp' && builtinToolType(tool) === undefined
}
/**
* A tool's `summary` is the name the LLM sees, and the worker rejects any name that does not match
* `^[a-zA-Z0-9_]+$` (`ai_executor.rs`), so an unvalidated name fails on every run of the flow.
@@ -21,10 +41,10 @@ export const WEBSEARCH_ENABLED_NAME = '__wm_web_search'
*/
export function getToolNameError(
name: string,
kind?: 'mcp' | 'websearch' | (string & {}),
kind?: 'mcp' | BuiltinToolType | (string & {}),
siblingNames?: string[]
): string | undefined {
if (kind === 'websearch') return undefined
if (kind && kind in BUILTIN_TOOLS) return undefined
if (kind === 'mcp') {
return name.length > 0 ? undefined : 'Tool name must not be empty'
}
@@ -36,7 +56,10 @@ export function getToolNameError(
if (!/^[a-zA-Z0-9_]+$/.test(name)) {
return 'Tool name must only contain letters, numbers and underscores'
}
if (forbiddenIds.includes(name) || name === WEBSEARCH_ENABLED_NAME) {
if (
forbiddenIds.includes(name) ||
Object.values(BUILTIN_TOOLS).some((builtin) => builtin.enabledName === name)
) {
return `'${name}' is a reserved name`
}
if (siblingNames && siblingNames.filter((n) => n === name).length > 1) {
@@ -48,6 +71,7 @@ export function getToolNameError(
export const SPECIAL_TOOL_KINDS = [
'mcpTool',
'websearchTool',
'questionTool',
'aiAgentTool',
'aiDecisionTool'
] as const
@@ -154,8 +178,9 @@ export function toolEnabledName(tool: AgentTool): string | undefined {
if (value?.tool_type === 'mcp') {
return (value?.resource_path as string | undefined)?.replace(/^\$res:/, '') || undefined
}
if (value?.tool_type === 'websearch') {
return WEBSEARCH_ENABLED_NAME
const builtin = builtinToolType(tool)
if (builtin) {
return BUILTIN_TOOLS[builtin].enabledName
}
return toolDisplayName(tool)
}
@@ -233,13 +258,17 @@ export function createMcpTool(id: string): McpTool {
export function createWebsearchTool(id: string): WebsearchTool {
return {
id,
summary: 'Web Search',
summary: 'Web search',
value: {
tool_type: 'websearch'
}
}
}
export function createQuestionTool(id: string): AgentTool {
return { id, summary: BUILTIN_TOOLS.question.label, value: { tool_type: 'question' } }
}
/**
* Convert a FlowModuleTool to a FlowModule for use with loadFlowModuleState etc.
* Strips the extra `tool_type` field and maps AgentTool fields to FlowModule fields.
frontend/src/lib/components/flows/content/AgentConfigModal.svelte
+8 -2
View File
@@ -15,7 +15,12 @@ is that one, since both answer the same question about what an assistant can see
summarizeAgentBrain,
type AIAgentConfig
} from '../agentResourceUtils'
import { toolDisplayName, type AgentTool } from '../agentToolUtils'
import {
BUILTIN_TOOLS,
builtinToolType,
toolDisplayName,
type AgentTool
} from '../agentToolUtils'
type Section = 'model' | 'instructions' | 'tools'
@@ -71,7 +76,8 @@ is that one, since both answer the same question about what an assistant can see
function toolKind(tool: AgentTool): string {
const value = tool?.value as Record<string, any> | undefined
if (value?.tool_type === 'mcp') return 'MCP server'
if (value?.tool_type === 'websearch') return 'Web search'
const builtin = builtinToolType(tool)
if (builtin) return BUILTIN_TOOLS[builtin].label
switch (value?.type) {
case 'script':
return `Script ${value.path ?? ''}`.trim()
frontend/src/lib/components/flows/content/AgentToolRoster.svelte
+12 -6
View File
@@ -1,7 +1,13 @@
<script lang="ts">
import { ChevronRight, Plus, Wrench, X } from 'lucide-svelte'
import { twMerge } from 'tailwind-merge'
import { getToolNameError, toolDisplayName } from '../agentToolUtils'
import {
BUILTIN_TOOLS,
builtinToolType,
getToolNameError,
isNamedTool,
toolDisplayName
} from '../agentToolUtils'
import type { AgentTool } from '../agentToolUtils'
import { Button } from '$lib/components/common'
import Popover from '$lib/components/meltComponents/Popover.svelte'
@@ -42,15 +48,14 @@
if (value?.type === 'aiagent') return 'agent'
if (value?.type === 'aidecision') return 'decision'
if (value?.tool_type === 'mcp') return 'MCP'
if (value?.tool_type === 'websearch') return 'web search'
const builtin = builtinToolType(tool)
if (builtin) return BUILTIN_TOOLS[builtin].label.toLowerCase()
return undefined
}
/** Only a flow module tool's summary is the name the model is given; an mcp or websearch
/** Only a flow module tool's summary is the name the model is given; an mcp or built-in tool's
* summary is never sent, so a blank one there is not an error. Same rule as `agentToolTree`. */
function isNamed(tool: AgentTool): boolean {
return tool?.value?.tool_type !== 'mcp' && tool?.value?.tool_type !== 'websearch'
}
const isNamed = isNamedTool
let siblingNames = $derived(tools.filter(isNamed).map((tool) => tool?.summary ?? ''))
@@ -103,6 +108,7 @@
)}
on:pickMcpTool={() => (onAddTool?.({ kind: 'mcpTool' }), close())}
on:pickWebsearchTool={() => (onAddTool?.({ kind: 'websearchTool' }), close())}
on:pickQuestionTool={() => (onAddTool?.({ kind: 'questionTool' }), close())}
on:pickAiAgentTool={() => (onAddTool?.({ kind: 'aiAgentTool' }), close())}
on:pickAiDecisionTool={() => (onAddTool?.({ kind: 'aiDecisionTool' }), close())}
/>
frontend/src/lib/components/flows/content/AgentToolWrapper.svelte
+6
View File
@@ -5,6 +5,7 @@
import FlowModuleComponent from './FlowModuleComponent.svelte'
import McpToolEditor from './McpToolEditor.svelte'
import WebsearchToolDisplay from './WebsearchToolDisplay.svelte'
import QuestionToolDisplay from './QuestionToolDisplay.svelte'
interface Props {
tool: AgentTool
@@ -62,6 +63,9 @@
isAgentTool={true}
{staticOnly}
bind:toolDescription={tool.description}
bind:toolRequiresApproval={
() => tool.requires_approval, (v) => (tool.requires_approval = v || undefined)
}
{siblingToolNames}
/>
{:else if isMcpTool(tool)}
@@ -69,4 +73,6 @@
<McpToolEditor bind:tool {noEditor} />
{:else if isWebsearchTool(tool)}
<WebsearchToolDisplay {noEditor} />
{:else if tool.value?.tool_type === 'question'}
<QuestionToolDisplay bind:tool {noEditor} />
{/if}
frontend/src/lib/components/flows/content/FlowModuleComponent.svelte
+37 -1
View File
@@ -71,7 +71,8 @@
signDebugRequest,
getDebugErrorMessage
} from '$lib/components/debug'
import { Bug, Terminal } from 'lucide-svelte'
import { Bug, Hand, Terminal } from 'lucide-svelte'
import Popover from '$lib/components/Popover.svelte'
import { sendUserToast } from '$lib/utils'
import { useOperatingWorkspace } from '$lib/components/operatingWorkspace.svelte'
@@ -133,6 +134,8 @@
/** Lets the agent's Tools section add a tool through the graph's own insert path. */
flowModuleSchemaMap?: import('../map/FlowModuleSchemaMap.svelte').default
toolDescription?: string | undefined
/** The tool's `requires_approval` setting; only an agent tool has one. */
toolRequiresApproval?: boolean
siblingToolNames?: string[]
}
@@ -153,6 +156,7 @@
staticOnly = false,
flowModuleSchemaMap = undefined,
toolDescription = $bindable(undefined),
toolRequiresApproval = $bindable(),
siblingToolNames = undefined
}: Props = $props()
@@ -1149,6 +1153,20 @@
label="Chat"
/>
{/if}
{#if isAgentTool}
<Tab value="tool_settings" label="Tool settings">
{#snippet extra()}
{#if toolRequiresApproval}
<div class="flex flex-row items-center gap-1 text-secondary">
<Popover notClickable>
<Hand size={12} />
{#snippet text()}Require approval · On{/snippet}
</Popover>
</div>
{/if}
{/snippet}
</Tab>
{/if}
{#if !preprocessorModule && !isAgentTool}
<Tab value="advanced" label="Run settings">
{#snippet extra()}
@@ -1395,6 +1413,24 @@
/>
</Section>
</div>
{:else if visibleSelected === 'tool_settings' && isAgentTool}
<div class="flex-1 overflow-auto p-4">
<Section label="Approval">
<Toggle
size="xs"
textClass="text-xs font-normal text-primary"
disabled={noEditor}
bind:checked={
() => toolRequiresApproval ?? false, (v) => (toolRequiresApproval = v)
}
options={{
right: 'Require approval',
rightTooltip:
'Each call waits for someone to approve or reject it from the run page or the chat, without holding a worker. Other calls in the same turn run right away. A rejected call is reported to the agent as not run. An agent used as a tool of another agent cannot wait, so it leaves such tools out.'
}}
/>
</Section>
</div>
{:else if visibleSelected === 'advanced'}
<FlowRunSettings
bind:this={runSettings}
frontend/src/lib/components/flows/content/McpToolEditor.svelte
+14
View File
@@ -31,6 +31,7 @@
import McpConnect from '$lib/components/mcp/McpConnect.svelte'
import type { FlowEditorContext } from '../types'
import { useOperatingWorkspace } from '$lib/components/operatingWorkspace.svelte'
import Toggle from '$lib/components/Toggle.svelte'
const operatingWorkspace = useOperatingWorkspace()
@@ -126,6 +127,19 @@
/>
</Label>
</div>
<Toggle
size="xs"
textClass="text-xs font-normal text-primary"
disabled={noEditor}
bind:checked={
() => tool.requires_approval ?? false, (v) => (tool.requires_approval = v || undefined)
}
options={{
right: 'Require approval',
rightTooltip:
'Each call waits for someone to approve or reject it from the run page or the chat, without holding a worker. Other calls in the same turn run right away. A rejected call is reported to the agent as not run. An agent used as a tool of another agent cannot wait, so it leaves such tools out.'
}}
/>
<Section label="Available Tools">
{#snippet action()}
frontend/src/lib/components/flows/content/QuestionToolDisplay.svelte
+32
View File
@@ -0,0 +1,32 @@
<script lang="ts">
import { SecondsInput } from '$lib/components/common'
import Label from '$lib/components/Label.svelte'
import FlowCard from '../common/FlowCard.svelte'
import { BUILTIN_TOOLS, type AgentTool } from '../agentToolUtils'
/** Matches `DEFAULT_TIMEOUT_SECS` in `windmill-worker/src/ai/pause.rs`. */
const DEFAULT_QUESTION_TIMEOUT_SECS = 1800
let { tool = $bindable(), noEditor = false }: { tool: AgentTool; noEditor?: boolean } = $props()
let value = $derived(tool.value as { tool_type: 'question'; timeout?: number })
</script>
<FlowCard
{noEditor}
title={BUILTIN_TOOLS.question.label}
subtitle="Lets the agent ask a person and wait for their answer, from the run page or the chat."
subtitleDocLink="https://www.windmill.dev/docs/core_concepts/ai_agents"
>
<div class="flex flex-col gap-4 overflow-auto p-4" style="scrollbar-gutter: stable">
<Label label="Timeout">
<p class="text-xs font-normal text-secondary">Past it, the agent is told nobody answered</p>
<SecondsInput
disabled={noEditor}
bind:seconds={
() => value.timeout ?? DEFAULT_QUESTION_TIMEOUT_SECS, (v) => (value.timeout = v)
}
/>
</Label>
</div>
</FlowCard>
frontend/src/lib/components/flows/content/WebsearchToolDisplay.svelte
+6 -9
View File
@@ -1,15 +1,12 @@
<script lang="ts">
import Alert from '$lib/components/common/alert/Alert.svelte'
import FlowCard from '../common/FlowCard.svelte'
let { noEditor = false }: { noEditor?: boolean } = $props()
</script>
<FlowCard {noEditor} title="Web search">
<div class="flex flex-col gap-4 overflow-auto p-4" style="scrollbar-gutter: stable">
<Alert title="Web search tool" type="info">
Gives the AI Agent the ability to search the web. Only works for openai, anthropic and google
models for now.
</Alert>
</div>
</FlowCard>
<FlowCard
{noEditor}
title="Web search"
subtitle="Works with OpenAI, Anthropic and Google models."
subtitleDocLink="https://www.windmill.dev/docs/core_concepts/ai_agents"
/>
frontend/src/lib/components/flows/conversations/FlowChatInterface.svelte
+50 -2
View File
@@ -12,7 +12,7 @@
import { tick, untrack } from 'svelte'
import type { Chat } from 'windmill-chat'
import { saveFlowChatInputs } from './flowChatProps'
import { ResourceService, type FlowModule } from '$lib/gen'
import { JobService, ResourceService, type FlowModule } from '$lib/gen'
import { resource } from 'runed'
import { useWorkspaceStorageConfigured } from '$lib/components/inputTransformEnv.svelte'
import {
@@ -22,6 +22,10 @@
} from './agentAttachmentInput'
import { deepEqual } from 'fast-equals'
import FlowChatModelSettings from './FlowChatModelSettings.svelte'
import {
pendingAgentApproval,
type PendingAgentRequest
} from '$lib/components/agentRequest'
import {
agentModelGap,
chatLinkedAgentPaths,
@@ -202,7 +206,8 @@
sendDisabled: () => deploymentInProgress || !!modelGap || !!wrongKindReason,
// The model controls only: a retry changes model when the reader did, but replays
// the run's own attachments rather than whatever the composer holds now.
inputsShownInComposer: () => composerOwnedInputs(modelWiring, undefined)
inputsShownInComposer: () => composerOwnedInputs(modelWiring, undefined),
agentRequest: () => agentRequest
})
setChatViewHost(chatHost)
@@ -248,6 +253,49 @@
})
})
// Only a flow that can ask a person something reads its running turn's status. Read off the
// flow, not the stream: an agent that does not stream sends no tool call. A linked agent's
// tools are not loaded here, so it counts; only a top-level step's request is answered from
// the chat.
const mayRequestApproval = $derived(
(flowModules ?? []).some((module) => {
const value = module.value as any
return (
value?.type === 'aiagent' &&
(!!value.agent ||
value.tools?.some(
(tool: any) => tool?.requires_approval || tool?.value?.tool_type === 'question'
))
)
})
)
const approvalJobId = $derived(mayRequestApproval ? chatHost.state.runningJobId : undefined)
// The stream only says a call is running; the request to answer, and the job that answers
// it, are on the flow's status.
let agentRequest = $state<PendingAgentRequest | undefined>(undefined)
$effect(() => {
const id = approvalJobId
agentRequest = undefined
if (!id || !workspace) return
const args = { workspace, id, noLogs: true }
let stopped = false
const load = async () => {
try {
const job = await JobService.getJob(args)
if (!stopped) agentRequest = pendingAgentApproval(job)
} catch {
// transient; the next poll reads it again
}
}
void load()
const interval = setInterval(load, 2000)
return () => {
stopped = true
clearInterval(interval)
}
})
// Older pages load when the reader reaches the top; the viewport stays where it was.
let scrollElement = $state<HTMLDivElement | undefined>(undefined)
let loadingOlder = false
frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts
+122 -6
View File
@@ -18,7 +18,11 @@ import type {
ChatSendRequestOptions,
ChatViewHost
} from '$lib/components/copilot/chat/chatViewHost'
import type { DisplayMessage, WebSearchResult } from '$lib/components/copilot/chat/shared'
import type {
DisplayMessage,
ToolDisplayMessage,
WebSearchResult
} from '$lib/components/copilot/chat/shared'
import type { AIAutonomyMode } from '$lib/components/copilot/chat/AIChatManager.svelte'
import { AIWorkMode } from '$lib/components/copilot/chat/workMode'
import { isPlanCardTool } from '$lib/components/copilot/chat/planMode'
@@ -36,6 +40,11 @@ import { JobService } from '$lib/gen'
import { sendUserToast } from '$lib/toast'
import { attachmentLanes } from './messageAttachments'
import type { AttachmentsTarget } from './agentAttachmentInput'
import {
answerAgentRequest,
type AgentRequest,
type PendingAgentRequest
} from '$lib/components/agentRequest'
export type FlowChatViewHostOptions = {
/** The flow inputs sent next to `user_message` with every turn. */
@@ -55,6 +64,8 @@ export type FlowChatViewHostOptions = {
/** Flow inputs edited by a control beside the composer, such as the model button. A retry
* takes their current value rather than the failed turn's. */
inputsShownInComposer?: () => string[]
/** The approval or question the running turn's agent step waits on. */
agentRequest?: () => PendingAgentRequest | undefined
/** Injectables for tests: the clock and scheduler behind the typewriter pacing. */
revealOptions?: Pick<TypewriterRevealOptions, 'instant' | 'now' | 'schedule' | 'cancel'>
}
@@ -133,6 +144,57 @@ export type DisplayLookups = {
workspace?: string
/** How much of a pending assistant row the pacing has put on screen. */
revealed?: (message: ChatMessage) => Revealed | undefined
/** The approval or question the running turn's agent waits on. */
request?: AgentRequest
}
const QUESTION_CALL = /^ask_question(_\d+)?$/
/** The calls a request holds, as the cards the copilot shows for a confirmation or a question.
* `row` is the streamed card of that call, if any: an agent that does not stream sends none. */
function requestCard(
request: AgentRequest,
toolName: string,
row: ToolDisplayMessage
): ToolDisplayMessage {
if (request.type === 'question') {
return {
...row,
content: 'Waiting for your answer',
isLoading: true,
userQuestion: {
question: request.question,
choices: request.choices ?? [],
multiSelect: request.multi_select
}
}
}
const call = request.calls.find((c) => c.function_name === toolName)
const others = request.calls.length - 1
return {
...row,
content:
`Waiting for approval to run ${toolName}` +
(others > 0 ? `, answered together with ${others} other call${others > 1 ? 's' : ''}` : ''),
parameters: row.parameters ?? call?.arguments,
showDetails: true,
isLoading: true,
needsConfirmation: true,
confirmAction: 'run'
}
}
function requestKey({ agentJobId, action }: PendingAgentRequest): string {
return `${agentJobId}:${action.call_id}`
}
function holds(request: AgentRequest, toolName: string | undefined): toolName is string {
return (
!!toolName &&
(request.type === 'question'
? QUESTION_CALL.test(toolName)
: request.calls.some((c) => c.function_name === toolName))
)
}
/**
@@ -196,7 +258,9 @@ export function toDisplayMessages(
for (let i = messages.length - 1; i >= 0 && latestUser < 0; i--) {
if (messages[i].role === 'user') latestUser = i
}
return messages.flatMap((message, i): DisplayMessage[] => {
const request = lookups.request
let requestShown = false
const shown = messages.flatMap((message, i): DisplayMessage[] => {
switch (message.role) {
case 'user': {
const index = userIndex++
@@ -253,7 +317,7 @@ export function toDisplayMessages(
!STRUCTURED_OUTPUT_CALL.test(message.tool.name)
? `${message.tool.name} did not finish`
: undefined
const call: DisplayMessage = {
let call: ToolDisplayMessage = {
role: 'tool',
tool_call_id: message.id,
// The card's header is the row's text, which the server only words once the
@@ -270,6 +334,10 @@ export function toDisplayMessages(
isLoading: message.pending && message.tool?.status === 'running',
jobId: message.jobId
}
if (request && i > latestUser && call.isLoading && holds(request, toolName)) {
call = requestCard(request, toolName, call)
requestShown = true
}
// The tool card has no thinking section: the thinking that led to the call reads
// as a card of its own, just before it.
return message.reasoning
@@ -308,6 +376,25 @@ export function toDisplayMessages(
}
}
})
if (!request || requestShown) return shown
// No streamed card to put it on: the agent does not stream, or the page was opened after
// the call was made.
const calls =
request.type === 'question'
? [{ function_name: 'ask_question', arguments: undefined }]
: request.calls
return [
...shown,
...calls.map((call, n) =>
requestCard(request, call.function_name, {
role: 'tool',
tool_call_id: `${request.call_id}:${n}`,
content: '',
toolName: call.function_name,
parameters: call.arguments
})
)
]
}
/** The two paced lanes of one streaming message, and how much of each has been fed in. */
@@ -450,9 +537,30 @@ export class FlowChatViewHost implements ChatViewHost, DraftSender<ComposerAttac
displayMessages = $derived.by(() =>
toDisplayMessages(this.#state.messages, isBusy(this.#state.status), this.#stoppedTurns, {
workspace: this.#options.workspace?.(),
revealed: (message) => this.#revealed[message.id]
revealed: (message) => this.#revealed[message.id],
request: this.#agentRequest?.action
})
)
/** The request answered here, by agent job and call id: its cards drop their buttons until
* the flow's status catches up. */
#answeredRequest = $state<string | undefined>(undefined)
get #agentRequest(): PendingAgentRequest | undefined {
const request = this.#options.agentRequest?.()
return request && requestKey(request) === this.#answeredRequest ? undefined : request
}
#answer(approved: boolean, answer?: string | string[]): boolean {
const request = this.#agentRequest
const workspace = this.#options.workspace?.()
if (!request || !workspace) return false
this.#answeredRequest = requestKey(request)
answerAgentRequest(workspace, request, approved, answer).catch((e: any) => {
this.#answeredRequest = undefined
sendUserToast(e?.body ?? e?.message ?? 'Failed to answer the request', true)
})
return true
}
/** The user message of each turn stopped in this view, by message id and, once the chat has
* read its row, row id: a reopened conversation or an older page rebuilds messages from rows,
* so a turn left before its row was read is not recognised there. Only this session knows;
@@ -698,8 +806,16 @@ export class FlowChatViewHost implements ChatViewHost, DraftSender<ComposerAttac
void this.#turns.send(composerDraft(message.content), replayInputs)
}
restartGeneration = () => {}
handleUserQuestionAnswer = () => false
handleToolConfirmation = () => {}
handleUserQuestionAnswer = (_toolId: string, choices: string[]) =>
this.#answer(
true,
this.#agentRequest?.action.type === 'question' && this.#agentRequest.action.multi_select
? choices
: choices[0]
)
handleToolConfirmation = (_toolId: string, confirmed: boolean) => {
this.#answer(confirmed)
}
hasPendingRunForm = false
isRunFormPending = () => false
frontend/src/lib/components/flows/conversations/flowChatViewHost.test.ts
+45
View File
@@ -177,6 +177,51 @@ describe('toDisplayMessages', () => {
expect(notCitations).toMatchObject({ toolName: 'websearch', result: { hits: 3 } })
})
it('puts a pending request on the card of the call it holds, or adds one when none streamed', () => {
const approval = {
type: 'approval' as const,
call_id: 'c1',
calls: [{ function_name: 'delete_database', arguments: { name: 'staging' } }],
expires_at: '2026-09-16T10:30:00Z'
}
const turn = [
message({ role: 'user', content: 'delete staging' }),
message({ role: 'tool', pending: true, tool: { name: 'list_databases', status: 'success' } }),
message({ role: 'tool', pending: true, tool: { name: 'delete_database', status: 'running' } })
]
const held = toDisplayMessages(turn, true, new Set(), { request: approval })
expect(held).toHaveLength(3)
expect(held[1]).not.toHaveProperty('needsConfirmation')
expect(held[2]).toMatchObject({
needsConfirmation: true,
isLoading: true,
parameters: { name: 'staging' }
})
const twice = {
...approval,
calls: [...approval.calls, { function_name: 'delete_database', arguments: { name: 'prod' } }]
}
const added = toDisplayMessages(turn.slice(0, 1), true, new Set(), { request: twice })
expect(added.slice(1).map((m) => (m as { parameters?: unknown }).parameters)).toEqual([
{ name: 'staging' },
{ name: 'prod' }
])
const question = {
type: 'question' as const,
call_id: 'c2',
question: 'Which one?',
expires_at: '2026-09-16T10:30:00Z'
}
const unstreamed = toDisplayMessages(turn.slice(0, 1), true, new Set(), { request: question })
expect(unstreamed[1]).toMatchObject({
role: 'tool',
isLoading: true,
userQuestion: { question: 'Which one?', choices: [] }
})
})
it('shows a call the turn finished without as an error, not as still running', () => {
const display = toDisplayMessages([
message({ role: 'user', content: 'hi' }),
frontend/src/lib/components/flows/flowStateUtils.svelte.ts
+4
View File
@@ -32,6 +32,7 @@ import {
createAiDecisionTool,
createMcpTool,
createWebsearchTool,
createQuestionTool,
newAiDecisionInputTransforms,
newFlowModuleAgentTool,
SPECIAL_TOOL_KINDS,
@@ -587,6 +588,9 @@ export async function insertNewModuleAtIndex(
const websearchTool = createWebsearchTool(module.id)
;(modules as AgentTool[]).splice(index, 0, websearchTool)
return modules as AgentTool[]
} else if (toolKind === 'questionTool') {
;(modules as AgentTool[]).splice(index, 0, createQuestionTool(module.id))
return modules as AgentTool[]
} else if (toolKind === 'aiAgentTool') {
// Create AI Agent tool (nested agent)
const aiAgentTool = createAiAgentTool(module.id)
frontend/src/lib/components/flows/map/InsertModuleInner.svelte
+11 -1
View File
@@ -243,12 +243,22 @@
}}
/>
<TopLevelNode
label="Web Search"
label="Web search"
onSelect={() => {
dispatch('pickWebsearchTool')
dispatch('close')
}}
/>
<!-- A nested agent cannot park, so the worker drops it from its roster. -->
{#if !nestedAgent}
<TopLevelNode
label="Ask a question"
onSelect={() => {
dispatch('pickQuestionTool')
dispatch('close')
}}
/>
{/if}
<TopLevelNode
label="AI Agent"
chevron={false}
frontend/src/lib/components/flows/pickers/TopLevelNode.svelte
+3 -1
View File
@@ -11,6 +11,7 @@
Repeat,
Scale,
Square,
MessageCircleQuestion,
Zap
} from 'lucide-svelte'
import type { ComponentType } from 'svelte'
@@ -54,7 +55,8 @@
'Claude Code': { icon: BotIcon, iconClass: 'text-ai' },
Pi: { icon: BotIcon, iconClass: 'text-ai' },
MCP: { icon: Plug },
'Web Search': { icon: Globe }
'Web search': { icon: Globe },
'Ask a question': { icon: MessageCircleQuestion }
}
const config = $derived(iconMap[label])
frontend/src/lib/components/graph/graphBuilder.svelte.ts
+1
View File
@@ -32,6 +32,7 @@ export type InsertKind =
| 'aidecision'
| 'mcpTool'
| 'websearchTool'
| 'questionTool'
| 'aiAgentTool'
| 'aiDecisionTool'
frontend/src/lib/components/graph/renderers/nodes/AIToolNode.svelte
+44 -11
View File
@@ -1,5 +1,10 @@
<script module lang="ts">
import { getToolNameError, type AgentTool } from '$lib/components/flows/agentToolUtils'
import {
BUILTIN_TOOLS,
builtinToolType,
getToolNameError,
type AgentTool
} from '$lib/components/flows/agentToolUtils'
export const AI_TOOL_BASE_OFFSET = 5
export const AI_TOOL_ROW_OFFSET = 30
@@ -9,6 +14,8 @@
export const AI_MCP_TOOL_CALL_PREFIX = '_wm_ai_mcp_tool_call'
export const AI_TOOL_MESSAGE_PREFIX = '_wm_ai_agent_message'
export const AI_WEBSEARCH_PREFIX = '_wm_ai_websearch'
export const AI_APPROVAL_PREFIX = '_wm_ai_approval'
export const AI_QUESTION_PREFIX = '_wm_ai_question'
const ROW_WIDTH = 275
const NEW_TOOL_NODE_WIDTH = 50
@@ -48,6 +55,8 @@
return action.module_id === tool.moduleId
case 'web_search':
return tool.type === 'websearch'
case 'question':
return tool.type === 'question'
case 'mcp_tool_call':
// One MCP node stands for a whole server and many function names, so the server path
// is the only join that holds. The worker strips `$res:` before building the action,
@@ -56,6 +65,8 @@
tool.type === 'mcp' &&
bareResourcePath(action.resource_path) === bareResourcePath(tool.resourcePath)
)
// The calls it holds are decorated by their own actions once they run.
case 'approval':
case 'message':
return false
}
@@ -76,6 +87,10 @@
return AI_MCP_TOOL_CALL_PREFIX + '-' + agentModuleId + '-' + idx
case 'web_search':
return AI_WEBSEARCH_PREFIX + '-' + agentModuleId + '-' + idx
case 'approval':
return AI_APPROVAL_PREFIX + '-' + agentModuleId + '-' + idx
case 'question':
return AI_QUESTION_PREFIX + '-' + agentModuleId + '-' + idx
case 'message':
return getToolCallId(idx, agentModuleId)
}
@@ -156,15 +171,13 @@
stateType?: GraphModuleState['type']
resourcePath?: string
}[] = sourceTools.map((t, idx) => {
// Handle FlowModule, MCP, and Websearch tools
// Handle FlowModule, MCP, and built-in tools
const toolType =
t.value.tool_type === 'mcp'
? 'mcp'
: t.value.tool_type === 'websearch'
? 'websearch'
: t.value.tool_type === 'flowmodule'
? t.value.type
: undefined
t.value.tool_type === 'mcp' || builtinToolType(t)
? t.value.tool_type
: t.value.tool_type === 'flowmodule'
? t.value.type
: undefined
return {
id: t.id,
name: t.summary ?? '',
@@ -183,7 +196,12 @@
if (a.type === 'tool_call' || a.type === 'mcp_tool_call') {
return { id, name: a.function_name }
} else if (a.type === 'web_search') {
return { id, name: 'Web Search', type: 'websearch' }
return { id, name: 'Web search', type: 'websearch' }
} else if (a.type === 'approval') {
const names = a.calls.map((call) => call.function_name).join(', ')
return { id, name: `Approval: ${names}`, type: 'approval' }
} else if (a.type === 'question') {
return { id, name: BUILTIN_TOOLS.question.label, type: 'question' }
} else {
return { id, name: 'Message' }
}
@@ -304,7 +322,18 @@
NewAiToolN,
NodeLayout
} from '../../graphBuilder.svelte'
import { Bot, Globe, MessageCircle, Play, Plug, Scale, Wrench, X } from 'lucide-svelte'
import {
Bot,
Globe,
MessageCircle,
MessageCircleQuestion,
Play,
Plug,
Scale,
UserCheck,
Wrench,
X
} from 'lucide-svelte'
import { twMerge } from 'tailwind-merge'
import type { Edge, Node } from '@xyflow/svelte'
@@ -403,6 +432,10 @@
<Play size={16} class="ml-1 shrink-0" />
{:else if data.type === 'websearch'}
<Globe size={16} class="ml-1 shrink-0" />
{:else if data.type === 'approval'}
<UserCheck size={16} class="ml-1 shrink-0" />
{:else if data.type === 'question'}
<MessageCircleQuestion size={16} class="ml-1 shrink-0" />
{:else if data.type === 'mcp'}
<Plug size={16} class="ml-1 shrink-0" />
{:else if data.type === 'aiagent'}
frontend/src/lib/components/graph/renderers/nodes/ModuleNode.svelte
+8 -1
View File
@@ -10,6 +10,7 @@
import { getContext } from 'svelte'
import { getGraphContext } from '../../graphContext'
import { getFlowRunStatusContext } from '../../flowRunStatus.svelte'
import { waitingRequest } from '$lib/components/agentRequest'
interface Props {
data: ModuleN['data']
@@ -22,9 +23,13 @@
const flowRunStatus = getFlowRunStatusContext()
let state = $derived.by(() => {
return data.testModuleState
const state = data.testModuleState
? (jobToGraphModuleState(data.testModuleState) ?? flowRunStatus?.getModuleState(data.id))
: flowRunStatus?.getModuleState(data.id)
// An agent waiting on a person reads like an approval step, though its job is running.
return state?.type === 'InProgress' && waitingRequest(state)
? { ...state, type: 'WaitingForEvents' as const }
: state
})
// A `message` is the one action with no tool node of its own, so the annotation is the only
@@ -33,6 +38,8 @@
{ type: 'tool_call', singular: 'tool call', plural: 'tool calls' },
{ type: 'mcp_tool_call', singular: 'MCP call', plural: 'MCP calls' },
{ type: 'web_search', singular: 'web search', plural: 'web searches' },
{ type: 'approval', singular: 'approval request', plural: 'approval requests' },
{ type: 'question', singular: 'question', plural: 'questions' },
{ type: 'message', singular: 'message', plural: 'messages' }
] as const
frontend/src/lib/components/graph/renderers/nodes/NewAIToolNode.svelte
+8
View File
@@ -106,6 +106,14 @@
})
close()
}}
on:pickQuestionTool={() => {
data.eventHandlers.insert({
index: -1,
agentId: data.agentModuleId,
kind: 'questionTool'
})
close()
}}
on:pickAiAgentTool={(e) => {
data.eventHandlers.insert({
index: -1,
frontend/src/lib/components/runs/FlowExecutionStatus.svelte
+14 -1
View File
@@ -1,5 +1,7 @@
<script lang="ts">
import FlowStatusWaitingForEvents from '../FlowStatusWaitingForEvents.svelte'
import AgentApprovalRequest from '../AgentApprovalRequest.svelte'
import { pendingAgentApproval } from '../agentRequest'
import type { FlowStatusModule, Job } from '$lib/gen'
import { type StateStore } from '$lib/utils'
import { fade, slide } from 'svelte/transition'
@@ -18,7 +20,9 @@
)
const isSuspended = $derived(suspendStatus.val && Object.keys(suspendStatus.val).length > 0)
const anyStatusActive = $derived(isWaitingForEvents || isSuspended)
const pendingApproval = $derived(pendingAgentApproval(job))
const anyStatusActive = $derived(isWaitingForEvents || isSuspended || !!pendingApproval)
</script>
{#if job && anyStatusActive}
@@ -28,6 +32,15 @@
>
{#if isWaitingForEvents}
<FlowStatusWaitingForEvents workspaceId={job.workspace_id} {job} {isOwner} />
{:else if pendingApproval}
{#key pendingApproval.action.call_id}
<AgentApprovalRequest
workspaceId={job.workspace_id ?? ''}
agentJobId={pendingApproval.agentJobId}
request={pendingApproval.action}
pending
/>
{/key}
{:else if isSuspended}
<div class="flex gap-2 flex-col" in:fade={{ duration: 150 }}>
{#each Object.values(suspendStatus.val) as suspendCount (suspendCount.job.id)}
frontend/src/lib/components/runs/JobStatusIcon.svelte
+2 -1
View File
@@ -13,6 +13,7 @@
X
} from 'lucide-svelte'
import type { Job } from '$lib/gen'
import { isSuspended } from '../agentRequest'
interface Props {
job: Job
@@ -60,7 +61,7 @@
<Badge color="red" {roundedFull} baseClass={roundedFull ? '' : '!px-1.5'}>
<X size={14} />
</Badge>
{:else if 'running' in job && job.running && job.suspend}
{:else if 'running' in job && job.running && isSuspended(job)}
<Badge color="violet" {roundedFull} baseClass={roundedFull ? '' : '!px-1.5'} title="Suspended">
<Hourglass size={14} />
</Badge>
openflow.openapi.yaml
+71 -1
View File
@@ -1008,7 +1008,7 @@ components:
description: Unique identifier for this tool. Cannot contain spaces - use underscores instead (e.g., 'get_user_data' not 'get user data')
summary:
type: string
description: "The name the AI agent calls this tool by, not a human label. On a flowmodule tool it must match ^[a-zA-Z0-9_]+$ - letters, numbers and underscores only (e.g. 'search_documentation', not 'Search documentation') - and always be set; on an mcp or websearch tool it is a plain label. Put the human-readable explanation in 'description'."
description: "The name the AI agent calls this tool by, not a human label. On a flowmodule tool it must match ^[a-zA-Z0-9_]+$ - letters, numbers and underscores only (e.g. 'search_documentation', not 'Search documentation') - and always be set; on an mcp, websearch or question tool it is a plain label. Put the human-readable explanation in 'description'."
description:
type: string
description: Free-text description of the tool given to the AI to decide when and how to call it. Overrides the description auto-derived from the underlying script.
@@ -1018,6 +1018,9 @@ components:
items:
type: string
description: "Scopes the token of this tool's jobs is restricted to, on top of the agent step's own restriction: a tool can only narrow it."
requires_approval:
type: boolean
description: "Flowmodule and mcp tools only. Each call waits for a person to approve or reject it from the flow's run page or chat, without the agent holding a worker. Calls in the same turn that do not require approval run right away. A rejected call, or one nobody answers within 30 minutes, is reported to the AI as not run."
value:
$ref: '#/components/schemas/ToolValue'
required:
@@ -1030,12 +1033,14 @@ components:
- $ref: '#/components/schemas/FlowModuleTool'
- $ref: '#/components/schemas/McpToolValue'
- $ref: '#/components/schemas/WebsearchToolValue'
- $ref: '#/components/schemas/QuestionToolValue'
discriminator:
propertyName: tool_type
mapping:
flowmodule: '#/components/schemas/FlowModuleTool'
mcp: '#/components/schemas/McpToolValue'
websearch: '#/components/schemas/WebsearchToolValue'
question: '#/components/schemas/QuestionToolValue'
FlowModuleTool:
description: A tool implemented as a flow module (script, flow, etc.). The AI can call this like any other flow module
@@ -1061,6 +1066,20 @@ components:
required:
- tool_type
QuestionToolValue:
type: object
description: "Built-in tool the AI calls as ask_question: the agent stops, releasing its worker, until someone answers from the flow's run page or chat or the question times out, then resumes with the answer as the tool result."
properties:
tool_type:
type: string
enum:
- question
timeout:
type: integer
description: Seconds to wait for an answer before the question is reported to the AI as timed out. Defaults to 1800.
required:
- tool_type
McpToolValue:
type: object
description: Reference to an external MCP (Model Context Protocol) tool. The AI can call tools from MCP servers
@@ -1494,6 +1513,57 @@ components:
enum: [web_search]
required:
- type
- type: object
properties:
type:
type: string
enum: [approval]
call_id:
type: string
calls:
type: array
items:
type: object
properties:
function_name:
type: string
arguments: {}
required:
- function_name
- arguments
expires_at:
type: string
format: date-time
required:
- type
- call_id
- calls
- expires_at
- type: object
properties:
type:
type: string
enum: [question]
call_id:
type: string
question:
type: string
choices:
type: array
items:
type: string
description: answers the model proposes; a free-text answer is always possible
multi_select:
type: boolean
description: whether several choices can be picked together
expires_at:
type: string
format: date-time
required:
- type
- call_id
- question
- expires_at
- type: object
properties:
type:
system_prompts/auto-generated/flow.md
+8 -5
View File
File diff suppressed because one or more lines are too long
system_prompts/auto-generated/prompts.ts
+8 -5
View File
File diff suppressed because one or more lines are too long
system_prompts/auto-generated/skills/write-flow/SKILL.md
+8 -5
View File
File diff suppressed because one or more lines are too long
system_prompts/base/flow-base.md
+7 -4
View File
@@ -68,7 +68,10 @@ value:
An `aiagent` module runs an LLM that can call tools. Each entry of `value.tools` is a module-shaped
object with an extra `value.tool_type`: `flowmodule` for a script/flow tool, `mcp` for an MCP server
tool, `websearch` for web search.
tool, `websearch` for web search, `question` for a built-in `ask_question` tool (the agent stops,
without holding a worker, until someone answers from the run page or the chat; optional
`value.timeout` in seconds, default 1800). Set `requires_approval: true` on a `flowmodule` or `mcp`
tool to have a person approve or reject each of its calls before it runs.
```json
{
@@ -164,9 +167,9 @@ needs becomes unreachable.
### Tool Naming Rules
These rules cover `flowmodule` tools, the ones the agent calls by name. A `websearch` tool's
`summary` is a plain label (`Web Search`), and an `mcp` tool exposes the MCP server's own tool
names, so neither is name-checked at all — leave those summaries as they are.
These rules cover `flowmodule` tools, the ones the agent calls by name. A `websearch` or `question`
tool's `summary` is a plain label (`Web Search`), and an `mcp` tool exposes the MCP
server's own tool names, so none is name-checked at all — leave those summaries as they are.
- A flowmodule tool's `summary` is the **name the agent calls it by**, not a human label. Put the
human-readable explanation in `description`