Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions crates/areev-run-core/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,6 @@ pub use step::{
};
pub use types::{
Ask, Budgets, Command, DecisionRecord, EdgeOutcome, EffectKind, EffectOutcome, EventIn,
FailCause, FoldRecord, JournalKey, NodeExecutor, OfferedTool, RunOutcome, DECIDE_TOOL,
DECIDE_URI, INBOX, PARKED_ASKS,
FailCause, FoldRecord, JournalKey, NarrowOffer, NodeExecutor, OfferedTool, RunOutcome,
DECIDE_TOOL, DECIDE_URI, INBOX, PARKED_ASKS,
};
42 changes: 30 additions & 12 deletions crates/areev-run-core/src/step.rs
Original file line number Diff line number Diff line change
Expand Up @@ -656,7 +656,7 @@ fn handle_llm_outcome(
path: &str,
outcome: &EffectOutcome,
) {
let NodeExecutor::Abstract { tools } = &env.executors[i] else { return };
let NodeExecutor::Abstract { tools, .. } = &env.executors[i] else { return };
let flow_id = flow_key(i, path);
// A summarizer turn resolves differently from a model turn: its result is
// not the node's answer, and any tool calls in it are ignored (it was
Expand Down Expand Up @@ -1661,7 +1661,7 @@ fn dispatch_flow_tool(
// Resolve the offered tool BEFORE booking anything: a resolution set
// that changed across a fork must fail the node loudly, not panic —
// and never leave a booked pending tool with no outstanding key.
let NodeExecutor::Abstract { tools } = &env.executors[i] else { return };
let NodeExecutor::Abstract { tools, .. } = &env.executors[i] else { return };
let Some(tool) = tools.iter().find(|t| t.tool_name == call.tool_name).cloned() else {
fail_abstract(
env,
Expand Down Expand Up @@ -1855,19 +1855,32 @@ fn emit_offer_decide(
out: &mut Vec<Command>,
) -> bool {
let Some(d) = env.decide.as_ref().filter(|d| d.calibrated) else { return false };
let NodeExecutor::Abstract { tools } = &env.executors[i] else { return false };
let NodeExecutor::Abstract { tools, narrow } = &env.executors[i] else { return false };
let narrow = narrow.as_ref().cloned().unwrap_or_default();
if !narrow.is_enabled() {
return false;
}
let top_k = narrow.top_k();
// A choice carries 2..=255 options; past that the question cannot be
// asked, and the full offer is the honest answer.
if tools.len() <= OFFER_TOP_K || tools.len() > 255 {
if tools.len() <= top_k || tools.len() > 255 {
return false;
}
let Some(flow) = st.abstract_flows.get(&flow_key(i, path)) else { return false };
let first = flow.messages.first().and_then(|m| m.get("content")).cloned().unwrap_or(Value::Null);
let describe = |t: &OfferedTool| {
d.tool_descriptions
.get(&t.tool_name)
.filter(|s| !s.trim().is_empty())
.cloned()
t.description
.as_deref()
.map(|s| s.lines().next().unwrap_or_default().trim())
.filter(|s| !s.is_empty())
.map(str::to_string)
.or_else(|| {
d.tool_descriptions
.get(&t.tool_name)
.map(|s| s.lines().next().unwrap_or_default().trim())
.filter(|s| !s.is_empty())
.map(str::to_string)
})
.unwrap_or_else(|| t.tool_name.clone())
};
let criteria: serde_json::Map<String, Value> =
Expand Down Expand Up @@ -2028,7 +2041,11 @@ fn handle_flow_decide(
/// or above [`OFFER_KEEP_P`], returned in MANIFEST order. `None` when the
/// table does not cover every pinned tool with a finite probability — a
/// malformed answer narrows nothing.
fn narrow_offer(tools: &[OfferedTool], probs: &serde_json::Map<String, Value>) -> Option<Vec<String>> {
fn narrow_offer(
tools: &[OfferedTool],
probs: &serde_json::Map<String, Value>,
top_k: usize,
) -> Option<Vec<String>> {
let mut ranked: Vec<(usize, f64)> = Vec::with_capacity(tools.len());
for (idx, t) in tools.iter().enumerate() {
let p = probs.get(&t.tool_name).and_then(Value::as_f64).filter(|p| p.is_finite())?;
Expand All @@ -2038,7 +2055,7 @@ fn narrow_offer(tools: &[OfferedTool], probs: &serde_json::Map<String, Value>) -
let keep: std::collections::BTreeSet<usize> = ranked
.iter()
.enumerate()
.filter(|(rank, (_, p))| *rank < OFFER_TOP_K || *p >= OFFER_KEEP_P)
.filter(|(rank, (_, p))| *rank < top_k || *p >= OFFER_KEEP_P)
.map(|(_, (idx, _))| *idx)
.collect();
Some(keep.into_iter().map(|idx| tools[idx].tool_name.clone()).collect())
Expand All @@ -2052,7 +2069,8 @@ fn resolve_offer_decide(
effect_seq: u32,
outcome: &EffectOutcome,
) {
let NodeExecutor::Abstract { tools } = &env.executors[i] else { return };
let NodeExecutor::Abstract { tools, narrow } = &env.executors[i] else { return };
let top_k = narrow.as_ref().map(|n| n.top_k()).unwrap_or(OFFER_TOP_K);
// Rule 2: only a CALIBRATED answer may omit a tool. Anything else — an
// uncalibrated answer, a failure, a malformed table — keeps the full
// offer, which is what the flow would have had with no backend.
Expand All @@ -2063,7 +2081,7 @@ fn resolve_offer_decide(
result
.pointer("/answers/tool/probabilities")
.and_then(Value::as_object)
.and_then(|probs| narrow_offer(tools, probs))
.and_then(|probs| narrow_offer(tools, probs, top_k))
.filter(|keep| !keep.is_empty() && keep.len() < tools.len())
.map(|keep| {
let mut record = decide_provenance(result);
Expand Down
38 changes: 37 additions & 1 deletion crates/areev-run-core/src/types.rs
Original file line number Diff line number Diff line change
Expand Up @@ -212,6 +212,38 @@ pub enum EventIn {
pub struct OfferedTool {
pub tool_name: String,
pub tool_hash: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
}

/// Configuration for tool-offer narrowing (C1) on an abstract node (#396).
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(untagged)]
pub enum NarrowOffer {
Enabled(bool),
Limit { top_k: usize },
}

impl Default for NarrowOffer {
fn default() -> Self {
Self::Enabled(true)
}
}

impl NarrowOffer {
pub fn is_enabled(&self) -> bool {
match self {
Self::Enabled(b) => *b,
Self::Limit { .. } => true,
}
}

pub fn top_k(&self) -> usize {
match self {
Self::Enabled(_) => crate::step::OFFER_TOP_K,
Self::Limit { top_k } => *top_k,
}
}
}

/// What executes a node (resolved by the driver's V7 freeze).
Expand All @@ -238,7 +270,11 @@ pub enum NodeExecutor {
/// interprets, choosing among the workflow's pinned Host tools. The
/// node runs as an LLM LOOP — turn, tool calls, results, next turn —
/// each effect journaled under its own `effect_seq`.
Abstract { tools: Vec<OfferedTool> },
Abstract {
tools: Vec<OfferedTool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
narrow: Option<NarrowOffer>,
},
/// A subgraph node (Wave 2): the binding resolves to another Workflow
/// grain; the child runs as its own run (own run_id, own journal),
/// linked by `parent_task_id`, and its final context becomes this
Expand Down
137 changes: 136 additions & 1 deletion crates/areev-run-core/tests/scheduler_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1025,8 +1025,13 @@ fn abstract_exec(names: &[&str]) -> Vec<NodeExecutor> {
vec![NodeExecutor::Abstract {
tools: names
.iter()
.map(|n| OfferedTool { tool_name: (*n).into(), tool_hash: "deadbeef".into() })
.map(|n| OfferedTool {
tool_name: (*n).into(),
tool_hash: "deadbeef".into(),
description: None,
})
.collect(),
narrow: None,
}]
}

Expand Down Expand Up @@ -1708,6 +1713,136 @@ fn an_uncalibrated_or_failed_narrowing_offers_everything() {
}
}

#[test]
fn narrow_false_keeps_full_offer_and_skips_decide_ask() {
let names = twelve_tools();
let refs: Vec<&str> = names.iter().map(String::as_str).collect();
let plan = PlanGraph::build(&wf(&["agent"])).unwrap();
let execs = vec![NodeExecutor::Abstract {
tools: refs
.iter()
.map(|n| OfferedTool {
tool_name: (*n).into(),
tool_hash: "deadbeef".into(),
description: None,
})
.collect(),
narrow: Some(areev_run_core::NarrowOffer::Enabled(false)),
}];
let descs: BTreeMap<String, String> =
names.iter().map(|n| (n.clone(), format!("does {n}"))).collect();
let behavior = scripted(vec![llm_text(r#"{"done": true}"#, 100)], &no_decision);
let r = run_with(decide_env(&plan, &execs, &descs, Some(true)), &behavior);
assert_eq!(r.state.outcome(), Some(&RunOutcome::Completed));
assert!(decide_dispatches(&r.commands).is_empty(), "no decide ask emitted");
let turn = llm_input(&r.commands, 0);
assert!(turn.get("offer").is_none(), "full offer kept without narrowing: {turn}");
}

#[test]
fn narrow_limit_sets_top_k_per_node() {
let names = twelve_tools();
let refs: Vec<&str> = names.iter().map(String::as_str).collect();
let plan = PlanGraph::build(&wf(&["agent"])).unwrap();
let execs = vec![NodeExecutor::Abstract {
tools: refs
.iter()
.map(|n| OfferedTool {
tool_name: (*n).into(),
tool_hash: "deadbeef".into(),
description: None,
})
.collect(),
narrow: Some(areev_run_core::NarrowOffer::Limit { top_k: 4 }),
}];
let descs = BTreeMap::new();
let probs = json!({
"t00": 0.40, "t01": 0.25, "t02": 0.15, "t03": 0.10,
"t04": 0.04, "t05": 0.02, "t06": 0.01, "t07": 0.01,
"t08": 0.01, "t09": 0.005, "t10": 0.003, "t11": 0.002,
});
let decide = |ask: &Value| {
assert_eq!(ask["purpose"], "offer");
ok(json!({
"model": "jev-test", "provider": "fake", "calibrated": true, "latency_ms": 3,
"answers": {"tool": {"type": "choice", "choice": "t00",
"probabilities": probs, "confidence": 0.2}},
}))
};
let behavior = scripted(vec![llm_text(r#"{"done": true}"#, 100)], &decide);
let r = run_with(decide_env(&plan, &execs, &descs, Some(true)), &behavior);
assert_eq!(r.state.outcome(), Some(&RunOutcome::Completed));
assert_eq!(decide_dispatches(&r.commands).len(), 1);
let turn = llm_input(&r.commands, 1);
let offered: Vec<&str> =
turn["offer"]["tools"].as_array().unwrap().iter().map(|v| v.as_str().unwrap()).collect();
assert_eq!(offered, vec!["t00", "t01", "t02", "t03"]);

let four_tools: Vec<String> = (0..4).map(|i| format!("t{i:02}")).collect();
let four_refs: Vec<&str> = four_tools.iter().map(String::as_str).collect();
let execs_four = vec![NodeExecutor::Abstract {
tools: four_refs
.iter()
.map(|n| OfferedTool {
tool_name: (*n).into(),
tool_hash: "deadbeef".into(),
description: None,
})
.collect(),
narrow: Some(areev_run_core::NarrowOffer::Limit { top_k: 4 }),
}];
let behavior_no_ask = scripted(vec![llm_text(r#"{"done": true}"#, 100)], &no_decision);
let r_four = run_with(decide_env(&plan, &execs_four, &descs, Some(true)), &behavior_no_ask);
assert!(decide_dispatches(&r_four.commands).is_empty(), "<= top_k is never narrowed");
}

#[test]
fn narrowing_request_carries_dynamic_tool_descriptions() {
let names = twelve_tools();
let plan = PlanGraph::build(&wf(&["agent"])).unwrap();
let tools: Vec<OfferedTool> = names
.iter()
.enumerate()
.map(|(i, n)| OfferedTool {
tool_name: n.clone(),
tool_hash: "deadbeef".into(),
description: match i {
0 => Some(" First line of description \nSecond line ignored".into()),
1 => Some("Dynamic description for t01".into()),
_ => None,
},
})
.collect();
let execs = vec![NodeExecutor::Abstract { tools, narrow: None }];
let mut descs = BTreeMap::new();
descs.insert("t02".into(), "Pinned description for t02".into());

let decide = |ask: &Value| {
assert_eq!(ask["purpose"], "offer");
let criteria = &ask["questions"]["tool"]["criteria"];
assert_eq!(criteria["t00"], "First line of description");
assert_eq!(criteria["t01"], "Dynamic description for t01");
assert_eq!(criteria["t02"], "Pinned description for t02");
assert_eq!(criteria["t03"], "t03");

let state_tools = ask["state"]["tools"].as_array().expect("state.tools is array");
assert_eq!(state_tools[0], json!({"name": "t00", "description": "First line of description"}));
assert_eq!(state_tools[1], json!({"name": "t01", "description": "Dynamic description for t01"}));
assert_eq!(state_tools[2], json!({"name": "t02", "description": "Pinned description for t02"}));
assert_eq!(state_tools[3], json!({"name": "t03", "description": "t03"}));

ok(json!({
"model": "jev-test", "provider": "fake", "calibrated": true, "latency_ms": 3,
"answers": {"tool": {"type": "choice", "choice": "t00",
"probabilities": offer_probabilities(), "confidence": 0.2}},
}))
};
let behavior = scripted(vec![llm_text(r#"{"done": true}"#, 100)], &decide);
let r = run_with(decide_env(&plan, &execs, &descs, Some(true)), &behavior);
assert_eq!(r.state.outcome(), Some(&RunOutcome::Completed));
assert_eq!(decide_dispatches(&r.commands).len(), 1);
}

/// A decision node (C3) dispatches like a Host tool — intent, then dispatch —
/// and its answer lands in state, where an edge branches on it in the frozen
/// grammar.
Expand Down
Loading