MythEclipse/zesdex
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: implement workflow findings sharing for inter-agent communication in workflows

Browse Source
This commit is contained in:
asepharyana
2026-07-13 03:12:37 +07:00
parent dc0f1c7647
commit a080957c26
11 changed files with 154 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/app/runtime/actions/mod.rs
+16 -1
View File
@@ -886,6 +886,11 @@ fn archive_message(db: &Option<std::sync::Arc<std::sync::Mutex<rusqlite::Connect
}
}
/// Maximum number of LLM call + tool-execution iterations per single
/// agent turn before bailing. Prevents runaway token consumption when
/// the agent gets stuck in a loop (e.g. an unachievable todo item).
const MAX_TURN_STEPS: usize = 10000;
/// Execute one full agent turn: stream the conversation to the LLM,
/// handle tool calls, and loop until the LLM produces a non-tool response
/// or runs out of unfinished todo items.
@@ -932,7 +937,17 @@ fn run_agent_turn(
msgs.insert(0, sys);
}
let mut turn_step = 0usize;
loop {
turn_step += 1;
if turn_step > MAX_TURN_STEPS {
anyhow::bail!(
"turn exceeded maximum steps ({}) — possible runaway loop. \
aborting to prevent excessive token usage",
MAX_TURN_STEPS,
);
}
let total_chars: usize = msgs.iter()
.filter_map(|m| m.content.as_deref())
.map(|c| c.len())
@@ -1423,7 +1438,7 @@ fn spawn_api_connectivity_check(state: &AppStateRest) {
let turn_events = state.turn_events.clone();
std::thread::spawn(move || {
let url = format!("{}/models", base_url.trim_end_matches('/'));
let url = format!("{}/chat/completions", base_url.trim_end_matches('/'));
let connected = match reqwest::blocking::Client::builder()
.timeout(std::time::Duration::from_secs(5))
.connect_timeout(std::time::Duration::from_secs(3))
src/app/state/rest.rs
+1
View File
@@ -297,6 +297,7 @@ impl AppStateRest {
graduated_checks: Vec::new(),
lsp_manager: self.lsp_manager.clone(),
turn_events: Some(self.turn_events.clone()),
workflow_findings: None,
}
}
}
src/app/subagent/context.rs
+8 -1
View File
@@ -2,19 +2,25 @@
//! including the default read-only tool set for reviewer agents.
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use super::spawn::AgentDefinition;
/// Default read-only tool names granted to `role == "reviewer"` agents.
pub const REVIEWER_ALLOWED: &[&str] = &["read", "grep", "glob", "recall", "remember"];
/// Per-invocation configuration for a subagent: prompt, allowed tools,
/// step budget, and the session directory it should operate against.
/// step budget, session directory, and optional workflow-findings Arc
/// for cross-agent communication within a workflow run.
pub struct SubagentContext {
pub system_prompt: String,
pub allowed_tools: Vec<String>,
pub max_steps: usize,
pub session_dir: PathBuf,
pub workspaces: Vec<PathBuf>,
/// Ephemeral findings shared between sibling subagents in the same
/// workflow run. Set by the workflow engine; `note_finding` writes
/// into this from tool code via `ToolCtx.workflow_findings`.
pub workflow_findings: Option<Arc<Mutex<Vec<String>>>>,
}
/// Build a `SubagentContext` from an `AgentDefinition`.
@@ -41,5 +47,6 @@ pub fn build_subagent_context(def: AgentDefinition) -> SubagentContext {
max_steps,
session_dir: PathBuf::new(),
workspaces: Vec::new(),
workflow_findings: None,
}
}
src/app/subagent/engine.rs
+1
View File
@@ -96,6 +96,7 @@ pub fn run_subagent(ctx: SubagentContext, tx: mpsc::Sender<SubagentEvent>) -> an
.session_dir(ctx.session_dir.clone())
.workspaces(ctx.workspaces.clone())
.origin(crate::app::state::types::Origin::SubAgent)
.workflow_findings(ctx.workflow_findings.clone())
.build();
// Build tool list once before the loop
src/app/workflow/engine.rs
+64 -30
View File
@@ -9,14 +9,16 @@
//! the previous one.
//! - `run_workflow_tracked` accepts a `LiveState` callback that receives
//! real-time agent status updates for the TUI panel.
//! - Findings (inter-agent notes) are scoped per invocation via an
//! `Arc<Mutex<Vec<String>>>` threaded through `execute_primitive` and
//! `spawn_single_agent` rather than a global static, preventing data
//! leaks between concurrent workflow runs.
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use serde::{Deserialize, Serialize};
use super::script::{ScriptPrimitive, WorkflowScript};
static FINDINGS: Mutex<Vec<String>> = Mutex::new(Vec::new());
/// The lifecycle state of an agent within a workflow run.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum AgentState {
@@ -73,9 +75,14 @@ pub type LiveStateFn = Arc<dyn Fn(String, AgentStatus) + Send + Sync>;
/// after to reflect Running → Completed/Failed transitions.
///
/// Flow: push agent as `Running` → build SubagentContext with prompt +
/// findings preamble → call `run_subagent` (draining the event channel into
/// a throwaway consumer so events are not blocked) → push `Completed` or
/// `Failed`.
/// findings preamble, linking the `workflow_findings` Arc so the subagent's
/// `note_finding` tool pushes into the same vec → call `run_subagent`
/// (draining the event channel into a consumer so events are not blocked)
/// → push `Completed` or `Failed`.
///
/// Why: the `workflow_findings` Arc is shared by all agents within the same
/// `execute_primitive` scope, so pipeline stages can pass data between each
/// other while different workflow invocations remain isolated.
///
/// Return: the agent's text output, or an error on failure.
fn spawn_single_agent(
@@ -83,6 +90,7 @@ fn spawn_single_agent(
agent_name: &str,
prompt: &str,
findings_snapshot: Vec<String>,
findings: &Arc<Mutex<Vec<String>>>,
live: Option<&LiveStateFn>,
session_dir: &std::path::Path,
workspaces: &[std::path::PathBuf],
@@ -124,6 +132,9 @@ fn spawn_single_agent(
};
ctx.system_prompt = format!("{}{}", prompt, findings_section);
// Link the shared findings Arc so note_finding calls within this
// subagent write into the same vec visible to sibling agents.
ctx.workflow_findings = Some(findings.clone());
// Create an mpsc channel and drain events in a background thread so
// run_subagent's blocking_send never blocks (previously the _rx was
@@ -131,10 +142,32 @@ fn spawn_single_agent(
// on a closed channel).
let (tx, rx) = tokio::sync::mpsc::channel(64);
let _drain_thread = std::thread::spawn(move || {
// Drain all events; we don't surface them individually to the UI
// (the live state callbacks handle coarse-grained status).
// Drain all events so run_subagent's blocking_send never blocks.
// Individual SubagentEvent items are not surfaced to the TUI —
// the live state callbacks above handle coarse-grained Running /
// Completed / Failed status. ToolCall / ToolResult / StepCompleted
// events are traced at debug level for observability.
use crate::app::subagent::event::SubagentEvent;
let mut rx = rx;
while rx.blocking_recv().is_some() {}
while let Some(event) = rx.blocking_recv() {
match &event {
SubagentEvent::ToolCall { _tool, _args } => {
tracing::debug!("[subagent] tool call: {}", _tool);
}
SubagentEvent::ToolResult { _tool, .. } => {
tracing::debug!("[subagent] tool result: {}", _tool);
}
SubagentEvent::StepCompleted { _step, .. } => {
tracing::trace!("[subagent] step {} completed", _step);
}
SubagentEvent::StepFailed { _step, _error } => {
tracing::warn!("[subagent] step {} failed: {}", _step, _error);
}
SubagentEvent::Completed { .. } => {
tracing::debug!("[subagent] completed");
}
}
}
});
let result = run_subagent(ctx, tx);
@@ -176,7 +209,9 @@ type ParallelResult = (usize, anyhow::Result<Vec<String>>);
///
/// Why: `Parallel` uses OS threads + a semaphore so the main async event
/// loop remains responsive. `Pipeline` is sequential so each stage sees
/// findings deposited by the previous one.
/// findings deposited by the previous one. Findings are scoped to an
/// `Arc<Mutex<Vec<String>>>` rather than a global static, so concurrent
/// workflow runs are isolated from each other.
///
/// Return: a `Vec<String>` of all agent outputs (or error strings) in
/// the order they were submitted.
@@ -188,14 +223,15 @@ pub fn execute_primitive(
live: Option<&LiveStateFn>,
session_dir: &std::path::Path,
workspaces: &[std::path::PathBuf],
findings: &Arc<Mutex<Vec<String>>>,
) -> anyhow::Result<Vec<String>> {
match primitive {
ScriptPrimitive::Agent(prompt) => {
let resolved = resolve_template(prompt, args);
let findings_snapshot = FINDINGS.lock().map(|f| f.clone()).unwrap_or_default();
let findings_snapshot = findings.lock().map(|f| f.clone()).unwrap_or_default();
let agent_id = uuid::Uuid::new_v4().to_string();
let agent_name = resolved.chars().take(40).collect::<String>();
match spawn_single_agent(&agent_id, &agent_name, &resolved, findings_snapshot, live, session_dir, workspaces) {
match spawn_single_agent(&agent_id, &agent_name, &resolved, findings_snapshot, findings, live, session_dir, workspaces) {
Ok(text) => Ok(vec![text]),
Err(e) => {
if continue_on_error {
@@ -211,6 +247,8 @@ pub fn execute_primitive(
// All branches run concurrently, capped by semaphore.
// This is the primary advantage over single-turn chat: multiple
// independent subagents work simultaneously.
// Each branch shares the same `findings` Arc so note_finding
// calls within any branch are visible to all other branches.
let semaphore = Arc::new(Semaphore::new(concurrency_cap.max(1)));
let results: Arc<Mutex<Vec<ParallelResult>>> =
Arc::new(Mutex::new(Vec::new()));
@@ -227,6 +265,7 @@ pub fn execute_primitive(
let live_clone = live.cloned();
let session_dir = session_dir.to_path_buf();
let workspaces = workspaces.to_vec();
let findings = Arc::clone(findings);
std::thread::spawn(move || {
let _permit = sem.acquire();
@@ -235,6 +274,7 @@ pub fn execute_primitive(
live_clone.as_ref(),
&session_dir,
&workspaces,
&findings,
);
if let Ok(mut locked) = results.lock() {
locked.push((idx, result));
@@ -264,11 +304,11 @@ pub fn execute_primitive(
//
// Why: parallel execution defeats the purpose of a pipeline whose
// stages are supposed to build on each other's output. Findings
// written by stage N are visible to stage N+1 because we share
// the global FINDINGS mutex.
// written by stage N are visible to stage N+1 through the shared
// `findings` Arc (same isolation scope as parent).
let mut all = Vec::new();
for (idx, script) in scripts.iter().enumerate() {
match execute_primitive(script, args, concurrency_cap, continue_on_error, live, session_dir, workspaces) {
match execute_primitive(script, args, concurrency_cap, continue_on_error, live, session_dir, workspaces, findings) {
Ok(outputs) => all.extend(outputs),
Err(e) => {
if continue_on_error {
@@ -283,7 +323,7 @@ pub fn execute_primitive(
}
ScriptPrimitive::Phase { name: _name, script } => {
execute_primitive(script, args, concurrency_cap, continue_on_error, live, session_dir, workspaces)
execute_primitive(script, args, concurrency_cap, continue_on_error, live, session_dir, workspaces, findings)
}
}
}
@@ -304,8 +344,13 @@ pub fn run_workflow(
/// Run a `WorkflowScript` with real-time live-state callbacks so the TUI
/// panel updates as each agent transitions between Idle/Running/Done/Failed.
///
/// Flow: clear the global finding store → cap concurrency to 8 → call
/// `execute_primitive` with the live callback → format results.
/// Flow: create an empty findings Arc (scoped to this invocation) → cap
/// concurrency to 8 → call `execute_primitive` with the live callback and
/// findings → format results.
///
/// Why: findings are scoped to an `Arc<Mutex<Vec<String>>>` rather than a
/// global static, so concurrent `run_workflow_tracked` calls from different
/// spawn_agents invocations remain fully isolated.
///
/// Return: a human-readable summary string.
pub fn run_workflow_tracked(
@@ -315,10 +360,6 @@ pub fn run_workflow_tracked(
session_dir: &std::path::Path,
workspaces: &[std::path::PathBuf],
) -> anyhow::Result<String> {
if let Ok(mut findings) = FINDINGS.lock() {
findings.clear();
}
let concurrency_cap = if script.options.max_concurrency > 0 {
script.options.max_concurrency.min(10) // allow up to 10 parallel agents
} else {
@@ -326,10 +367,11 @@ pub fn run_workflow_tracked(
};
let live_ref = live.as_ref();
let findings = Arc::new(Mutex::new(Vec::new()));
let results = execute_primitive(
&script.script, args, concurrency_cap,
script.options.continue_on_error, live_ref,
session_dir, workspaces,
session_dir, workspaces, &findings,
)?;
let summary = if results.is_empty() {
@@ -351,14 +393,6 @@ pub fn run_workflow_tracked(
Ok(summary)
}
/// Add a finding text to the global workflow findings list, making it
/// visible to sibling agents spawned later in the same run.
pub fn note_finding(text: &str) {
if let Ok(mut findings) = FINDINGS.lock() {
findings.push(text.to_string());
}
}
/// Simple template engine: replace `{{key}}` placeholders with values
/// from `args`.
///
src/main.rs
-1
View File
@@ -564,7 +564,6 @@ fn run_attach(session_id: &str) -> Result<()> {
let _ = disable_raw_mode();
let _ = client_state.settings.save();
core::mem::drop(_rt);
Ok(())
}
src/service/provider.rs
+22 -4
View File
@@ -30,11 +30,13 @@ impl LlmClient {
/// Construct a client, falling back to built-in defaults for empty inputs.
///
/// Flow: empty api_key/model → substitute defaults → build reqwest client
/// with connect/request timeouts (falling back to an untimed client if
/// the builder fails) → normalize base_url.
/// with connect/request timeouts → if TLS config fails, retry with just
/// request timeout (no connect timeout) → normalize base_url.
///
/// Why: empty strings are treated as "unset" rather than errors so callers
/// can pass through unconfigured settings without special-casing them.
/// Timeouts are always enforced — the pure-default-client fallback is only
/// used as a last resort when even the no-connect-timeout build fails.
pub fn new(mut api_key: String, model: String, base_url: Option<String>) -> Self {
if api_key.is_empty() {
api_key = DEFAULT_API_KEY.to_string();
@@ -51,8 +53,24 @@ impl LlmClient {
{
Ok(c) => c,
Err(e) => {
tracing::warn!("warning: failed to build reqwest client with timeouts: {}. Using default client without timeouts.", e);
reqwest::blocking::Client::new()
tracing::warn!(
"failed to build reqwest client with connect timeout: {}. \
retrying without connect timeout",
e,
);
match reqwest::blocking::Client::builder()
.timeout(REQUEST_TIMEOUT)
.build()
{
Ok(c) => c,
Err(e2) => {
tracing::warn!(
"also failed: {}. using default client (no configured timeouts)",
e2,
);
reqwest::blocking::Client::new()
}
}
}
};
LlmClient {
src/tool/mod.rs
+13 -1
View File
@@ -38,7 +38,7 @@ pub struct GraduatedCheck {
}
/// Shared execution context passed to every `Tool::run` call: workspace roots, session
/// paths, and cached directory state.
/// paths, cached directory state, and workflow-level findings sharing.
#[derive(Clone)]
pub struct ToolCtx {
pub workspaces: Vec<PathBuf>,
@@ -51,6 +51,12 @@ pub struct ToolCtx {
pub graduated_checks: Vec<GraduatedCheck>,
pub lsp_manager: Arc<Mutex<crate::app::lsp::LspManager>>,
pub turn_events: Option<Arc<Mutex<std::collections::VecDeque<crate::app::state::runtime::TurnEvent>>>>,
/// Ephemeral findings shared between sibling subagents in a workflow run.
/// Set by the workflow engine before spawning subagents; tools like
/// `note_finding` write into this vec so later pipeline stages can
/// reference earlier results. `None` means "not inside a workflow" —
/// `note_finding` becomes a no-op.
pub workflow_findings: Option<Arc<Mutex<Vec<String>>>>,
}
/// Find which graduated checks apply to a given file path/content pair.
@@ -88,6 +94,7 @@ pub struct ToolCtxBuilder {
pub graduated_checks: Vec<GraduatedCheck>,
pub lsp_manager: Arc<Mutex<crate::app::lsp::LspManager>>,
pub turn_events: Option<Arc<Mutex<std::collections::VecDeque<crate::app::state::runtime::TurnEvent>>>>,
pub workflow_findings: Option<Arc<Mutex<Vec<String>>>>,
}
impl Default for ToolCtxBuilder {
@@ -103,6 +110,7 @@ impl Default for ToolCtxBuilder {
graduated_checks: Vec::new(),
lsp_manager: Arc::new(Mutex::new(crate::app::lsp::LspManager::new())),
turn_events: None,
workflow_findings: None,
}
}
}
@@ -117,6 +125,9 @@ impl ToolCtxBuilder {
/// Set the lsp_manager.
#[allow(dead_code)]
pub fn lsp_manager(mut self, v: Arc<Mutex<crate::app::lsp::LspManager>>) -> Self { self.lsp_manager = v; self }
/// Set the workflow-level findings sharing Arc (for subagent-to-subagent
/// communication within a workflow run).
pub fn workflow_findings(mut self, v: Option<Arc<Mutex<Vec<String>>>>) -> Self { self.workflow_findings = v; self }
/// Consume the builder and produce the final `ToolCtx`.
pub fn build(self) -> ToolCtx {
ToolCtx {
@@ -130,6 +141,7 @@ impl ToolCtxBuilder {
graduated_checks: self.graduated_checks,
lsp_manager: self.lsp_manager,
turn_events: self.turn_events,
workflow_findings: self.workflow_findings,
}
}
}
src/tool/spawn.rs
+11 -2
View File
@@ -88,7 +88,7 @@ impl Tool for SpawnAgents {
},
};
use std::sync::Arc;
use std::sync::{Arc, Mutex};
let live: Option<crate::app::workflow::engine::LiveStateFn> = _ctx.turn_events.as_ref().map(|turn_events| {
let turn_events = turn_events.clone();
let f: crate::app::workflow::engine::LiveStateFn = Arc::new(move |agent_id: String, status| {
@@ -104,6 +104,10 @@ impl Tool for SpawnAgents {
f
});
// Create a per-invocation findings scope so subagents spawned
// by this tool call are isolated from any other concurrent
// spawn_agents or workflow_run invocations.
let findings: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
let results = crate::app::workflow::engine::execute_primitive(
&wf.script,
&HashMap::new(),
@@ -112,6 +116,7 @@ impl Tool for SpawnAgents {
live.as_ref(),
&_ctx.session_dir,
&_ctx.workspaces,
&findings,
)?;
format_results(results, "parallel")
}
@@ -173,7 +178,7 @@ impl Tool for SpawnPipeline {
},
};
use std::sync::Arc;
use std::sync::{Arc, Mutex};
let live: Option<crate::app::workflow::engine::LiveStateFn> = _ctx.turn_events.as_ref().map(|turn_events| {
let turn_events = turn_events.clone();
let f: crate::app::workflow::engine::LiveStateFn = Arc::new(move |agent_id: String, status| {
@@ -189,6 +194,9 @@ impl Tool for SpawnPipeline {
f
});
// Per-invocation findings scope isolates this pipeline from any
// other concurrent spawn_agents / spawn_pipeline / workflow_run.
let findings: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
let results = crate::app::workflow::engine::execute_primitive(
&wf.script,
&HashMap::new(),
@@ -197,6 +205,7 @@ impl Tool for SpawnPipeline {
live.as_ref(),
&_ctx.session_dir,
&_ctx.workspaces,
&findings,
)?;
format_results(results, "pipeline")
}
src/tool/workflow.rs
+18 -7
View File
@@ -109,21 +109,32 @@ impl Tool for NoteFinding {
/// Record `text` as a finding visible to sibling agents in the run.
///
/// Flow: extract `text` argument → forward to
/// `app::workflow::engine::note_finding` → return a truncated
/// confirmation echo.
/// Flow: extract `text` argument → push into
/// `ctx.workflow_findings` (the per-invocation Arc threaded through
/// `execute_primitive`) → return a truncated confirmation echo.
///
/// Why: findings are ephemeral (not persisted to memory) and are
/// meant to be prepended to sibling agents' next tool-round context.
/// Why: findings are scoped per workflow invocation, not global,
/// so concurrent workflow runs are isolated from each other.
/// If no workflow findings Arc is set (called outside a workflow),
/// the call is silently ignored.
///
/// Return: confirmation string containing up to the first 80 chars
/// of the recorded text.
fn run(&self, _ctx: &ToolCtx, args: &Value) -> Result<String> {
fn run(&self, ctx: &ToolCtx, args: &Value) -> Result<String> {
let text = args.get("text")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow!("missing required argument: text"))?;
crate::app::workflow::engine::note_finding(text);
if let Some(ref findings) = ctx.workflow_findings {
if let Ok(mut f) = findings.lock() {
f.push(text.to_string());
}
} else {
tracing::debug!(
"[note_finding] called outside a workflow run — discarding: {}",
text.chars().take(80).collect::<String>(),
);
}
Ok(format!("finding recorded: {}", text.chars().take(80).collect::<String>()))
}
}