refactor: drop agents from workflow (tailcallhq#1483)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: ForgeCode <noreply@forgecode.dev>

Author: 3 people

AGENTS.md

@@ -75,6 +75,7 @@ This document contains guidelines and best practices for AI agents working with
   assert_eq(actual.b, expected.b);
 
 ## Verification
+Always verify changes by running tests and linting the codebase
 
 1. Run crate specific tests to ensure they pass.
 

Cargo.lock

@@ -113,6 +113,7 @@ humantime = "2.2.0"
 gray_matter = "0.3.2"
 num-format = "0.4"
 
+
 # Internal crates
 forge_api = { path = "crates/forge_api" }
 forge_app = { path = "crates/forge_app" }

Cargo.toml

@@ -20,6 +20,8 @@ pub trait API: Sync + Send {
 
  /// Provides a list of models available in the current environment
  async fn models(&self) -> Result<Vec<Model>>;
+ /// Provides a list of agents available in the current environment
+ async fn get_agents(&self) -> Result<Vec<Agent>>;
 
  /// Executes a chat request and returns a stream of responses
  async fn chat(&self, chat: ChatRequest) -> Result<MpscStream<Result<ChatResponse>>>;

crates/forge_api/src/api.rs

@@ -4,9 +4,9 @@ use std::sync::Arc;
 use anyhow::{Context, Result};
 use forge_app::dto::{AppConfig, InitAuth};
 use forge_app::{
- AppConfigService, AuthService, ConversationService, EnvironmentService, FileDiscoveryService,
- ForgeApp, McpConfigManager, ProviderRegistry, ProviderService, Services, User, UserUsage,
- Walker, WorkflowService,
+ AgentLoaderService, AppConfigService, AuthService, ConversationService, EnvironmentService,
+ FileDiscoveryService, ForgeApp, McpConfigManager, ProviderRegistry, ProviderService, Services,
+ User, UserUsage, Walker, WorkflowService,
 };
 use forge_domain::*;
 use forge_infra::ForgeInfra;
@@ -53,6 +53,9 @@ impl<A: Services, F: CommandInfra> API for ForgeAPI<A, F> {
  .models(self.provider().await.context("User is not logged in")?)
  .await?)
  }
+ async fn get_agents(&self) -> Result<Vec<Agent>> {
+ Ok(self.services.get_agents().await?)
+ }
 
  async fn chat(
  &self,
@@ -67,11 +70,14 @@ impl<A: Services, F: CommandInfra> API for ForgeAPI<A, F> {
  &self,
  workflow: W,
  ) -> anyhow::Result<Conversation> {
- self.services.create_conversation(workflow.into()).await
+ let agents = self.get_agents().await?;
+ self.services
+ .init_conversation(workflow.into(), agents)
+ .await
  }
 
  async fn upsert_conversation(&self, conversation: Conversation) -> anyhow::Result<()> {
- self.services.upsert(conversation).await
+ self.services.upsert_conversation(conversation).await
  }
 
  async fn compact_conversation(
@@ -112,7 +118,7 @@ impl<A: Services, F: CommandInfra> API for ForgeAPI<A, F> {
  &self,
  conversation_id: &ConversationId,
  ) -> anyhow::Result<Option<Conversation>> {
- self.services.find(conversation_id).await
+ self.services.find_conversation(conversation_id).await
  }
 
  async fn execute_shell_command(

crates/forge_api/src/forge_api.rs

@@ -73,6 +73,6 @@ impl<T: Services> AgentService for T {
  }
 
  async fn update(&self, conversation: Conversation) -> anyhow::Result<()> {
- self.upsert(conversation).await
+ self.upsert_conversation(conversation).await
  }
 }

crates/forge_app/src/agent.rs

@@ -10,7 +10,7 @@ use futures::StreamExt;
 use tokio::sync::RwLock;
 
 use crate::error::Error;
-use crate::{ConversationService, Services, WorkflowService};
+use crate::{AgentLoaderService, ConversationService, Services, WorkflowService};
 
 #[derive(Clone)]
 pub struct AgentExecutor<S> {
@@ -24,15 +24,14 @@ impl<S: Services> AgentExecutor<S> {
  }
 
  /// Returns a list of tool definitions for all available agents.
- pub async fn tool_agents(&self) -> anyhow::Result<Vec<ToolDefinition>> {
+ pub async fn agent_definitions(&self) -> anyhow::Result<Vec<ToolDefinition>> {
  if let Some(tool_agents) = self.tool_agents.read().await.clone() {
  return Ok(tool_agents);
  }
- let workflow = self.services.read_merged(None).await?;
-
- let agents: Vec<ToolDefinition> = workflow.agents.into_iter().map(Into::into).collect();
- *self.tool_agents.write().await = Some(agents.clone());
- Ok(agents)
+ let agents = self.services.get_agents().await?;
+ let tools: Vec<ToolDefinition> = agents.into_iter().map(Into::into).collect();
+ *self.tool_agents.write().await = Some(tools.clone());
+ Ok(tools)
  }
 
  /// Executes an agent tool call by creating a new chat request for the
@@ -54,8 +53,10 @@ impl<S: Services> AgentExecutor<S> {
 
  // Create a new conversation for agent execution
  let workflow = self.services.read_merged(None).await?;
+ let agents = self.services.get_agents().await?;
  let conversation =
- ConversationService::create_conversation(self.services.as_ref(), workflow).await?;
+ ConversationService::init_conversation(self.services.as_ref(), workflow, agents)
+ .await?;
 
  // Execute the request through the ForgeApp
  let app = crate::ForgeApp::new(self.services.clone());
@@ -99,7 +100,7 @@ impl<S: Services> AgentExecutor<S> {
  }
 
  pub async fn contains_tool(&self, tool_name: &ToolName) -> anyhow::Result<bool> {
- let agent_tools = self.tool_agents().await?;
+ let agent_tools = self.agent_definitions().await?;
  Ok(agent_tools.iter().any(|tool| tool.name == *tool_name))
  }
 }

crates/forge_app/src/agent_executor.rs

@@ -11,10 +11,9 @@ use crate::dto::InitAuth;
 use crate::orch::Orchestrator;
 use crate::services::{CustomInstructionsService, TemplateService};
 use crate::tool_registry::ToolRegistry;
-use crate::workflow_manager::WorkflowManager;
 use crate::{
  AppConfigService, AttachmentService, ConversationService, EnvironmentService,
- FileDiscoveryService, ProviderRegistry, ProviderService, Services, Walker,
+ FileDiscoveryService, ProviderRegistry, ProviderService, Services, Walker, WorkflowService,
 };
 
 /// ForgeApp handles the core chat functionality by orchestrating various
@@ -24,7 +23,6 @@ pub struct ForgeApp<S> {
  services: Arc<S>,
  tool_registry: ToolRegistry<S>,
  authenticator: Authenticator<S>,
- workflow_manager: WorkflowManager<S>,
 }
 
 impl<S: Services> ForgeApp<S> {
@@ -33,7 +31,6 @@ impl<S: Services> ForgeApp<S> {
  Self {
  tool_registry: ToolRegistry::new(services.clone()),
  authenticator: Authenticator::new(services.clone()),
- workflow_manager: WorkflowManager::new(services.clone()),
             services,
  }
  }
@@ -48,7 +45,7 @@ impl<S: Services> ForgeApp<S> {
 
  // Get the conversation for the chat request
  let conversation = services
- .find(&chat.conversation_id)
+ .find_conversation(&chat.conversation_id)
  .await
  .unwrap_or_default()
  .expect("conversation for the request should've been created at this point.");
@@ -63,11 +60,7 @@ impl<S: Services> ForgeApp<S> {
  let models = services.models(provider).await?;
 
  // Discover files using the discovery service
- let workflow = self
- .workflow_manager
- .read_merged(None)
- .await
- .unwrap_or_default();
+ let workflow = self.services.read_merged(None).await.unwrap_or_default();
  let max_depth = workflow.max_walker_depth;
  let environment = services.get_environment();
 
@@ -117,15 +110,13 @@ impl<S: Services> ForgeApp<S> {
  let stream = MpscStream::spawn(
             |tx: tokio::sync::mpsc::Sender<Result<ChatResponse, anyhow::Error>>| {
  async move {
- let tx = Arc::new(tx);
-
  // Execute dispatch and always save conversation afterwards
  let mut orch = orch.sender(tx.clone());
  let dispatch_result = orch.chat(chat.event).await;
 
  // Always save conversation using get_conversation()
  let conversation = orch.get_conversation().clone();
- let save_result = services.upsert(conversation).await;
+ let save_result = services.upsert_conversation(conversation).await;
 
  // Send any error to the stream (prioritize dispatch error over save error)
  #[allow(clippy::collapsible_if)]
@@ -153,7 +144,7 @@ impl<S: Services> ForgeApp<S> {
  // Get the conversation
  let mut conversation = self
  .services
- .find(conversation_id)
+ .find_conversation(conversation_id)
  .await?
  .ok_or_else(|| anyhow::anyhow!("Conversation not found: {}", conversation_id))?;
 
@@ -191,7 +182,7 @@ impl<S: Services> ForgeApp<S> {
         conversation.context = Some(compacted_context);
 
  // Save the updated conversation
- self.services.upsert(conversation).await?;
+ self.services.upsert_conversation(conversation).await?;
 
  // Return the compaction metrics
  Ok(CompactionResult::new(
@@ -215,13 +206,13 @@ impl<S: Services> ForgeApp<S> {
  self.authenticator.logout().await
  }
  pub async fn read_workflow(&self, path: Option<&Path>) -> Result<Workflow> {
- self.workflow_manager.read_workflow(path).await
+ self.services.read_workflow(path).await
  }
 
  pub async fn read_workflow_merged(&self, path: Option<&Path>) -> Result<Workflow> {
- self.workflow_manager.read_merged(path).await
+ self.services.read_merged(path).await
  }
  pub async fn write_workflow(&self, path: Option<&Path>, workflow: &Workflow) -> Result<()> {
- self.workflow_manager.write_workflow(path, workflow).await
+ self.services.write_workflow(path, workflow).await
  }
 }

crates/forge_app/src/app.rs

@@ -19,7 +19,6 @@ mod truncation;
 mod user;
 mod utils;
 mod walker;
-mod workflow_manager;
 
 pub use agent::*;
 pub use app::*;

crates/forge_app/src/lib.rs

@@ -12,7 +12,6 @@ use tracing::{debug, info, warn};
 
 use crate::agent::AgentService;
 use crate::compact::Compactor;
-pub type ArcSender = Arc<tokio::sync::mpsc::Sender<anyhow::Result<ChatResponse>>>;
 
 #[derive(Clone, Setters)]
 #[setters(into, strip_option)]