use crate::error::RuntimeError;
use crate::storage;
use crate::tool::{BoxFut, Tier, Tool, ToolArgs, ToolCtx, ToolResult};
use crate::value::Value;
use atman_dsl::ast::{Expr, FlowDecl, Literal, Stmt, TypeExpr};
use std::path::Path;
const DEFAULT_SEARCH_LIMIT: usize = 10;
const MAX_SEARCH_LIMIT: usize = 50;
pub struct FlowList;
pub struct FlowSearch;
pub struct FlowDescribe;
#[derive(Clone)]
struct FlowParameter {
name: String,
ty: String,
default: Option<Value>,
}
#[derive(Clone)]
struct FlowEntry {
name: String,
reference: String,
version: String,
summary: String,
params: Vec<FlowParameter>,
}
struct FlowFile {
name: String,
description: String,
flows: Vec<FlowEntry>,
}
struct FlowCatalog {
fingerprint: String,
files: Vec<FlowFile>,
}
impl FlowCatalog {
fn load() -> Result<Self, RuntimeError> {
let config_dir = storage::config_dir()
.map_err(|error| RuntimeError::ToolFailed(format!("flow catalog: {error}")))?;
Self::load_from(&config_dir.join("commands"))
}
fn load_from(commands_dir: &Path) -> Result<Self, RuntimeError> {
let mut files = Vec::new();
if commands_dir.is_dir() {
let read = std::fs::read_dir(commands_dir).map_err(|error| {
RuntimeError::ToolFailed(format!(
"flow catalog: read {}: {error}",
commands_dir.display()
))
})?;
for entry in read.flatten() {
let path = entry.path();
if path.extension().and_then(|extension| extension.to_str()) != Some("at") {
continue;
}
if let Ok(file) = scan_flow_file(&path) {
files.push(file);
}
}
}
files.sort_by(|left, right| left.name.cmp(&right.name));
let mut hasher = blake3::Hasher::new();
for file in &files {
for flow in &file.flows {
hasher.update(flow.reference.as_bytes());
hasher.update(&[0]);
hasher.update(flow.version.as_bytes());
hasher.update(&[0]);
}
}
Ok(Self {
fingerprint: format!("blake3:{}", hasher.finalize().to_hex()),
files,
})
}
fn searchable_entries(&self) -> Vec<&FlowEntry> {
let mut entries = self
.files
.iter()
.flat_map(|file| file.flows.iter())
.filter(|flow| flow.name != "describe")
.collect::<Vec<_>>();
entries.sort_by(|left, right| left.reference.cmp(&right.reference));
entries
}
fn find(&self, flow_ref: &str) -> Option<&FlowEntry> {
if flow_ref.contains('@') {
let normalized = normalize_flow_ref(flow_ref)?;
return self
.files
.iter()
.flat_map(|file| file.flows.iter())
.find(|flow| flow.reference == normalized);
}
let file_name = normalize_flow_file(flow_ref)?;
self.files
.iter()
.find(|file| file.name == file_name)?
.flows
.iter()
.find(|flow| flow.name != "describe")
}
fn legacy_value(&self) -> Value {
Value::List(
self.files
.iter()
.map(|file| {
Value::Struct(vec![
("file".into(), Value::Str(file.name.clone())),
("description".into(), Value::Str(file.description.clone())),
(
"flows".into(),
Value::List(file.flows.iter().map(legacy_flow_value).collect()),
),
])
})
.collect(),
)
}
}
impl Tool for FlowList {
fn name(&self) -> &str {
"flow.list"
}
fn tier(&self) -> Tier {
Tier::Zero
}
fn description(&self) -> Option<&str> {
Some(
"Return the complete flow catalog for compatibility. The result is unbounded; use \
flow.search followed by flow.describe for model-driven discovery.",
)
}
fn input_schema(&self) -> serde_json::Value {
serde_json::json!({"type": "object", "properties": {}})
}
fn call<'a>(&'a self, _args: ToolArgs, _ctx: &'a ToolCtx) -> BoxFut<'a, ToolResult> {
Box::pin(async move { Ok(FlowCatalog::load()?.legacy_value()) })
}
}
impl Tool for FlowSearch {
fn name(&self) -> &str {
"flow.search"
}
fn tier(&self) -> Tier {
Tier::Zero
}
fn description(&self) -> Option<&str> {
Some(
"Search installed DSL flows by ranked keywords without loading the complete catalog into context. \
Results contain an exact flow ref, a source fingerprint, and a short summary. \
Use flow.describe on one result before flow.spawn when its parameters are unknown.",
)
}
fn input_schema(&self) -> serde_json::Value {
serde_json::json!({
"type": "object",
"properties": {
"query": {"type": "string", "description": "Case-insensitive keywords ranked across flow name, ref, and summary. Empty string lists the first page."},
"limit": {"type": "integer", "minimum": 1, "maximum": MAX_SEARCH_LIMIT, "default": DEFAULT_SEARCH_LIMIT},
"cursor": {"type": "string", "minLength": 1, "description": "Pagination only. Omit this field for the first page. For a later page, pass the non-empty next_cursor returned by the immediately preceding flow.search call verbatim; never send an empty or invented value."}
},
"required": ["query"],
"additionalProperties": false
})
}
fn call<'a>(&'a self, args: ToolArgs, _ctx: &'a ToolCtx) -> BoxFut<'a, ToolResult> {
Box::pin(async move {
let query = string_arg(&args, "query", "flow.search")?;
let limit = limit_arg(&args, "flow.search")?;
let cursor = optional_string_arg(&args, "cursor", "flow.search")?;
search_catalog(&FlowCatalog::load()?, query, limit, cursor.as_deref())
})
}
}
impl Tool for FlowDescribe {
fn name(&self) -> &str {
"flow.describe"
}
fn tier(&self) -> Tier {
Tier::Zero
}
fn description(&self) -> Option<&str> {
Some(
"Describe one installed DSL flow. Returns its exact ref, current source fingerprint, \
summary, and parameter contract. Accepts an exact ref or installed flow-file shorthand. \
An optional version rejects stale search results.",
)
}
fn input_schema(&self) -> serde_json::Value {
serde_json::json!({
"type": "object",
"properties": {
"ref": {"type": "string", "description": "Exact flow ref returned by flow.search, or installed flow file such as subagent.at."},
"version": {"type": "string", "minLength": 1, "description": "Staleness guard. Pass the non-empty source fingerprint returned by the current flow.search result verbatim. Omit this field when no fingerprint is available; never send an empty or invented value."}
},
"required": ["ref"],
"additionalProperties": false
})
}
fn call<'a>(&'a self, args: ToolArgs, _ctx: &'a ToolCtx) -> BoxFut<'a, ToolResult> {
Box::pin(async move {
let flow_ref = string_arg(&args, "ref", "flow.describe")?;
let expected_version = optional_string_arg(&args, "version", "flow.describe")?;
describe_catalog_entry(&FlowCatalog::load()?, flow_ref, expected_version.as_deref())
})
}
}
fn search_catalog(
catalog: &FlowCatalog,
query: &str,
limit: usize,
cursor: Option<&str>,
) -> ToolResult {
let normalized_query = query.trim().to_lowercase();
let search_fingerprint = search_fingerprint(&catalog.fingerprint, &normalized_query);
let offset = cursor
.map(|cursor| parse_cursor(cursor, &search_fingerprint))
.transpose()?
.unwrap_or(0);
let mut entries = catalog
.searchable_entries()
.into_iter()
.filter_map(|flow| search_score(flow, &normalized_query).map(|score| (score, flow)))
.collect::<Vec<_>>();
entries.sort_by(|(left_score, left), (right_score, right)| {
right_score
.cmp(left_score)
.then_with(|| left.reference.cmp(&right.reference))
});
if offset > entries.len() {
return Err(RuntimeError::ToolFailed(
"flow.search: cursor offset is outside the result set".into(),
));
}
let end = (offset + limit).min(entries.len());
let items = entries[offset..end]
.iter()
.map(|(_, flow)| {
Value::Struct(vec![
("ref".into(), Value::Str(flow.reference.clone())),
("version".into(), Value::Str(flow.version.clone())),
("summary".into(), Value::Str(flow.summary.clone())),
])
})
.collect();
let next_cursor = if end < entries.len() {
Value::Str(format!("{search_fingerprint}:{end}"))
} else {
Value::Unit
};
Ok(Value::Struct(vec![
("items".into(), Value::List(items)),
("next_cursor".into(), next_cursor),
("total".into(), Value::Int(entries.len() as i64)),
(
"catalog_fingerprint".into(),
Value::Str(catalog.fingerprint.clone()),
),
]))
}
fn search_score(flow: &FlowEntry, query: &str) -> Option<u32> {
if query.is_empty() {
return Some(0);
}
let name = flow.name.to_lowercase();
let reference = flow.reference.to_lowercase();
let summary = flow.summary.to_lowercase();
let mut score = if name.contains(query) || reference.contains(query) || summary.contains(query)
{
100
} else {
0
};
for token in query
.split(|character: char| !character.is_alphanumeric())
.filter(|token| !token.is_empty())
{
if name == token {
score += 24;
} else if name.contains(token) {
score += 12;
}
if reference.contains(token) {
score += 8;
}
if summary.contains(token) {
score += 4;
}
}
(score > 0).then_some(score)
}
fn describe_catalog_entry(
catalog: &FlowCatalog,
flow_ref: &str,
expected_version: Option<&str>,
) -> ToolResult {
let flow = catalog.find(flow_ref).ok_or_else(|| {
RuntimeError::ToolFailed(format!("flow.describe: flow `{flow_ref}` not found"))
})?;
if expected_version.is_some_and(|version| version != flow.version) {
return Err(RuntimeError::ToolFailed(format!(
"flow.describe: stale version for `{}`; search again",
flow.reference
)));
}
Ok(flow_detail_value(flow))
}
fn scan_flow_file(path: &Path) -> Result<FlowFile, RuntimeError> {
let source = std::fs::read_to_string(path).map_err(|error| {
RuntimeError::ToolFailed(format!("flow catalog: read {}: {error}", path.display()))
})?;
let parsed = atman_dsl::parse::parse_file(&source).map_err(|error| {
RuntimeError::ToolFailed(format!("flow catalog: parse {}: {error}", path.display()))
})?;
let file_name = path
.file_name()
.and_then(|name| name.to_str())
.unwrap_or_default()
.to_string();
let description = parsed
.flows
.iter()
.find(|flow| flow.name.name == "describe")
.and_then(extract_return_string_literal)
.unwrap_or_default();
let version = format!("blake3:{}", blake3::hash(source.as_bytes()).to_hex());
let flows = parsed
.flows
.iter()
.map(|flow| flow_entry(&file_name, &description, &version, flow))
.collect();
Ok(FlowFile {
name: file_name,
description,
flows,
})
}
fn flow_entry(file_name: &str, description: &str, version: &str, flow: &FlowDecl) -> FlowEntry {
FlowEntry {
name: flow.name.name.clone(),
reference: format!("{file_name}@{}", flow.name.name),
version: version.to_string(),
summary: description.to_string(),
params: flow
.params
.iter()
.map(|parameter| FlowParameter {
name: parameter.name.name.clone(),
ty: render_type(¶meter.ty),
default: parameter.default.as_ref().map(expr_to_value),
})
.collect(),
}
}
fn flow_detail_value(flow: &FlowEntry) -> Value {
Value::Struct(vec![
("name".into(), Value::Str(flow.name.clone())),
("ref".into(), Value::Str(flow.reference.clone())),
("version".into(), Value::Str(flow.version.clone())),
("summary".into(), Value::Str(flow.summary.clone())),
(
"params".into(),
Value::List(
flow.params
.iter()
.map(|parameter| {
Value::Struct(vec![
("name".into(), Value::Str(parameter.name.clone())),
("type".into(), Value::Str(parameter.ty.clone())),
("required".into(), Value::Bool(parameter.default.is_none())),
(
"default".into(),
parameter.default.clone().unwrap_or(Value::Unit),
),
])
})
.collect(),
),
),
])
}
fn legacy_flow_value(flow: &FlowEntry) -> Value {
Value::Struct(vec![
("name".into(), Value::Str(flow.name.clone())),
("ref".into(), Value::Str(flow.reference.clone())),
(
"params".into(),
Value::List(
flow.params
.iter()
.map(|parameter| {
Value::Struct(vec![
("name".into(), Value::Str(parameter.name.clone())),
("ty".into(), Value::Str(parameter.ty.clone())),
(
"default".into(),
parameter.default.clone().unwrap_or(Value::Unit),
),
])
})
.collect(),
),
),
])
}
fn normalize_flow_ref(flow_ref: &str) -> Option<String> {
let (file, flow) = flow_ref.split_once('@')?;
if file.is_empty() || flow.is_empty() {
return None;
}
let file = normalize_flow_file(file)?;
Some(format!("{file}@{flow}"))
}
fn normalize_flow_file(file: &str) -> Option<String> {
let file = file.trim();
if file.is_empty() {
return None;
}
Some(if file.ends_with(".at") {
file.to_string()
} else {
format!("{file}.at")
})
}
fn search_fingerprint(catalog_fingerprint: &str, query: &str) -> String {
let mut hasher = blake3::Hasher::new();
hasher.update(catalog_fingerprint.as_bytes());
hasher.update(&[0]);
hasher.update(query.as_bytes());
format!("blake3:{}", hasher.finalize().to_hex())
}
fn parse_cursor(cursor: &str, expected_fingerprint: &str) -> Result<usize, RuntimeError> {
let (fingerprint, offset) = cursor.rsplit_once(':').ok_or_else(|| {
RuntimeError::ToolFailed("flow.search: malformed cursor; search again".into())
})?;
if fingerprint != expected_fingerprint {
return Err(RuntimeError::ToolFailed(
"flow.search: stale cursor; search again".into(),
));
}
offset
.parse::<usize>()
.map_err(|_| RuntimeError::ToolFailed("flow.search: malformed cursor; search again".into()))
}
fn string_arg<'a>(args: &'a ToolArgs, name: &str, tool: &str) -> Result<&'a str, RuntimeError> {
match args.named(name) {
Some(Value::Str(value)) => Ok(value),
Some(other) => Err(RuntimeError::ToolFailed(format!(
"{tool}: `{name}` must be a string, got {}",
other.kind_name()
))),
None => Err(RuntimeError::MissingArg(name.to_string())),
}
}
fn optional_string_arg(
args: &ToolArgs,
name: &str,
tool: &str,
) -> Result<Option<String>, RuntimeError> {
match args.named(name) {
Some(Value::Str(value)) if value.trim().is_empty() => Ok(None),
Some(Value::Str(value)) => Ok(Some(value.clone())),
Some(Value::Unit) | None => Ok(None),
Some(other) => Err(RuntimeError::ToolFailed(format!(
"{tool}: `{name}` must be a string, got {}",
other.kind_name()
))),
}
}
fn limit_arg(args: &ToolArgs, tool: &str) -> Result<usize, RuntimeError> {
match args.named("limit") {
None | Some(Value::Unit) => Ok(DEFAULT_SEARCH_LIMIT),
Some(Value::Int(limit)) if (1..=MAX_SEARCH_LIMIT as i64).contains(limit) => {
Ok(*limit as usize)
}
Some(Value::Int(_)) => Err(RuntimeError::ToolFailed(format!(
"{tool}: `limit` must be between 1 and {MAX_SEARCH_LIMIT}"
))),
Some(other) => Err(RuntimeError::ToolFailed(format!(
"{tool}: `limit` must be an integer, got {}",
other.kind_name()
))),
}
}
fn extract_return_string_literal(flow: &FlowDecl) -> Option<String> {
flow.body.iter().find_map(|statement| match statement {
Stmt::Return {
value: Expr::Literal(Literal::Str(value)),
} => Some(value.clone()),
_ => None,
})
}
pub(super) fn render_type(ty: &TypeExpr) -> String {
match ty {
TypeExpr::Named(name) => name.name.clone(),
TypeExpr::List(inner) => format!("[{}]", render_type(inner)),
TypeExpr::Struct(fields) => format!(
"{{{}}}",
fields
.iter()
.map(|(name, ty)| format!("{}: {}", name.name, render_type(ty)))
.collect::<Vec<_>>()
.join(", ")
),
}
}
fn expr_to_value(expr: &Expr) -> Value {
match expr {
Expr::Literal(Literal::Int(value)) => Value::Int(*value),
Expr::Literal(Literal::Float(value)) => Value::Float(*value),
Expr::Literal(Literal::Bool(value)) => Value::Bool(*value),
Expr::Literal(Literal::Str(value)) => Value::Str(value.clone()),
_ => Value::Unit,
}
}
#[cfg(test)]
mod tests {
use super::*;
fn write_flow(dir: &Path, name: &str, source: &str) {
std::fs::write(dir.join(name), source).unwrap();
}
fn next_cursor(value: &Value) -> Option<&str> {
match value.field("next_cursor") {
Some(Value::Str(cursor)) => Some(cursor),
_ => None,
}
}
#[test]
fn optional_discovery_values_treat_blank_strings_as_omitted() {
for value in ["", " ", "\t\n"] {
let args = ToolArgs {
named: vec![("value".into(), Value::Str(value.into()))],
..Default::default()
};
assert_eq!(
optional_string_arg(&args, "value", "flow.test").unwrap(),
None
);
}
let args = ToolArgs {
named: vec![("value".into(), Value::Str("blake3:123".into()))],
..Default::default()
};
assert_eq!(
optional_string_arg(&args, "value", "flow.test").unwrap(),
Some("blake3:123".into())
);
}
#[test]
fn search_is_bounded_and_cursor_is_bound_to_catalog_revision() {
let dir = tempfile::tempdir().unwrap();
write_flow(
dir.path(),
"agents.at",
r#"
flow describe() -> string { return "Delegated work" }
flow alpha(goal: string) -> string { return goal }
flow beta(goal: string) -> string { return goal }
flow gamma(goal: string) -> string { return goal }
"#,
);
let catalog = FlowCatalog::load_from(dir.path()).unwrap();
let first = search_catalog(&catalog, "", 2, None).unwrap();
let cursor = next_cursor(&first).unwrap().to_string();
assert_eq!(
first.field("items").and_then(|items| match items {
Value::List(items) => Some(items.len()),
_ => None,
}),
Some(2)
);
let second = search_catalog(&catalog, "", 2, Some(&cursor)).unwrap();
assert!(next_cursor(&second).is_none());
write_flow(
dir.path(),
"extra.at",
"flow delta(goal: string) -> string { return goal }",
);
let changed = FlowCatalog::load_from(dir.path()).unwrap();
let error = search_catalog(&changed, "", 2, Some(&cursor)).unwrap_err();
assert!(error.to_string().contains("stale cursor"));
}
#[test]
fn describe_returns_parameter_contract_and_rejects_stale_version() {
let dir = tempfile::tempdir().unwrap();
write_flow(
dir.path(),
"review.at",
r#"
flow describe() -> string { return "Review code" }
flow review(goal: string, retries: int = 3) -> string { return goal }
"#,
);
let catalog = FlowCatalog::load_from(dir.path()).unwrap();
let flow = catalog.find("review@review").unwrap();
let described =
describe_catalog_entry(&catalog, "review.at@review", Some(&flow.version)).unwrap();
assert_eq!(
described.field("summary").and_then(as_str),
Some("Review code")
);
let params = match described.field("params") {
Some(Value::List(params)) => params,
_ => panic!("params must be a list"),
};
assert_eq!(params.len(), 2);
assert_eq!(params[0].field("required").and_then(as_bool), Some(true));
assert_eq!(params[1].field("default").and_then(as_int), Some(3));
let error =
describe_catalog_entry(&catalog, "review@review", Some("blake3:stale")).unwrap_err();
assert!(error.to_string().contains("stale version"));
}
#[test]
fn search_ranks_partial_natural_language_matches() {
let dir = tempfile::tempdir().unwrap();
write_flow(
dir.path(),
"subagent.at",
r#"
flow describe() -> string { return "Sub-agent flows for isolated research, verification, implementation, and review. The research role reads files without making changes." }
flow subagent(goal: string, role: string = "research") -> string { return goal }
"#,
);
write_flow(
dir.path(),
"review.at",
r#"
flow describe() -> string { return "Review files" }
flow review(goal: string) -> string { return goal }
"#,
);
let catalog = FlowCatalog::load_from(dir.path()).unwrap();
let result = search_catalog(&catalog, "subagent research read files", 5, None).unwrap();
let items = match result.field("items") {
Some(Value::List(items)) => items,
_ => panic!("items must be a list"),
};
assert_eq!(
items[0].field("ref").and_then(as_str),
Some("subagent.at@subagent")
);
}
#[test]
fn describe_accepts_the_same_file_shorthand_as_spawn() {
let dir = tempfile::tempdir().unwrap();
write_flow(
dir.path(),
"subagent.at",
r#"
flow describe() -> string { return "Delegated work" }
flow subagent(goal: string) -> string { return goal }
flow research_loop(goal: string) -> string { return goal }
"#,
);
let catalog = FlowCatalog::load_from(dir.path()).unwrap();
let described = describe_catalog_entry(&catalog, "subagent.at", None).unwrap();
assert_eq!(
described.field("ref").and_then(as_str),
Some("subagent.at@subagent")
);
}
fn as_str(value: &Value) -> Option<&str> {
match value {
Value::Str(value) => Some(value),
_ => None,
}
}
fn as_bool(value: &Value) -> Option<bool> {
match value {
Value::Bool(value) => Some(*value),
_ => None,
}
}
fn as_int(value: &Value) -> Option<i64> {
match value {
Value::Int(value) => Some(*value),
_ => None,
}
}
}