tkmind/tkmind_go
1
0
Fork 0
Code Issues Pull Requests Actions 4 Packages Projects Releases Wiki Activity

Revert "Mnovich/temporal scheduler (#2745)" (#2839)

Browse Source
This commit is contained in:
Max Novich
2025-06-09 15:41:46 -07:00
committed by GitHub
parent e6f89cde76
commit 996f727ec5
35 changed files with 172 additions and 4896 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/goose-cli/src/cli.rs
+2 -25
View File
@@ -11,8 +11,7 @@ use crate::commands::project::{handle_project_default, handle_projects_interacti
use crate::commands::recipe::{handle_deeplink, handle_validate};
// Import the new handlers from commands::schedule
use crate::commands::schedule::{
handle_schedule_add, handle_schedule_cron_help, handle_schedule_list, handle_schedule_remove,
handle_schedule_run_now, handle_schedule_services_status, handle_schedule_services_stop,
handle_schedule_add, handle_schedule_list, handle_schedule_remove, handle_schedule_run_now,
handle_schedule_sessions,
};
use crate::commands::session::{handle_session_list, handle_session_remove};
@@ -124,11 +123,7 @@ enum SchedulerCommand {
Add {
#[arg(long, help = "Unique ID for the job")]
id: String,
#[arg(
long,
help = "Cron expression for the schedule",
long_help = "Cron expression for when to run the job. Examples:\n '0 * * * *' - Every hour at minute 0\n '0 */2 * * *' - Every 2 hours\n '@hourly' - Every hour (shorthand)\n '0 9 * * *' - Every day at 9:00 AM\n '0 9 * * 1' - Every Monday at 9:00 AM\n '0 0 1 * *' - First day of every month at midnight"
)]
#[arg(long, help = "Cron string for the schedule (e.g., '0 0 * * * *')")]
cron: String,
#[arg(
long,
@@ -160,15 +155,6 @@ enum SchedulerCommand {
#[arg(long, help = "ID of the schedule to run")] // Explicitly make it --id
id: String,
},
/// Check status of Temporal services (temporal scheduler only)
#[command(about = "Check status of Temporal services")]
ServicesStatus {},
/// Stop Temporal services (temporal scheduler only)
#[command(about = "Stop Temporal services")]
ServicesStop {},
/// Show cron expression examples and help
#[command(about = "Show cron expression examples and help")]
CronHelp {},
}
#[derive(Subcommand)]
@@ -782,15 +768,6 @@ pub async fn cli() -> Result<()> {
// New arm
handle_schedule_run_now(id).await?;
}
SchedulerCommand::ServicesStatus {} => {
handle_schedule_services_status().await?;
}
SchedulerCommand::ServicesStop {} => {
handle_schedule_services_stop().await?;
}
SchedulerCommand::CronHelp {} => {
handle_schedule_cron_help().await?;
}
}
return Ok(());
}
crates/goose-cli/src/commands/schedule.rs
+8 -208
View File
@@ -1,11 +1,9 @@
use anyhow::{bail, Context, Result};
use base64::engine::{general_purpose::STANDARD as BASE64_STANDARD, Engine};
use goose::scheduler::{
get_default_scheduled_recipes_dir, get_default_scheduler_storage_path, ScheduledJob,
get_default_scheduled_recipes_dir, get_default_scheduler_storage_path, ScheduledJob, Scheduler,
SchedulerError,
};
use goose::scheduler_factory::SchedulerFactory;
use goose::temporal_scheduler::TemporalScheduler;
use std::path::Path;
// Base64 decoding function - might be needed if recipe_source_arg can be base64
@@ -17,64 +15,6 @@ async fn _decode_base64_recipe(source: &str) -> Result<String> {
String::from_utf8(bytes).with_context(|| "Decoded Base64 recipe source is not valid UTF-8.")
}
fn validate_cron_expression(cron: &str) -> Result<()> {
// Basic validation and helpful suggestions
if cron.trim().is_empty() {
bail!("Cron expression cannot be empty");
}
// Check for common mistakes and provide helpful suggestions
let parts: Vec<&str> = cron.split_whitespace().collect();
match parts.len() {
5 => {
// Standard 5-field cron (minute hour day month weekday)
println!("✅ Using standard 5-field cron format: {}", cron);
}
6 => {
// 6-field cron with seconds (second minute hour day month weekday)
println!("✅ Using 6-field cron format with seconds: {}", cron);
}
1 if cron.starts_with('@') => {
// Shorthand expressions like @hourly, @daily, etc.
let valid_shorthands = [
"@yearly",
"@annually",
"@monthly",
"@weekly",
"@daily",
"@midnight",
"@hourly",
];
if valid_shorthands.contains(&cron) {
println!("✅ Using cron shorthand: {}", cron);
} else {
println!(
"⚠️ Unknown cron shorthand '{}'. Valid options: {}",
cron,
valid_shorthands.join(", ")
);
}
}
_ => {
println!("⚠️ Unusual cron format detected: '{}'", cron);
println!(" Common formats:");
println!(" - 5 fields: '0 * * * *' (minute hour day month weekday)");
println!(" - 6 fields: '0 0 * * * *' (second minute hour day month weekday)");
println!(" - Shorthand: '@hourly', '@daily', '@weekly', '@monthly'");
}
}
// Provide examples for common scheduling needs
if cron == "* * * * *" {
println!("⚠️ This will run every minute! Did you mean:");
println!(" - '0 * * * *' for every hour?");
println!(" - '0 0 * * *' for every day?");
}
Ok(())
}
pub async fn handle_schedule_add(
id: String,
cron: String,
@@ -85,9 +25,6 @@ pub async fn handle_schedule_add(
id, cron, recipe_source_arg
);
// Validate cron expression and provide helpful feedback
validate_cron_expression(&cron)?;
// The Scheduler's add_scheduled_job will handle copying the recipe from recipe_source_arg
// to its internal storage and validating the path.
let job = ScheduledJob {
@@ -103,7 +40,7 @@ pub async fn handle_schedule_add(
let scheduler_storage_path =
get_default_scheduler_storage_path().context("Failed to get scheduler storage path")?;
let scheduler = SchedulerFactory::create(scheduler_storage_path)
let scheduler = Scheduler::new(scheduler_storage_path)
.await
.context("Failed to initialize scheduler")?;
@@ -148,28 +85,19 @@ pub async fn handle_schedule_add(
pub async fn handle_schedule_list() -> Result<()> {
let scheduler_storage_path =
get_default_scheduler_storage_path().context("Failed to get scheduler storage path")?;
let scheduler = SchedulerFactory::create(scheduler_storage_path)
let scheduler = Scheduler::new(scheduler_storage_path)
.await
.context("Failed to initialize scheduler")?;
let jobs = scheduler.list_scheduled_jobs().await?;
let jobs = scheduler.list_scheduled_jobs().await;
if jobs.is_empty() {
println!("No scheduled jobs found.");
} else {
println!("Scheduled Jobs:");
for job in jobs {
let status = if job.currently_running {
"🟢 RUNNING"
} else if job.paused {
"⏸️ PAUSED"
} else {
"⏹️ IDLE"
};
println!(
"- ID: {}\n Status: {}\n Cron: {}\n Recipe Source (in store): {}\n Last Run: {}",
"- ID: {}\n Cron: {}\n Recipe Source (in store): {}\n Last Run: {}",
job.id,
status,
job.cron,
job.source, // This source is now the path within scheduled_recipes_dir
job.last_run
@@ -183,7 +111,7 @@ pub async fn handle_schedule_list() -> Result<()> {
pub async fn handle_schedule_remove(id: String) -> Result<()> {
let scheduler_storage_path =
get_default_scheduler_storage_path().context("Failed to get scheduler storage path")?;
let scheduler = SchedulerFactory::create(scheduler_storage_path)
let scheduler = Scheduler::new(scheduler_storage_path)
.await
.context("Failed to initialize scheduler")?;
@@ -205,7 +133,7 @@ pub async fn handle_schedule_remove(id: String) -> Result<()> {
pub async fn handle_schedule_sessions(id: String, limit: Option<u32>) -> Result<()> {
let scheduler_storage_path =
get_default_scheduler_storage_path().context("Failed to get scheduler storage path")?;
let scheduler = SchedulerFactory::create(scheduler_storage_path)
let scheduler = Scheduler::new(scheduler_storage_path)
.await
.context("Failed to initialize scheduler")?;
@@ -238,7 +166,7 @@ pub async fn handle_schedule_sessions(id: String, limit: Option<u32>) -> Result<
pub async fn handle_schedule_run_now(id: String) -> Result<()> {
let scheduler_storage_path =
get_default_scheduler_storage_path().context("Failed to get scheduler storage path")?;
let scheduler = SchedulerFactory::create(scheduler_storage_path)
let scheduler = Scheduler::new(scheduler_storage_path)
.await
.context("Failed to initialize scheduler")?;
@@ -258,131 +186,3 @@ pub async fn handle_schedule_run_now(id: String) -> Result<()> {
}
Ok(())
}
pub async fn handle_schedule_services_status() -> Result<()> {
// Check if we're using temporal scheduler
let scheduler_type =
std::env::var("GOOSE_SCHEDULER_TYPE").unwrap_or_else(|_| "temporal".to_string());
if scheduler_type != "temporal" {
println!("Service management is only available for temporal scheduler.");
println!("Set GOOSE_SCHEDULER_TYPE=temporal to use Temporal services.");
return Ok(());
}
println!("Checking Temporal services status...");
// Create a temporary TemporalScheduler to check status
match TemporalScheduler::new().await {
Ok(scheduler) => {
let info = scheduler.get_service_info().await;
println!("{}", info);
}
Err(e) => {
println!("Failed to check services: {}", e);
println!("\nThis might mean:");
println!("- Temporal CLI is not installed");
println!("- Go service binary is not built");
println!("- Services are not running");
}
}
Ok(())
}
pub async fn handle_schedule_services_stop() -> Result<()> {
// Check if we're using temporal scheduler
let scheduler_type =
std::env::var("GOOSE_SCHEDULER_TYPE").unwrap_or_else(|_| "temporal".to_string());
if scheduler_type != "temporal" {
println!("Service management is only available for temporal scheduler.");
println!("Set GOOSE_SCHEDULER_TYPE=temporal to use Temporal services.");
return Ok(());
}
println!("Stopping Temporal services...");
// Create a temporary TemporalScheduler to stop services
match TemporalScheduler::new().await {
Ok(scheduler) => match scheduler.stop_services().await {
Ok(result) => {
println!("{}", result);
println!("\nNote: Services were running independently and have been stopped.");
println!("They will be automatically restarted when needed.");
}
Err(e) => {
println!("Failed to stop services: {}", e);
}
},
Err(e) => {
println!("Failed to initialize scheduler: {}", e);
println!("Services may not be running or may have already been stopped.");
}
}
Ok(())
}
pub async fn handle_schedule_cron_help() -> Result<()> {
println!("📅 Cron Expression Guide for Goose Scheduler");
println!("===========================================\\n");
println!("🕐 HOURLY SCHEDULES (Most Common Request):");
println!(" 0 * * * * - Every hour at minute 0 (e.g., 1:00, 2:00, 3:00...)");
println!(" 30 * * * * - Every hour at minute 30 (e.g., 1:30, 2:30, 3:30...)");
println!(" 0 */2 * * * - Every 2 hours at minute 0 (e.g., 2:00, 4:00, 6:00...)");
println!(" 0 */3 * * * - Every 3 hours at minute 0 (e.g., 3:00, 6:00, 9:00...)");
println!(" @hourly - Every hour (same as \"0 * * * *\")\\n");
println!("📅 DAILY SCHEDULES:");
println!(" 0 9 * * * - Every day at 9:00 AM");
println!(" 30 14 * * * - Every day at 2:30 PM");
println!(" 0 0 * * * - Every day at midnight");
println!(" @daily - Every day at midnight\\n");
println!("📆 WEEKLY SCHEDULES:");
println!(" 0 9 * * 1 - Every Monday at 9:00 AM");
println!(" 0 17 * * 5 - Every Friday at 5:00 PM");
println!(" 0 0 * * 0 - Every Sunday at midnight");
println!(" @weekly - Every Sunday at midnight\\n");
println!("🗓️ MONTHLY SCHEDULES:");
println!(" 0 9 1 * * - First day of every month at 9:00 AM");
println!(" 0 0 15 * * - 15th of every month at midnight");
println!(" @monthly - First day of every month at midnight\\n");
println!("📝 CRON FORMAT:");
println!(" Standard 5-field: minute hour day month weekday");
println!(" ┌───────────── minute (0 - 59)");
println!(" │ ┌─────────── hour (0 - 23)");
println!(" │ │ ┌───────── day of month (1 - 31)");
println!(" │ │ │ ┌─────── month (1 - 12)");
println!(" │ │ │ │ ┌───── day of week (0 - 7, Sunday = 0 or 7)");
println!(" │ │ │ │ │");
println!(" * * * * *\\n");
println!("🔧 SPECIAL CHARACTERS:");
println!(" * - Any value (every minute, hour, day, etc.)");
println!(" */n - Every nth interval (*/5 = every 5 minutes)");
println!(" n-m - Range (1-5 = 1,2,3,4,5)");
println!(" n,m - List (1,3,5 = 1 or 3 or 5)\\n");
println!("⚡ SHORTHAND EXPRESSIONS:");
println!(" @yearly - Once a year (0 0 1 1 *)");
println!(" @monthly - Once a month (0 0 1 * *)");
println!(" @weekly - Once a week (0 0 * * 0)");
println!(" @daily - Once a day (0 0 * * *)");
println!(" @hourly - Once an hour (0 * * * *)\\n");
println!("💡 EXAMPLES:");
println!(
" goose schedule add --id hourly-report --cron \"0 * * * *\" --recipe-source report.yaml"
);
println!(
" goose schedule add --id daily-backup --cron \"@daily\" --recipe-source backup.yaml"
);
println!(" goose schedule add --id weekly-summary --cron \"0 9 * * 1\" --recipe-source summary.yaml");
Ok(())
}
crates/goose-scheduler-executor/Cargo.toml
-16
View File
@@ -1,16 +0,0 @@
[package]
name = "goose-scheduler-executor"
version = "0.1.0"
edition = "2021"
[dependencies]
goose = { path = "../goose" }
mcp-core = { path = "../mcp-core" }
anyhow = "1.0"
tokio = { version = "1.0", features = ["full"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
clap = { version = "4.0", features = ["derive"] }
futures = "0.3"
serde_json = "1.0"
serde_yaml = "0.9"
crates/goose-scheduler-executor/src/main.rs
-212
View File
@@ -1,212 +0,0 @@
use anyhow::{anyhow, Result};
use clap::Parser;
use goose::agents::{Agent, SessionConfig};
use goose::config::Config;
use goose::message::Message;
use goose::providers::create;
use goose::recipe::Recipe;
use goose::session;
use std::env;
use std::fs;
use std::path::Path;
use tracing::info;
#[derive(Parser, Debug)]
#[command(author, version, about, long_about = None)]
struct Args {
/// Job ID for the scheduled job
job_id: String,
/// Path to the recipe file to execute
recipe_path: String,
}
#[tokio::main]
async fn main() -> Result<()> {
// Initialize tracing
tracing_subscriber::fmt()
.with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
.init();
let args = Args::parse();
info!("Starting goose-scheduler-executor for job: {}", args.job_id);
info!("Recipe path: {}", args.recipe_path);
// Execute the recipe and get session ID
let session_id = execute_recipe(&args.job_id, &args.recipe_path).await?;
// Output session ID to stdout (this is what the Go service expects)
println!("{}", session_id);
Ok(())
}
async fn execute_recipe(job_id: &str, recipe_path: &str) -> Result<String> {
let recipe_path_buf = Path::new(recipe_path);
// Check if recipe file exists
if !recipe_path_buf.exists() {
return Err(anyhow!("Recipe file not found: {}", recipe_path));
}
// Read and parse recipe
let recipe_content = fs::read_to_string(recipe_path_buf)?;
let recipe: Recipe = {
let extension = recipe_path_buf
.extension()
.and_then(|os_str| os_str.to_str())
.unwrap_or("yaml")
.to_lowercase();
match extension.as_str() {
"json" | "jsonl" => serde_json::from_str::<Recipe>(&recipe_content)
.map_err(|e| anyhow!("Failed to parse JSON recipe '{}': {}", recipe_path, e))?,
"yaml" | "yml" => serde_yaml::from_str::<Recipe>(&recipe_content)
.map_err(|e| anyhow!("Failed to parse YAML recipe '{}': {}", recipe_path, e))?,
_ => {
return Err(anyhow!(
"Unsupported recipe file extension '{}' for: {}",
extension,
recipe_path
));
}
}
};
// Create agent
let agent = Agent::new();
// Get provider configuration
let global_config = Config::global();
let provider_name: String = global_config.get_param("GOOSE_PROVIDER").map_err(|_| {
anyhow!("GOOSE_PROVIDER not configured. Run 'goose configure' or set env var.")
})?;
let model_name: String = global_config.get_param("GOOSE_MODEL").map_err(|_| {
anyhow!("GOOSE_MODEL not configured. Run 'goose configure' or set env var.")
})?;
let model_config = goose::model::ModelConfig::new(model_name);
let provider = create(&provider_name, model_config)
.map_err(|e| anyhow!("Failed to create provider '{}': {}", provider_name, e))?;
// Set provider on agent
agent
.update_provider(provider)
.await
.map_err(|e| anyhow!("Failed to set provider on agent: {}", e))?;
info!(
"Agent configured with provider '{}' for job '{}'",
provider_name, job_id
);
// Generate session ID
let session_id = session::generate_session_id();
// Check if recipe has a prompt
let Some(prompt_text) = recipe.prompt else {
info!(
"Recipe '{}' has no prompt to execute for job '{}'",
recipe_path, job_id
);
// Create empty session for consistency
let session_file_path = goose::session::storage::get_path(
goose::session::storage::Identifier::Name(session_id.clone()),
);
let metadata = goose::session::storage::SessionMetadata {
working_dir: env::current_dir().unwrap_or_default(),
description: "Empty job - no prompt".to_string(),
schedule_id: Some(job_id.to_string()),
message_count: 0,
..Default::default()
};
goose::session::storage::save_messages_with_metadata(&session_file_path, &metadata, &[])
.map_err(|e| anyhow!("Failed to persist metadata for empty job: {}", e))?;
return Ok(session_id);
};
// Create session configuration
let current_dir =
env::current_dir().map_err(|e| anyhow!("Failed to get current directory: {}", e))?;
let session_config = SessionConfig {
id: goose::session::storage::Identifier::Name(session_id.clone()),
working_dir: current_dir.clone(),
schedule_id: Some(job_id.to_string()),
};
// Execute the recipe
let mut messages = vec![Message::user().with_text(prompt_text)];
info!("Executing recipe for job '{}' with prompt", job_id);
let mut stream = agent
.reply(&messages, Some(session_config))
.await
.map_err(|e| anyhow!("Agent failed to reply for recipe '{}': {}", recipe_path, e))?;
// Process the response stream
use futures::StreamExt;
use goose::agents::AgentEvent;
while let Some(message_result) = stream.next().await {
match message_result {
Ok(AgentEvent::Message(msg)) => {
if msg.role == mcp_core::role::Role::Assistant {
info!("[Job {}] Assistant response received", job_id);
}
messages.push(msg);
}
Ok(AgentEvent::McpNotification(_)) => {
// Handle notifications if needed
}
Err(e) => {
return Err(anyhow!("Error receiving message from agent: {}", e));
}
}
}
// Save session
let session_file_path = goose::session::storage::get_path(
goose::session::storage::Identifier::Name(session_id.clone()),
);
// Try to read updated metadata, or create fallback
match goose::session::storage::read_metadata(&session_file_path) {
Ok(mut updated_metadata) => {
updated_metadata.message_count = messages.len();
goose::session::storage::save_messages_with_metadata(
&session_file_path,
&updated_metadata,
&messages,
)
.map_err(|e| anyhow!("Failed to persist final messages: {}", e))?;
}
Err(_) => {
let fallback_metadata = goose::session::storage::SessionMetadata {
working_dir: current_dir,
description: format!("Scheduled job: {}", job_id),
schedule_id: Some(job_id.to_string()),
message_count: messages.len(),
..Default::default()
};
goose::session::storage::save_messages_with_metadata(
&session_file_path,
&fallback_metadata,
&messages,
)
.map_err(|e| anyhow!("Failed to persist messages with fallback metadata: {}", e))?;
}
}
info!(
"Finished executing job '{}', session: {}",
job_id, session_id
);
Ok(session_id)
}
crates/goose-server/src/commands/agent.rs
+2 -2
View File
@@ -6,7 +6,7 @@ use anyhow::Result;
use etcetera::{choose_app_strategy, AppStrategy};
use goose::agents::Agent;
use goose::config::APP_STRATEGY;
use goose::scheduler_factory::SchedulerFactory;
use goose::scheduler::Scheduler as GooseScheduler;
use tower_http::cors::{Any, CorsLayer};
use tracing::info;
@@ -28,7 +28,7 @@ pub async fn run() -> Result<()> {
.data_dir()
.join("schedules.json");
let scheduler_instance = SchedulerFactory::create(schedule_file_path).await?;
let scheduler_instance = GooseScheduler::new(schedule_file_path).await?;
app_state.set_scheduler(scheduler_instance).await;
let cors = CorsLayer::new()
crates/goose-server/src/routes/config_management.rs
+1 -1
View File
@@ -472,7 +472,7 @@ mod tests {
.unwrap()
.data_dir()
.join("schedules.json");
let sched = goose::scheduler_factory::SchedulerFactory::create_legacy(sched_storage_path)
let sched = goose::scheduler::Scheduler::new(sched_storage_path)
.await
.unwrap();
test_state.set_scheduler(sched).await;
crates/goose-server/src/routes/reply.rs
+3 -4
View File
@@ -541,10 +541,9 @@ mod tests {
let state = AppState::new(Arc::new(agent), "test-secret".to_string()).await;
let scheduler_path = goose::scheduler::get_default_scheduler_storage_path()
.expect("Failed to get default scheduler storage path");
let scheduler =
goose::scheduler_factory::SchedulerFactory::create_legacy(scheduler_path)
.await
.unwrap();
let scheduler = goose::scheduler::Scheduler::new(scheduler_path)
.await
.unwrap();
state.set_scheduler(scheduler).await;
let app = routes(state);
crates/goose-server/src/routes/schedule.rs
+2 -17
View File
@@ -108,11 +108,6 @@ async fn create_schedule(
.scheduler()
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
tracing::info!(
"Server: Calling scheduler.add_scheduled_job() for job '{}'",
req.id
);
let job = ScheduledJob {
id: req.id,
source: req.recipe_source,
@@ -152,12 +147,7 @@ async fn list_schedules(
.scheduler()
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
tracing::info!("Server: Calling scheduler.list_scheduled_jobs()");
let jobs = scheduler.list_scheduled_jobs().await.map_err(|e| {
eprintln!("Error listing schedules: {:?}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
let jobs = scheduler.list_scheduled_jobs().await;
Ok(Json(ListSchedulesResponse { jobs }))
}
@@ -220,8 +210,6 @@ async fn run_now_handler(
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
tracing::info!("Server: Calling scheduler.run_now() for job '{}'", id);
match scheduler.run_now(&id).await {
Ok(session_id) => Ok(Json(RunNowResponse { session_id })),
Err(e) => {
@@ -420,10 +408,7 @@ async fn update_schedule(
})?;
// Return the updated schedule
let jobs = scheduler.list_scheduled_jobs().await.map_err(|e| {
eprintln!("Error listing schedules after update: {:?}", e);
StatusCode::INTERNAL_SERVER_ERROR
})?;
let jobs = scheduler.list_scheduled_jobs().await;
let updated_job = jobs
.into_iter()
.find(|job| job.id == id)
crates/goose-server/src/state.rs
+4 -4
View File
@@ -1,5 +1,5 @@
use goose::agents::Agent;
use goose::scheduler_trait::SchedulerTrait;
use goose::scheduler::Scheduler;
use std::sync::Arc;
use tokio::sync::Mutex;
@@ -9,7 +9,7 @@ pub type AgentRef = Arc<Agent>;
pub struct AppState {
agent: Option<AgentRef>,
pub secret_key: String,
pub scheduler: Arc<Mutex<Option<Arc<dyn SchedulerTrait>>>>,
pub scheduler: Arc<Mutex<Option<Arc<Scheduler>>>>,
}
impl AppState {
@@ -27,12 +27,12 @@ impl AppState {
.ok_or_else(|| anyhow::anyhow!("Agent needs to be created first."))
}
pub async fn set_scheduler(&self, sched: Arc<dyn SchedulerTrait>) {
pub async fn set_scheduler(&self, sched: Arc<Scheduler>) {
let mut guard = self.scheduler.lock().await;
*guard = Some(sched);
}
pub async fn scheduler(&self) -> Result<Arc<dyn SchedulerTrait>, anyhow::Error> {
pub async fn scheduler(&self) -> Result<Arc<Scheduler>, anyhow::Error> {
self.scheduler
.lock()
.await
crates/goose/src/lib.rs
-3
View File
@@ -8,10 +8,7 @@ pub mod prompt_template;
pub mod providers;
pub mod recipe;
pub mod scheduler;
pub mod scheduler_factory;
pub mod scheduler_trait;
pub mod session;
pub mod temporal_scheduler;
pub mod token_counter;
pub mod tool_monitor;
pub mod tracing;
crates/goose/src/scheduler.rs
-56
View File
@@ -5,7 +5,6 @@ use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::{anyhow, Result};
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use etcetera::{choose_app_strategy, AppStrategy};
use serde::{Deserialize, Serialize};
@@ -19,7 +18,6 @@ use crate::message::Message;
use crate::providers::base::Provider as GooseProvider; // Alias to avoid conflict in test section
use crate::providers::create;
use crate::recipe::Recipe;
use crate::scheduler_trait::SchedulerTrait;
use crate::session;
use crate::session::storage::SessionMetadata;
@@ -1373,57 +1371,3 @@ mod tests {
Ok(())
}
}
#[async_trait]
impl SchedulerTrait for Scheduler {
async fn add_scheduled_job(&self, job: ScheduledJob) -> Result<(), SchedulerError> {
self.add_scheduled_job(job).await
}
async fn list_scheduled_jobs(&self) -> Result<Vec<ScheduledJob>, SchedulerError> {
Ok(self.list_scheduled_jobs().await)
}
async fn remove_scheduled_job(&self, id: &str) -> Result<(), SchedulerError> {
self.remove_scheduled_job(id).await
}
async fn pause_schedule(&self, id: &str) -> Result<(), SchedulerError> {
self.pause_schedule(id).await
}
async fn unpause_schedule(&self, id: &str) -> Result<(), SchedulerError> {
self.unpause_schedule(id).await
}
async fn run_now(&self, id: &str) -> Result<String, SchedulerError> {
self.run_now(id).await
}
async fn sessions(
&self,
sched_id: &str,
limit: usize,
) -> Result<Vec<(String, SessionMetadata)>, SchedulerError> {
self.sessions(sched_id, limit).await
}
async fn update_schedule(
&self,
sched_id: &str,
new_cron: String,
) -> Result<(), SchedulerError> {
self.update_schedule(sched_id, new_cron).await
}
async fn kill_running_job(&self, sched_id: &str) -> Result<(), SchedulerError> {
self.kill_running_job(sched_id).await
}
async fn get_running_job_info(
&self,
sched_id: &str,
) -> Result<Option<(String, DateTime<Utc>)>, SchedulerError> {
self.get_running_job_info(sched_id).await
}
}
crates/goose/src/scheduler_factory.rs
-75
View File
@@ -1,75 +0,0 @@
use std::path::PathBuf;
use std::sync::Arc;
use crate::config::Config;
use crate::scheduler::{Scheduler, SchedulerError};
use crate::scheduler_trait::SchedulerTrait;
use crate::temporal_scheduler::TemporalScheduler;
pub enum SchedulerType {
Legacy,
Temporal,
}
impl SchedulerType {
/// Determine scheduler type from configuration
pub fn from_config() -> Self {
let config = Config::global();
match config.get_param::<String>("GOOSE_SCHEDULER_TYPE") {
Ok(scheduler_type) => match scheduler_type.to_lowercase().as_str() {
"temporal" => SchedulerType::Temporal,
"legacy" => SchedulerType::Legacy,
_ => {
tracing::warn!(
"Unknown scheduler type '{}', defaulting to legacy",
scheduler_type
);
SchedulerType::Legacy
}
},
Err(_) => {
// Default to temporal scheduler
SchedulerType::Temporal
}
}
}
}
/// Factory for creating scheduler instances
pub struct SchedulerFactory;
impl SchedulerFactory {
/// Create a scheduler instance based on configuration
pub async fn create(storage_path: PathBuf) -> Result<Arc<dyn SchedulerTrait>, SchedulerError> {
let scheduler_type = SchedulerType::from_config();
match scheduler_type {
SchedulerType::Legacy => {
tracing::info!("Creating legacy scheduler");
let scheduler = Scheduler::new(storage_path).await?;
Ok(scheduler as Arc<dyn SchedulerTrait>)
}
SchedulerType::Temporal => {
tracing::info!("Creating Temporal scheduler");
let scheduler = TemporalScheduler::new().await?;
Ok(scheduler as Arc<dyn SchedulerTrait>)
}
}
}
/// Create a specific scheduler type (for testing or explicit use)
pub async fn create_legacy(
storage_path: PathBuf,
) -> Result<Arc<dyn SchedulerTrait>, SchedulerError> {
tracing::info!("Creating legacy scheduler (explicit)");
let scheduler = Scheduler::new(storage_path).await?;
Ok(scheduler as Arc<dyn SchedulerTrait>)
}
/// Create a Temporal scheduler (for testing or explicit use)
pub async fn create_temporal() -> Result<Arc<dyn SchedulerTrait>, SchedulerError> {
tracing::info!("Creating Temporal scheduler (explicit)");
let scheduler = TemporalScheduler::new().await?;
Ok(scheduler as Arc<dyn SchedulerTrait>)
}
}
crates/goose/src/scheduler_trait.rs
-47
View File
@@ -1,47 +0,0 @@
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use crate::scheduler::{ScheduledJob, SchedulerError};
use crate::session::storage::SessionMetadata;
/// Common trait for all scheduler implementations
#[async_trait]
pub trait SchedulerTrait: Send + Sync {
/// Add a new scheduled job
async fn add_scheduled_job(&self, job: ScheduledJob) -> Result<(), SchedulerError>;
/// List all scheduled jobs
async fn list_scheduled_jobs(&self) -> Result<Vec<ScheduledJob>, SchedulerError>;
/// Remove a scheduled job by ID
async fn remove_scheduled_job(&self, id: &str) -> Result<(), SchedulerError>;
/// Pause a scheduled job
async fn pause_schedule(&self, id: &str) -> Result<(), SchedulerError>;
/// Unpause a scheduled job
async fn unpause_schedule(&self, id: &str) -> Result<(), SchedulerError>;
/// Run a job immediately
async fn run_now(&self, id: &str) -> Result<String, SchedulerError>;
/// Get sessions for a scheduled job
async fn sessions(
&self,
sched_id: &str,
limit: usize,
) -> Result<Vec<(String, SessionMetadata)>, SchedulerError>;
/// Update a schedule's cron expression
async fn update_schedule(&self, sched_id: &str, new_cron: String)
-> Result<(), SchedulerError>;
/// Kill a running job
async fn kill_running_job(&self, sched_id: &str) -> Result<(), SchedulerError>;
/// Get information about a running job
async fn get_running_job_info(
&self,
sched_id: &str,
) -> Result<Option<(String, DateTime<Utc>)>, SchedulerError>;
}
crates/goose/src/temporal_scheduler.rs
-870
View File
@@ -1,870 +0,0 @@
use std::process::Command;
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use reqwest::Client;
use serde::{Deserialize, Serialize};
use tokio::time::sleep;
use tracing::{info, warn};
use crate::scheduler::{ScheduledJob, SchedulerError};
use crate::scheduler_trait::SchedulerTrait;
use crate::session::storage::SessionMetadata;
const TEMPORAL_SERVICE_URL: &str = "http://localhost:8080";
const TEMPORAL_SERVICE_STARTUP_TIMEOUT: Duration = Duration::from_secs(30);
const TEMPORAL_SERVICE_HEALTH_CHECK_INTERVAL: Duration = Duration::from_secs(2);
#[derive(Serialize, Deserialize, Debug)]
struct JobRequest {
action: String,
job_id: Option<String>,
cron: Option<String>,
recipe_path: Option<String>,
}
#[derive(Serialize, Deserialize, Debug)]
struct JobResponse {
success: bool,
message: String,
jobs: Option<Vec<TemporalJobStatus>>,
data: Option<serde_json::Value>,
}
#[derive(Serialize, Deserialize, Debug)]
struct TemporalJobStatus {
id: String,
cron: String,
recipe_path: String,
last_run: Option<String>,
next_run: Option<String>,
currently_running: bool,
paused: bool,
created_at: String,
}
#[derive(Serialize, Deserialize, Debug)]
struct RunNowResponse {
session_id: String,
}
pub struct TemporalScheduler {
http_client: Client,
service_url: String,
}
impl TemporalScheduler {
pub async fn new() -> Result<Arc<Self>, SchedulerError> {
let http_client = Client::new();
let service_url = TEMPORAL_SERVICE_URL.to_string();
let scheduler = Arc::new(Self {
http_client,
service_url,
});
// Check if services are running, start them if needed
scheduler.ensure_services_running().await?;
// Wait for service to be ready
scheduler.wait_for_service_ready().await?;
info!("TemporalScheduler initialized successfully");
Ok(scheduler)
}
async fn ensure_services_running(&self) -> Result<(), SchedulerError> {
info!("Checking if Temporal services are running...");
// First, check if both services are already running
let temporal_running = self.check_temporal_server().await;
let go_service_running = self.health_check().await.unwrap_or(false);
if temporal_running && go_service_running {
info!("Both Temporal server and Go service are already running");
return Ok(());
}
// If Go service is running but Temporal server is not, this is an unusual state
if go_service_running && !temporal_running {
warn!("Go service is running but Temporal server is not - this may indicate a configuration issue");
return Err(SchedulerError::SchedulerInternalError(
"Go service is running but Temporal server is not accessible. Please check your Temporal server configuration.".to_string()
));
}
// If Temporal server is running but Go service is not, start the Go service
if temporal_running && !go_service_running {
info!("Temporal server is running, starting Go service...");
self.start_go_service().await?;
return Ok(());
}
// If neither is running, start both
if !temporal_running {
info!("Starting Temporal server...");
self.start_temporal_server().await?;
// Wait for Temporal server to be ready
self.wait_for_temporal_server().await?;
}
// Now start the Go service
if !self.health_check().await.unwrap_or(false) {
info!("Starting Temporal Go service...");
self.start_go_service().await?;
}
Ok(())
}
async fn check_temporal_server(&self) -> bool {
// Temporal server uses gRPC on port 7233, not HTTP
// We should check the web UI port (8233) instead, or use a different method
// First try the web UI (which uses HTTP)
if let Ok(response) = self.http_client.get("http://localhost:8233/").send().await {
if response.status().is_success() {
return true;
}
}
// Alternative: check if we can establish a TCP connection to the gRPC port
use std::net::SocketAddr;
use std::time::Duration;
let addr: SocketAddr = "127.0.0.1:7233".parse().unwrap();
match std::net::TcpStream::connect_timeout(&addr, Duration::from_secs(2)) {
Ok(_) => {
info!("Detected Temporal server on port 7233 (gRPC connection successful)");
true
}
Err(_) => false,
}
}
async fn start_temporal_server(&self) -> Result<(), SchedulerError> {
info!("Starting Temporal server in background...");
// Check if port 7233 is already in use
if self.check_port_in_use(7233).await {
// Port is in use - check if it's a Temporal server we can connect to
if self.check_temporal_server().await {
info!("Port 7233 is in use by a Temporal server we can connect to");
return Ok(());
} else {
return Err(SchedulerError::SchedulerInternalError(
"Port 7233 is already in use by something other than a Temporal server."
.to_string(),
));
}
}
let output = Command::new("sh")
.arg("-c")
.arg("nohup temporal server start-dev --db-filename temporal.db --port 7233 --ui-port 8233 --log-level warn > temporal-server.log 2>&1 & echo $!")
.output()
.map_err(|e| SchedulerError::SchedulerInternalError(
format!("Failed to start Temporal server: {}. Make sure 'temporal' CLI is installed.", e)
))?;
if !output.status.success() {
return Err(SchedulerError::SchedulerInternalError(format!(
"Failed to start Temporal server: {}",
String::from_utf8_lossy(&output.stderr)
)));
}
let pid_output = String::from_utf8_lossy(&output.stdout);
let pid = pid_output.trim();
info!("Temporal server started with PID: {}", pid);
Ok(())
}
async fn check_port_in_use(&self, port: u16) -> bool {
use std::net::{SocketAddr, TcpListener};
let addr: SocketAddr = format!("127.0.0.1:{}", port).parse().unwrap();
TcpListener::bind(addr).is_err()
}
async fn wait_for_temporal_server(&self) -> Result<(), SchedulerError> {
info!("Waiting for Temporal server to be ready...");
let start_time = std::time::Instant::now();
while start_time.elapsed() < TEMPORAL_SERVICE_STARTUP_TIMEOUT {
if self.check_temporal_server().await {
info!("Temporal server is ready");
return Ok(());
}
sleep(TEMPORAL_SERVICE_HEALTH_CHECK_INTERVAL).await;
}
Err(SchedulerError::SchedulerInternalError(
"Temporal server failed to become ready within timeout".to_string(),
))
}
async fn start_go_service(&self) -> Result<(), SchedulerError> {
info!("Starting Temporal Go service in background...");
// Check if port 8080 is already in use
if self.check_port_in_use(8080).await {
// Port is in use - check if it's our Go service we can connect to
if self.health_check().await.unwrap_or(false) {
info!("Port 8080 is in use by a Go service we can connect to");
return Ok(());
} else {
return Err(SchedulerError::SchedulerInternalError(
"Port 8080 is already in use by something other than our Go service."
.to_string(),
));
}
}
// Check if the temporal-service binary exists - try multiple possible locations
let binary_path = Self::find_go_service_binary()?;
let working_dir = std::path::Path::new(&binary_path).parent().ok_or_else(|| {
SchedulerError::SchedulerInternalError(
"Could not determine working directory for Go service".to_string(),
)
})?;
info!("Found Go service binary at: {}", binary_path);
info!("Using working directory: {}", working_dir.display());
let command = format!(
"cd '{}' && nohup ./temporal-service > temporal-service.log 2>&1 & echo $!",
working_dir.display()
);
let output = Command::new("sh")
.arg("-c")
.arg(&command)
.output()
.map_err(|e| {
SchedulerError::SchedulerInternalError(format!(
"Failed to start Go temporal service: {}",
e
))
})?;
if !output.status.success() {
return Err(SchedulerError::SchedulerInternalError(format!(
"Failed to start Go service: {}",
String::from_utf8_lossy(&output.stderr)
)));
}
let pid_output = String::from_utf8_lossy(&output.stdout);
let pid = pid_output.trim();
info!("Temporal Go service started with PID: {}", pid);
Ok(())
}
fn find_go_service_binary() -> Result<String, SchedulerError> {
// Try to find the Go service binary by looking for it relative to the current executable
// or in common locations
let possible_paths = vec![
// Relative to current working directory (original behavior)
"./temporal-service/temporal-service",
];
// Also try to find it relative to the current executable path
if let Ok(exe_path) = std::env::current_exe() {
if let Some(exe_dir) = exe_path.parent() {
// Try various relative paths from the executable directory
let exe_relative_paths = vec![
exe_dir.join("temporal-service/temporal-service"),
exe_dir.join("../temporal-service/temporal-service"),
exe_dir.join("../../temporal-service/temporal-service"),
exe_dir.join("../../../temporal-service/temporal-service"),
exe_dir.join("../../../../temporal-service/temporal-service"),
];
for path in exe_relative_paths {
if path.exists() {
return Ok(path.to_string_lossy().to_string());
}
}
}
}
// Try the original relative paths
for path in &possible_paths {
if std::path::Path::new(path).exists() {
return Ok(path.to_string());
}
}
Err(SchedulerError::SchedulerInternalError(
"Go service binary not found. Tried paths relative to current executable and working directory. Please ensure the temporal-service binary is built and available.".to_string()
))
}
async fn wait_for_service_ready(&self) -> Result<(), SchedulerError> {
info!("Waiting for Temporal service to be ready...");
let start_time = std::time::Instant::now();
while start_time.elapsed() < TEMPORAL_SERVICE_STARTUP_TIMEOUT {
match self.health_check().await {
Ok(true) => {
info!("Temporal service is ready");
return Ok(());
}
Ok(false) => {
// Service responded but not healthy
sleep(TEMPORAL_SERVICE_HEALTH_CHECK_INTERVAL).await;
}
Err(_) => {
// Service not responding yet
sleep(TEMPORAL_SERVICE_HEALTH_CHECK_INTERVAL).await;
}
}
}
Err(SchedulerError::SchedulerInternalError(
"Temporal service failed to become ready within timeout".to_string(),
))
}
async fn health_check(&self) -> Result<bool, SchedulerError> {
let url = format!("{}/health", self.service_url);
match self.http_client.get(&url).send().await {
Ok(response) => {
if response.status().is_success() {
Ok(true)
} else {
Ok(false)
}
}
Err(_) => Ok(false),
}
}
pub async fn add_scheduled_job(&self, job: ScheduledJob) -> Result<(), SchedulerError> {
tracing::info!(
"TemporalScheduler: add_scheduled_job() called for job '{}'",
job.id
);
let request = JobRequest {
action: "create".to_string(),
job_id: Some(job.id.clone()),
cron: Some(job.cron.clone()),
recipe_path: Some(job.source.clone()),
};
let response = self.make_request(request).await?;
if response.success {
info!("Successfully created scheduled job: {}", job.id);
Ok(())
} else {
Err(SchedulerError::SchedulerInternalError(response.message))
}
}
pub async fn list_scheduled_jobs(&self) -> Result<Vec<ScheduledJob>, SchedulerError> {
tracing::info!("TemporalScheduler: list_scheduled_jobs() called");
let request = JobRequest {
action: "list".to_string(),
job_id: None,
cron: None,
recipe_path: None,
};
let response = self.make_request(request).await?;
if response.success {
let jobs = response.jobs.unwrap_or_default();
let scheduled_jobs = jobs
.into_iter()
.map(|tj| {
ScheduledJob {
id: tj.id,
source: tj.recipe_path,
cron: tj.cron,
last_run: tj.last_run.and_then(|s| s.parse::<DateTime<Utc>>().ok()),
currently_running: tj.currently_running,
paused: tj.paused,
current_session_id: None, // Not provided by Temporal service
process_start_time: None, // Not provided by Temporal service
}
})
.collect();
Ok(scheduled_jobs)
} else {
Err(SchedulerError::SchedulerInternalError(response.message))
}
}
pub async fn remove_scheduled_job(&self, id: &str) -> Result<(), SchedulerError> {
let request = JobRequest {
action: "delete".to_string(),
job_id: Some(id.to_string()),
cron: None,
recipe_path: None,
};
let response = self.make_request(request).await?;
if response.success {
info!("Successfully removed scheduled job: {}", id);
Ok(())
} else {
Err(SchedulerError::SchedulerInternalError(response.message))
}
}
pub async fn pause_schedule(&self, id: &str) -> Result<(), SchedulerError> {
let request = JobRequest {
action: "pause".to_string(),
job_id: Some(id.to_string()),
cron: None,
recipe_path: None,
};
let response = self.make_request(request).await?;
if response.success {
info!("Successfully paused scheduled job: {}", id);
Ok(())
} else {
Err(SchedulerError::SchedulerInternalError(response.message))
}
}
pub async fn unpause_schedule(&self, id: &str) -> Result<(), SchedulerError> {
let request = JobRequest {
action: "unpause".to_string(),
job_id: Some(id.to_string()),
cron: None,
recipe_path: None,
};
let response = self.make_request(request).await?;
if response.success {
info!("Successfully unpaused scheduled job: {}", id);
Ok(())
} else {
Err(SchedulerError::SchedulerInternalError(response.message))
}
}
pub async fn run_now(&self, id: &str) -> Result<String, SchedulerError> {
tracing::info!("TemporalScheduler: run_now() called for job '{}'", id);
let request = JobRequest {
action: "run_now".to_string(),
job_id: Some(id.to_string()),
cron: None,
recipe_path: None,
};
let response = self.make_request(request).await?;
if response.success {
if let Some(data) = response.data {
if let Ok(run_response) = serde_json::from_value::<RunNowResponse>(data) {
info!("Successfully started job execution for: {}", id);
Ok(run_response.session_id)
} else {
Err(SchedulerError::SchedulerInternalError(
"Invalid response format for run_now".to_string(),
))
}
} else {
Err(SchedulerError::SchedulerInternalError(
"No session ID returned from run_now".to_string(),
))
}
} else {
Err(SchedulerError::SchedulerInternalError(response.message))
}
}
// Note: This method fetches sessions from the session storage directly
// since Temporal service doesn't track session metadata
pub async fn sessions(
&self,
sched_id: &str,
limit: usize,
) -> Result<Vec<(String, SessionMetadata)>, SchedulerError> {
use crate::session::storage;
// Get all session files
let all_session_files = storage::list_sessions().map_err(|e| {
SchedulerError::SchedulerInternalError(format!("Failed to list sessions: {}", e))
})?;
let mut schedule_sessions: Vec<(String, SessionMetadata)> = Vec::new();
for (session_name, session_path) in all_session_files {
match storage::read_metadata(&session_path) {
Ok(metadata) => {
// Check if this session belongs to the requested schedule
if metadata.schedule_id.as_deref() == Some(sched_id) {
schedule_sessions.push((session_name, metadata));
}
}
Err(e) => {
tracing::warn!(
"Failed to read metadata for session file {}: {}. Skipping.",
session_path.display(),
e
);
}
}
}
// Sort by session_name (timestamp string) in descending order (newest first)
schedule_sessions.sort_by(|a, b| b.0.cmp(&a.0));
// Take only the requested limit
let result_sessions: Vec<(String, SessionMetadata)> =
schedule_sessions.into_iter().take(limit).collect();
tracing::info!(
"Found {} sessions for schedule '{}'",
result_sessions.len(),
sched_id
);
Ok(result_sessions)
}
pub async fn update_schedule(
&self,
_sched_id: &str,
_new_cron: String,
) -> Result<(), SchedulerError> {
warn!("update_schedule() method not implemented for TemporalScheduler - delete and recreate job instead");
Err(SchedulerError::SchedulerInternalError(
"update_schedule not supported - delete and recreate job instead".to_string(),
))
}
pub async fn kill_running_job(&self, _sched_id: &str) -> Result<(), SchedulerError> {
warn!("kill_running_job() method not implemented for TemporalScheduler");
Err(SchedulerError::SchedulerInternalError(
"kill_running_job not supported by TemporalScheduler".to_string(),
))
}
pub async fn get_running_job_info(
&self,
sched_id: &str,
) -> Result<Option<(String, DateTime<Utc>)>, SchedulerError> {
tracing::info!(
"TemporalScheduler: get_running_job_info() called for job '{}'",
sched_id
);
// First check if the job is marked as currently running
let jobs = self.list_scheduled_jobs().await?;
let job = jobs.iter().find(|j| j.id == sched_id);
if let Some(job) = job {
if job.currently_running {
// For now, we'll return a placeholder session ID and current time
// In a more complete implementation, we would track the actual session ID
// and start time from the Temporal workflow execution
let session_id =
format!("temporal-{}-{}", sched_id, chrono::Utc::now().timestamp());
let start_time = chrono::Utc::now(); // This should be the actual start time
Ok(Some((session_id, start_time)))
} else {
Ok(None)
}
} else {
Err(SchedulerError::JobNotFound(sched_id.to_string()))
}
}
async fn make_request(&self, request: JobRequest) -> Result<JobResponse, SchedulerError> {
let url = format!("{}/jobs", self.service_url);
tracing::info!(
"TemporalScheduler: Making HTTP request to {} with action '{}'",
url,
request.action
);
let response = self
.http_client
.post(&url)
.json(&request)
.send()
.await
.map_err(|e| {
SchedulerError::SchedulerInternalError(format!("HTTP request failed: {}", e))
})?;
if !response.status().is_success() {
return Err(SchedulerError::SchedulerInternalError(format!(
"HTTP request failed with status: {}",
response.status()
)));
}
let job_response: JobResponse = response.json().await.map_err(|e| {
SchedulerError::SchedulerInternalError(format!("Failed to parse response JSON: {}", e))
})?;
Ok(job_response)
}
}
impl Drop for TemporalScheduler {
fn drop(&mut self) {
// Services continue running independently - no cleanup needed
info!("TemporalScheduler dropped - Temporal services continue running independently");
}
}
// Service management utilities
impl TemporalScheduler {
/// Check if Temporal services are running
pub async fn check_services_status(&self) -> (bool, bool) {
let temporal_server_running = self.check_temporal_server().await;
let go_service_running = self.health_check().await.unwrap_or(false);
(temporal_server_running, go_service_running)
}
/// Get service information
pub async fn get_service_info(&self) -> String {
let (temporal_running, go_running) = self.check_services_status().await;
format!(
"Temporal Services Status:\n\
- Temporal Server ({}:7233): {}\n\
- Temporal Web UI: http://localhost:8233\n\
- Go Service ({}:8080): {}\n\
- Service logs: temporal-server.log, temporal-service/temporal-service.log",
if temporal_running {
"localhost"
} else {
"not running"
},
if temporal_running {
"✅ Running"
} else {
"❌ Not Running"
},
if go_running {
"localhost"
} else {
"not running"
},
if go_running {
"✅ Running"
} else {
"❌ Not Running"
}
)
}
/// Stop Temporal services (for manual management)
pub async fn stop_services(&self) -> Result<String, SchedulerError> {
info!("Stopping Temporal services...");
let mut results = Vec::new();
// Stop Go service
let go_result = Command::new("pkill")
.args(["-f", "temporal-service"])
.output();
match go_result {
Ok(output) if output.status.success() => {
results.push("✅ Go service stopped".to_string());
}
Ok(_) => {
results.push("⚠️ Go service was not running or failed to stop".to_string());
}
Err(e) => {
results.push(format!("❌ Failed to stop Go service: {}", e));
}
}
// Stop Temporal server
let temporal_result = Command::new("pkill")
.args(["-f", "temporal server start-dev"])
.output();
match temporal_result {
Ok(output) if output.status.success() => {
results.push("✅ Temporal server stopped".to_string());
}
Ok(_) => {
results.push("⚠️ Temporal server was not running or failed to stop".to_string());
}
Err(e) => {
results.push(format!("❌ Failed to stop Temporal server: {}", e));
}
}
let result_message = results.join("\n");
info!("Service stop results: {}", result_message);
Ok(result_message)
}
}
#[async_trait]
impl SchedulerTrait for TemporalScheduler {
async fn add_scheduled_job(&self, job: ScheduledJob) -> Result<(), SchedulerError> {
self.add_scheduled_job(job).await
}
async fn list_scheduled_jobs(&self) -> Result<Vec<ScheduledJob>, SchedulerError> {
self.list_scheduled_jobs().await
}
async fn remove_scheduled_job(&self, id: &str) -> Result<(), SchedulerError> {
self.remove_scheduled_job(id).await
}
async fn pause_schedule(&self, id: &str) -> Result<(), SchedulerError> {
self.pause_schedule(id).await
}
async fn unpause_schedule(&self, id: &str) -> Result<(), SchedulerError> {
self.unpause_schedule(id).await
}
async fn run_now(&self, id: &str) -> Result<String, SchedulerError> {
self.run_now(id).await
}
async fn sessions(
&self,
sched_id: &str,
limit: usize,
) -> Result<Vec<(String, SessionMetadata)>, SchedulerError> {
self.sessions(sched_id, limit).await
}
async fn update_schedule(
&self,
sched_id: &str,
new_cron: String,
) -> Result<(), SchedulerError> {
self.update_schedule(sched_id, new_cron).await
}
async fn kill_running_job(&self, sched_id: &str) -> Result<(), SchedulerError> {
self.kill_running_job(sched_id).await
}
async fn get_running_job_info(
&self,
sched_id: &str,
) -> Result<Option<(String, DateTime<Utc>)>, SchedulerError> {
self.get_running_job_info(sched_id).await
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_sessions_method_exists_and_compiles() {
// This test verifies that the sessions method exists and compiles correctly
// It doesn't require Temporal services to be running
// Create a mock scheduler instance (this will fail if services aren't running, but that's OK)
let result = TemporalScheduler::new().await;
// Even if scheduler creation fails, we can still test the method signature
match result {
Ok(scheduler) => {
// If services are running, test the actual method
let sessions_result = scheduler.sessions("test-schedule", 5).await;
// The method should return a Result, regardless of success/failure
match sessions_result {
Ok(sessions) => {
// Verify the return type is correct
assert!(sessions.len() <= 5); // Should respect the limit
println!("✅ sessions() method returned {} sessions", sessions.len());
}
Err(e) => {
// Even errors are OK - the method is implemented
println!(
"⚠️ sessions() method returned error (expected if no sessions): {}",
e
);
}
}
}
Err(_) => {
// Services not running - that's fine, we just verified the method compiles
println!("⚠️ Temporal services not running - method signature test passed");
}
}
}
#[test]
fn test_sessions_method_signature() {
// This test verifies the method signature is correct at compile time
// We just need to verify the method exists and can be called
// This will fail to compile if the method doesn't exist or has wrong signature
let _test_fn = |scheduler: &TemporalScheduler, id: &str, limit: usize| {
// This is a compile-time check - we don't actually call it
let _future = scheduler.sessions(id, limit);
};
println!("✅ sessions() method signature is correct");
}
#[test]
fn test_port_check_functionality() {
// Test the port checking functionality
use tokio::runtime::Runtime;
let rt = Runtime::new().unwrap();
rt.block_on(async {
let scheduler = TemporalScheduler {
http_client: reqwest::Client::new(),
service_url: "http://localhost:8080".to_string(),
};
// Test with a port that should be available (high port number)
let high_port_in_use = scheduler.check_port_in_use(65432).await;
// Test with a port that might be in use (port 80)
let low_port_in_use = scheduler.check_port_in_use(80).await;
println!("✅ Port checking functionality works");
println!(" High port (65432) in use: {}", high_port_in_use);
println!(" Low port (80) in use: {}", low_port_in_use);
});
}
#[test]
fn test_find_go_service_binary() {
// Test the Go service binary finding logic
match TemporalScheduler::find_go_service_binary() {
Ok(path) => {
println!("✅ Found Go service binary at: {}", path);
assert!(
std::path::Path::new(&path).exists(),
"Binary should exist at found path"
);
}
Err(e) => {
println!("⚠️ Go service binary not found: {}", e);
// This is expected if the binary isn't built or available
}
}
}
}