From 141ecc050c4198e9b527cbb2036332277f00d311 Mon Sep 17 00:00:00 2001 From: pasitnusso Date: Tue, 20 Oct 2026 09:13:26 +0700 Subject: [PATCH] feat: per-node tool-offer narrowing opt-out and dynamic tool descriptions --- crates/areev-run-core/src/lib.rs | 4 +- crates/areev-run-core/src/step.rs | 42 ++++-- crates/areev-run-core/src/types.rs | 38 ++++- .../areev-run-core/tests/scheduler_tests.rs | 137 +++++++++++++++++- crates/areev-run/src/manifest.rs | 90 ++++++++++-- crates/areev-run/src/runner.rs | 3 +- docs/run.md | 11 ++ 7 files changed, 293 insertions(+), 32 deletions(-) diff --git a/crates/areev-run-core/src/lib.rs b/crates/areev-run-core/src/lib.rs index d1213cfa..0f5eede7 100644 --- a/crates/areev-run-core/src/lib.rs +++ b/crates/areev-run-core/src/lib.rs @@ -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, }; diff --git a/crates/areev-run-core/src/step.rs b/crates/areev-run-core/src/step.rs index fb19dba0..471cb7ff 100644 --- a/crates/areev-run-core/src/step.rs +++ b/crates/areev-run-core/src/step.rs @@ -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 @@ -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, @@ -1855,19 +1855,32 @@ fn emit_offer_decide( out: &mut Vec, ) -> 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 = @@ -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) -> Option> { +fn narrow_offer( + tools: &[OfferedTool], + probs: &serde_json::Map, + top_k: usize, +) -> Option> { 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())?; @@ -2038,7 +2055,7 @@ fn narrow_offer(tools: &[OfferedTool], probs: &serde_json::Map) - let keep: std::collections::BTreeSet = 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()) @@ -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. @@ -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); diff --git a/crates/areev-run-core/src/types.rs b/crates/areev-run-core/src/types.rs index ae5dff65..1aec0cbb 100644 --- a/crates/areev-run-core/src/types.rs +++ b/crates/areev-run-core/src/types.rs @@ -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, +} + +/// 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). @@ -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 }, + Abstract { + tools: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + narrow: Option, + }, /// 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 diff --git a/crates/areev-run-core/tests/scheduler_tests.rs b/crates/areev-run-core/tests/scheduler_tests.rs index 58ce3ec8..ffb47359 100644 --- a/crates/areev-run-core/tests/scheduler_tests.rs +++ b/crates/areev-run-core/tests/scheduler_tests.rs @@ -1025,8 +1025,13 @@ fn abstract_exec(names: &[&str]) -> Vec { 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, }] } @@ -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 = + 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 = (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 = 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. diff --git a/crates/areev-run/src/manifest.rs b/crates/areev-run/src/manifest.rs index 4a1d4fef..35cfae04 100644 --- a/crates/areev-run/src/manifest.rs +++ b/crates/areev-run/src/manifest.rs @@ -87,6 +87,9 @@ pub struct PinnedTool { /// pin, so existing manifests serialize byte-identically. #[serde(default, skip_serializing_if = "Option::is_none")] pub decide: Option, + /// Tool-offer narrowing configuration for an abstract node (#396). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub narrow: Option, } /// The `executor` a decision node (C3) pins as. @@ -422,6 +425,7 @@ impl RunManifest { read: Some(spec), ask_kind: None, decide: None, + narrow: None, }); continue; } @@ -447,6 +451,7 @@ impl RunManifest { read: None, ask_kind: None, decide: None, + narrow: None, } } else { pin_from_definition(node, &h, &g)? @@ -481,18 +486,28 @@ impl RunManifest { ), }); } - None if llm_available => PinnedTool { - node: node.clone(), - tool_hash: String::new(), - tool_name: node.clone(), - executor: "abstract".into(), - executor_uri: None, - runtime: None, - runtime_limits: None, - capabilities: None, - read: None, - ask_kind: None, - decide: None, + None if llm_available => { + let narrow = parse_narrow( + node, + fields + .and_then(|f| f.get("llm")) + .and_then(|l| l.get(node)) + .and_then(|n| n.get("narrow")), + )?; + PinnedTool { + node: node.clone(), + tool_hash: String::new(), + tool_name: node.clone(), + executor: "abstract".into(), + executor_uri: None, + runtime: None, + runtime_limits: None, + capabilities: None, + read: None, + ask_kind: None, + decide: None, + narrow, + } }, None => return Err(RunError::NoToolLlm { node: node.clone() }), } @@ -673,6 +688,7 @@ impl RunManifest { .map(|p| areev_run_core::OfferedTool { tool_name: p.tool_name.clone(), tool_hash: p.tool_hash.clone(), + description: None, }) .collect(); self.pinned @@ -695,7 +711,10 @@ impl RunManifest { spec, } } - "abstract" => NodeExecutor::Abstract { tools: offered.clone() }, + "abstract" => NodeExecutor::Abstract { + tools: offered.clone(), + narrow: p.narrow.clone(), + }, // Never Host, for the memory-read reason: a decision that // fell through to `--tool-cmd` would let a tool answer it. DECIDE_EXECUTOR => NodeExecutor::Decide { @@ -1065,6 +1084,7 @@ pub fn pin_from_definition( read: None, ask_kind, decide, + narrow: None, }) } @@ -1114,6 +1134,43 @@ fn parse_decide( Ok(out) } +fn parse_narrow( + node: &str, + decl: Option<&serde_json::Value>, +) -> std::result::Result, RunError> { + let bad = |why: String| RunError::InvalidPlan { + why: format!("node '{node}' declares `llm.{node}.narrow`: {why}"), + }; + match decl { + None | Some(serde_json::Value::Null) => Ok(None), + Some(serde_json::Value::Bool(b)) => Ok(Some(areev_run_core::NarrowOffer::Enabled(*b))), + Some(serde_json::Value::Object(o)) => { + for k in o.keys() { + if k != "top_k" { + return Err(bad(format!("unknown key {k:?} (accepted: top_k)"))); + } + } + match o.get("top_k") { + Some(serde_json::Value::Number(n)) => { + if let Some(k) = n.as_u64() { + if k > 0 && k <= 255 { + Ok(Some(areev_run_core::NarrowOffer::Limit { + top_k: k as usize, + })) + } else { + Err(bad(format!("top_k must be between 1 and 255, got {k}"))) + } + } else { + Err(bad("`top_k` must be a positive integer".into())) + } + } + _ => Err(bad("`top_k` must be a positive integer".into())), + } + } + Some(other) => Err(bad(format!("must be a boolean or an object with `top_k`, got {other}"))), + } +} + /// A Client ask a SECOND person must answer — the default, and what every /// Client ask meant before #294. pub const ASK_KIND_APPROVAL: &str = "approval"; @@ -1261,6 +1318,7 @@ mod tests { read: None, ask_kind: None, decide: None, + narrow: None, } } @@ -1284,6 +1342,7 @@ mod tests { read: None, ask_kind: None, decide: None, + narrow: None, }, host_pin("reply_done", "reply_email"), host_pin("reply_rejected", "reply_email"), @@ -1291,7 +1350,7 @@ mod tests { let executors = m.executors(); assert_eq!(executors.len(), 4); match &executors[1] { - NodeExecutor::Abstract { tools } => { + NodeExecutor::Abstract { tools, .. } => { let names: Vec<&str> = tools.iter().map(|t| t.tool_name.as_str()).collect(); assert_eq!(names, vec!["parse_attachments", "reply_email"]); assert_eq!(tools[1].tool_hash, "hash-reply_email"); @@ -1329,6 +1388,7 @@ mod tests { read, ask_kind: None, decide: None, + narrow: None, }; let mut m = bare(); m.pinned = vec![ @@ -1338,7 +1398,7 @@ mod tests { ]; let executors = m.executors(); assert_eq!(executors[1], NodeExecutor::MemoryRead { op: "entity_at".into(), spec }); - let NodeExecutor::Abstract { tools } = &executors[2] else { panic!("{executors:?}") }; + let NodeExecutor::Abstract { tools, .. } = &executors[2] else { panic!("{executors:?}") }; let offered: Vec<&str> = tools.iter().map(|t| t.tool_name.as_str()).collect(); assert_eq!(offered, vec!["parse_attachments"], "a read is never offered to a model"); diff --git a/crates/areev-run/src/runner.rs b/crates/areev-run/src/runner.rs index 95945679..aac3b48b 100644 --- a/crates/areev-run/src/runner.rs +++ b/crates/areev-run/src/runner.rs @@ -2374,7 +2374,8 @@ impl Runner { // (transcript translation + pinned Definitions from // the store), then hand the pure job to the pool. let llm_prepared = if key.kind == areev_run_core::EffectKind::Llm { - let areev_run_core::NodeExecutor::Abstract { tools } = &executor + let areev_run_core::NodeExecutor::Abstract { tools, .. } = + &executor else { return Err(RunError::Storage { detail: "llm effect on a non-abstract node".into(), diff --git a/docs/run.md b/docs/run.md index c6d5addb..97e46bc8 100644 --- a/docs/run.md +++ b/docs/run.md @@ -1939,6 +1939,17 @@ shown is unknown to it: calling one is the unknown-tool re-prompt. An uncalibrated answer, a failure or a table that does not cover every tool leaves the full offer. +A node may opt out of narrowing with `llm..narrow: false`, keeping the full +offer for that node (frozen in the manifest; default `true` preserves narrowing). +A node may also customize the ranking cutoff with `llm..narrow: {top_k: n}` +(where `n` is between 1 and 255), narrowing to the top `n` by probability plus +any tool at `p ≥ 0.05`. + +When dynamic tools are offered (e.g. via `tools_from`), the narrowing request +carries each dynamic tool's `description` (normalized to its first trimmed line) +in the question `criteria` and the `state.tools` list, exactly as a pinned +Definition's `tool_description` is carried. + The narrowed set rides every turn's journaled input as `offer` — `{tools, seq, provider, model, calibrated, latency_ms}`. The driver offers exactly those Definitions, keyed off the journal the way the summarizer's