diff --git a/crates/switchyard-cli/src/host.rs b/crates/switchyard-cli/src/host.rs index cb4d605..274d59b 100644 --- a/crates/switchyard-cli/src/host.rs +++ b/crates/switchyard-cli/src/host.rs @@ -458,7 +458,10 @@ fn runtime_host_job_event_type_for_runtime_event(event: &RuntimeEvent) -> Option | RuntimeEvent::WorkerSpawned { .. } | RuntimeEvent::WorkerStateChanged { .. } | RuntimeEvent::WorkerRetrying { .. } - | RuntimeEvent::WorkerTerminated { .. } => None, + | RuntimeEvent::WorkerTerminated { .. } + | RuntimeEvent::GoalIterationStarted { .. } + | RuntimeEvent::GoalCompleted { .. } + | RuntimeEvent::GoalPlanUpdated { .. } => None, } } @@ -3400,7 +3403,10 @@ fn apply_runtime_event(job: &mut HostJobState, event: &RuntimeEvent) { | RuntimeEvent::WorkerSpawned { .. } | RuntimeEvent::WorkerStateChanged { .. } | RuntimeEvent::WorkerRetrying { .. } - | RuntimeEvent::WorkerTerminated { .. } => {} + | RuntimeEvent::WorkerTerminated { .. } + | RuntimeEvent::GoalIterationStarted { .. } + | RuntimeEvent::GoalCompleted { .. } + | RuntimeEvent::GoalPlanUpdated { .. } => {} } } diff --git a/crates/switchyard-core/src/goal_runner.rs b/crates/switchyard-core/src/goal_runner.rs new file mode 100644 index 0000000..39a1c72 --- /dev/null +++ b/crates/switchyard-core/src/goal_runner.rs @@ -0,0 +1,784 @@ +//! Switchyard-native **goal mode**. +//! +//! Given a high-level goal, drive the active Core provider turn-after-turn — +//! reusing the full routed-turn machinery (persistent instance via the proxy, +//! delegation, context composition) — until the goal is judged achieved or the +//! user stops. Provider-agnostic, so CLIs whose own `/goal` is only reachable in +//! an interactive TUI (agy) or has no goal at all (kt) still get goal behavior. +//! +//! Design mirrors how the coding CLIs engineer goal mode (all source/binary +//! verified): **no hard iteration/time cap by default** (run until done or the +//! user stops — like Claude Code / Codex / Antigravity), and completion decided +//! two ways, first hit wins: +//! - **self-signal fast-path**: the agent emits [`GOAL_DONE_MARKER`] (cheap; the +//! pattern Codex/Antigravity use — the working agent declares done). +//! - **evaluator** (optional, Claude-Code-style): after each turn a *separate, +//! read-only, tool-less* judge turn decides yes/no + reason; "no" feeds the +//! reason back into the next turn as guidance. +//! +//! The only automatic brake is a stuck-guard (`max_no_progress` consecutive +//! empty/identical turns) so a dumb loop can't spin forever; iteration/time +//! caps are opt-in. + +use std::path::{Path, PathBuf}; +use std::time::Instant; + +use switchyard_provider_api::{ + CancellationToken, EventType, ExecutionPolicy, GOAL_DONE_MARKER, GOAL_PLAN_BEGIN, + GOAL_PLAN_END, InputAttachment, LiveInstanceRegistry, PeerCatalog, Provider, + contains_goal_done, extract_display_text, parse_goal_plan, +}; +use switchyard_session::{GoalPlanItem, Session, Turn, TurnRole, TurnStatus}; +use switchyard_store::{CanonicalStore, JsonlStore}; + +use crate::error::CoreError; +use crate::event_mapper::map_provider_event; +use crate::router::{ + RouterPromptInjection, run_routed_turn_observable_with_display_and_prompt_injection, + run_routed_turn_observable_with_policy_attachments_and_prompt_injection, +}; +use crate::runtime_events::RuntimeEvent; +use crate::turn_runner::run_turn_with_policy; + +/// Consecutive no-progress iterations (empty or identical response) before the +/// loop gives up. A safety brake against a stuck model, NOT a turn cap — a goal +/// that keeps producing new work runs unbounded. +pub const GOAL_STUCK_GUARD: u32 = 6; + +/// Optional bounds for a goal run. Defaults are **unbounded** (run until the +/// goal is achieved or the user stops), matching Claude Code / Codex / agy. +#[derive(Debug, Clone, Copy)] +pub struct GoalLimits { + /// Hard cap on iterations. `None` = unbounded (default). + pub max_iterations: Option, + /// Hard wall-clock cap in seconds. `None` = unbounded (default). + pub max_wall_secs: Option, + /// Stuck-guard: stop after this many consecutive no-progress turns. `0` + /// disables the guard. + pub max_no_progress: u32, +} + +impl Default for GoalLimits { + fn default() -> Self { + Self { + max_iterations: None, + max_wall_secs: None, + max_no_progress: GOAL_STUCK_GUARD, + } + } +} + +/// Terminal state of a goal run. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum GoalStatus { + /// Goal judged achieved (self-signal or evaluator). + Achieved, + /// Stopped by user cancel, an opt-in cap, or the stuck-guard. + Stopped, + /// A turn failed. + Failed, +} + +impl GoalStatus { + pub fn as_str(self) -> &'static str { + match self { + GoalStatus::Achieved => "achieved", + GoalStatus::Stopped => "stopped", + GoalStatus::Failed => "failed", + } + } +} + +/// Result of a completed goal run. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct GoalOutcome { + pub status: GoalStatus, + pub iterations: u32, +} + +/// Build the per-iteration message injected into the routed turn. `feedback` is +/// the evaluator's most-recent "not yet done" reason, threaded in as guidance +/// for the next turn (Claude-Code-style). +pub fn compose_goal_prompt(goal: &str, iteration: u32, feedback: Option<&str>) -> String { + let mut msg = if iteration <= 1 { + format!( + "You are running in autonomous GOAL MODE. Work continuously toward the goal below, \ + taking whatever concrete steps and tool calls are needed each turn.\n\n\ + GOAL: {goal}\n\n\ + When — and ONLY when — the goal is FULLY achieved, output a line containing exactly \ + {GOAL_DONE_MARKER} and nothing else on that line. Never output it prematurely. If the \ + goal is not yet done, make concrete progress this turn." + ) + } else { + format!( + "Continue autonomously toward the GOAL below (iteration {iteration}). Build on the work \ + already done and take the next concrete step. Output the exact line {GOAL_DONE_MARKER} \ + only once the goal is fully achieved.\n\nGOAL: {goal}" + ) + }; + // Ask the agent to maintain a plan/todo list Switchyard can track outside + // the chat (native-goal CLIs keep their own; this is for the self-driven + // loop). Full list re-emitted every turn, most recent block wins. + msg.push_str(&format!( + "\n\nAlso maintain a short PLAN as a markdown checklist inside a block delimited by \ + {GOAL_PLAN_BEGIN} and {GOAL_PLAN_END}. One step per line: \"- [ ] step\" for todo, \ + \"- [x] step\" for done. Re-output the FULL updated plan every turn, marking finished \ + steps done. Keep it to the concrete steps for THIS goal." + )); + if let Some(reason) = feedback.map(str::trim).filter(|r| !r.is_empty()) { + msg.push_str(&format!( + "\n\nThe completion check says the goal is NOT yet met: {reason}\nAddress that now." + )); + } + msg +} + +/// Pure pre-iteration stop check (cancel / opt-in caps). Extracted so the stop +/// policy is unit-testable without running turns. +fn precheck_stop( + iteration: u32, + elapsed_secs: u64, + cancelled: bool, + limits: &GoalLimits, +) -> Option { + if cancelled { + return Some(GoalStatus::Stopped); + } + if limits.max_iterations.is_some_and(|max| iteration >= max) { + return Some(GoalStatus::Stopped); + } + if limits.max_wall_secs.is_some_and(|max| elapsed_secs >= max) { + return Some(GoalStatus::Stopped); + } + None +} + +/// Parse a judge turn's response into (met, reason). `met` is true when the +/// first non-empty line starts with "YES"; the reason is the remaining text. +fn parse_judge_verdict(response: &str) -> (bool, String) { + let mut lines = response.lines().map(str::trim).filter(|l| !l.is_empty()); + let first = lines.next().unwrap_or(""); + let met = first.to_ascii_uppercase().starts_with("YES"); + let reason = lines.collect::>().join(" "); + let reason = if reason.is_empty() { + first.to_string() + } else { + reason + }; + (met, reason) +} + +/// Run a separate, read-only, tool-less evaluator turn against `judge` to decide +/// whether `goal` is achieved given the agent's latest output. Runs on a +/// throwaway store so it never pollutes the real session history (mirrors Claude +/// Code's evaluator, which judges the transcript without adding a turn). Returns +/// `None` if the judge turn errors — caller treats that as "not yet, no reason". +async fn evaluate_goal_met( + judge: &dyn Provider, + judge_store: &mut JsonlStore, + goal: &str, + latest_output: &str, + cwd: &Path, +) -> Option<(bool, String)> { + let prompt = format!( + "You are a strict goal-completion evaluator. Do NOT use any tools or take any action — \ + judge only from the text below.\n\nGOAL:\n{goal}\n\nThe agent's latest output:\n---\n\ + {latest_output}\n---\n\nBased ONLY on the output above, is the GOAL now FULLY achieved? \ + Reply with exactly YES or NO on the first line, then a one-sentence reason on the second." + ); + let mut judge_session = Session::new("goal-judge".to_string()); + let out = run_turn_with_policy( + judge_store, + &mut judge_session, + judge, + prompt, + cwd.to_path_buf(), + ExecutionPolicy::read_only(cwd.to_path_buf()), + ) + .await + .ok()?; + let response = out.response.unwrap_or_default(); + if response.trim().is_empty() { + return None; + } + Some(parse_judge_verdict(&response)) +} + +fn emit(tx: Option<&tokio::sync::mpsc::Sender>, event: RuntimeEvent) { + if let Some(tx) = tx { + let _ = tx.try_send(event); + } +} + +/// Run the goal loop until achieved / stopped / failed. Each iteration is a full +/// routed turn (persisted to the session history), so progress is visible in the +/// chat. Mutates + persists `session.goal_status` / `session.goal_iteration`. +/// +/// `judge`, when `Some`, enables the evaluator: a fresh provider instance used +/// for the read-only completion check (should be distinct from `core_provider` +/// so it doesn't pollute the persistent instance's context). When `None`, only +/// the self-signal marker decides completion. +#[allow(clippy::too_many_arguments)] +pub async fn run_goal_loop( + store: &mut (impl CanonicalStore + ?Sized), + session: &mut Session, + core_provider: &dyn Provider, + peer_catalog: &PeerCatalog, + resolve_peer: &(dyn Fn(&str) -> Option> + Send + Sync), + registry: Option>, + goal: &str, + cwd: PathBuf, + artifact_dir: Option<&Path>, + runtime_tx: Option<&tokio::sync::mpsc::Sender>, + cancel: CancellationToken, + base_policy: ExecutionPolicy, + limits: GoalLimits, + judge: Option<&dyn Provider>, +) -> Result { + let provider_name = session.active_core.clone(); + let started = Instant::now(); + let mut iteration: u32 = 0; + let mut no_progress: u32 = 0; + let mut last_response = String::new(); + let mut feedback: Option = None; + + // Throwaway store for evaluator turns — keeps judge turns out of the real + // session history. Created once and reused (each judge uses a fresh session). + let mut judge_store = judge.map(|_| { + let dir = + std::env::temp_dir().join(format!("switchyard-goal-judge-{}", uuid::Uuid::now_v7())); + let _ = std::fs::create_dir_all(&dir); + (JsonlStore::new(dir.clone()), dir) + }); + + let status = loop { + if let Some(stop) = precheck_stop( + iteration, + started.elapsed().as_secs(), + cancel.is_cancelled(), + &limits, + ) { + break stop; + } + + iteration += 1; + session.goal_iteration = iteration; + session.goal_status = Some("running".to_string()); + let _ = store.save_session(session); + emit( + runtime_tx, + RuntimeEvent::GoalIterationStarted { + session_id: session.session_id, + provider: provider_name.clone(), + iteration, + }, + ); + + let message = compose_goal_prompt(goal, iteration, feedback.as_deref()); + // Show a clean goal line in the chat bubble; the verbose autonomous-mode + // instructions in `message` drive the model but stay out of the UI. No + // iteration number — the user tracks progress via the header timer + plan + // panel, not per-turn counters. + let display = format!("🎯 {goal}"); + let result = run_routed_turn_observable_with_display_and_prompt_injection( + store, + session, + core_provider, + peer_catalog, + resolve_peer, + registry.clone(), + message, + Vec::::new(), + cwd.clone(), + artifact_dir, + runtime_tx, + cancel.clone(), + base_policy.clone(), + Some(display), + RouterPromptInjection::Clean, + ) + .await; + + let response = match result { + Ok(routed) => routed.response.unwrap_or_default(), + Err(_) => break GoalStatus::Failed, + }; + + // Track the agent's plan/todo list outside the chat (persist + emit so + // the UI updates live). Only replaces the plan when a block is present. + if let Some(items) = parse_goal_plan(&response) { + session.goal_plan = items + .into_iter() + .map(|(text, done)| GoalPlanItem { text, done }) + .collect(); + let _ = store.save_session(session); + emit( + runtime_tx, + RuntimeEvent::GoalPlanUpdated { + session_id: session.session_id, + provider: provider_name.clone(), + plan: session.goal_plan.clone(), + }, + ); + } + + // Completion, first hit wins: (1) self-signal marker, (2) evaluator. + if contains_goal_done(&response) { + break GoalStatus::Achieved; + } + if let (Some(judge), Some((judge_store, _))) = (judge, judge_store.as_mut()) + && let Some((met, reason)) = + evaluate_goal_met(judge, judge_store, goal, &response, &cwd).await + { + if met { + break GoalStatus::Achieved; + } + feedback = Some(reason); + } + + // Stuck-guard: give up only after repeated genuinely-empty/identical work. + let progressed = !response.trim().is_empty() && response != last_response; + no_progress = if progressed { 0 } else { no_progress + 1 }; + if limits.max_no_progress > 0 && no_progress >= limits.max_no_progress { + break GoalStatus::Stopped; + } + last_response = response; + }; + + if let Some((_, dir)) = judge_store.take() { + let _ = std::fs::remove_dir_all(&dir); + } + + session.goal_status = Some(status.as_str().to_string()); + let _ = store.save_session(session); + emit( + runtime_tx, + RuntimeEvent::GoalCompleted { + session_id: session.session_id, + provider: provider_name, + status: status.as_str().to_string(), + iterations: iteration, + }, + ); + + Ok(GoalOutcome { + status, + iterations: iteration, + }) +} + +/// Native slash-goal passthrough (Increment 2): send `" "` as a +/// SINGLE turn and let the provider's own goal engine loop inside that turn — +/// used for Claude Code, whose `/goal` runs its evaluator loop to completion +/// within one persistent-instance send. No Switchyard loop, no evaluator: the +/// CLI does everything; we just stream + record the outcome. `base_policy` +/// should be unbounded (timeout 0) since one turn spans the whole native loop. +#[allow(clippy::too_many_arguments)] +pub async fn run_native_slash_goal( + store: &mut (impl CanonicalStore + ?Sized), + session: &mut Session, + core_provider: &dyn Provider, + peer_catalog: &PeerCatalog, + resolve_peer: &(dyn Fn(&str) -> Option> + Send + Sync), + registry: Option>, + slash: &str, + goal: &str, + cwd: PathBuf, + artifact_dir: Option<&Path>, + runtime_tx: Option<&tokio::sync::mpsc::Sender>, + cancel: CancellationToken, + base_policy: ExecutionPolicy, +) -> Result { + let provider_name = session.active_core.clone(); + session.goal_iteration = 1; + session.goal_status = Some("running".to_string()); + let _ = store.save_session(session); + emit( + runtime_tx, + RuntimeEvent::GoalIterationStarted { + session_id: session.session_id, + provider: provider_name.clone(), + iteration: 1, + }, + ); + + let message = format!("{slash} {goal}"); + let result = run_routed_turn_observable_with_policy_attachments_and_prompt_injection( + store, + session, + core_provider, + peer_catalog, + resolve_peer, + registry, + message, + Vec::::new(), + cwd, + artifact_dir, + runtime_tx, + cancel.clone(), + base_policy, + RouterPromptInjection::Clean, + ) + .await; + + // The native /goal loop ends (auto-clears) only when its own evaluator says + // the condition is met, so a clean turn completion = achieved. A cancel = + // stopped; any other error = failed. + let status = match &result { + Ok(_) => GoalStatus::Achieved, + Err(_) if cancel.is_cancelled() => GoalStatus::Stopped, + Err(_) => GoalStatus::Failed, + }; + session.goal_status = Some(status.as_str().to_string()); + let _ = store.save_session(session); + emit( + runtime_tx, + RuntimeEvent::GoalCompleted { + session_id: session.session_id, + provider: provider_name, + status: status.as_str().to_string(), + iterations: 1, + }, + ); + Ok(GoalOutcome { + status, + iterations: 1, + }) +} + +/// If `pe` is the codex-adapter `goal_status` marker event, return its status. +fn goal_status_from_event(pe: &switchyard_provider_api::ProviderEvent) -> Option { + if pe.payload.get("item_type").and_then(|v| v.as_str()) != Some("goal_status") { + return None; + } + pe.payload + .get("status") + .and_then(|v| v.as_str()) + .map(str::to_string) +} + +/// Map a terminal codex goal status to a Switchyard [`GoalStatus`]. `None` for +/// non-terminal (active/paused) statuses. +fn terminal_goal_status(status: &str) -> Option { + match status { + "complete" => Some(GoalStatus::Achieved), + "blocked" => Some(GoalStatus::Failed), + "budget_limited" | "usage_limited" => Some(GoalStatus::Stopped), + _ => None, + } +} + +/// Finalize one persisted codex turn: stamp the accumulated response + mark it +/// completed. `append_turn` is an append-only log (latest row wins), so this +/// re-appends the same `turn_id` with terminal state. +fn finalize_codex_turn( + store: &mut (impl CanonicalStore + ?Sized), + mut turn: Turn, + response: &str, +) -> Result<(), CoreError> { + turn.status = TurnStatus::Completed; + turn.completed_at = Some(chrono::Utc::now()); + if !response.trim().is_empty() { + turn.provider_response = Some(response.trim().to_string()); + } + store.append_turn(&turn)?; + Ok(()) +} + +/// **Native codex goal drive** (Increment: codex `thread/goal/*`). Instead of +/// Switchyard's own loop, hand the objective to codex's engine via +/// [`LiveInstance::drive_native_goal`] and persist each autonomous continuation +/// turn as a canonical Turn (so the canonical-session + UI invariants hold), +/// stopping when codex reports a terminal goal status. +/// +/// Spike scope: minimal persistence — one Turn per codex turn with its events +/// mapped through `map_provider_event`; local diffs are captured at the run +/// level by the caller's file-watcher (not per turn). The plan panel stays +/// empty (codex maintains its own todos in its turn output). +pub async fn run_native_codex_goal( + store: &mut (impl CanonicalStore + ?Sized), + session: &mut Session, + registry: std::sync::Arc, + goal: &str, + runtime_tx: Option<&tokio::sync::mpsc::Sender>, + cancel: CancellationToken, +) -> Result { + let provider = session.active_core.clone(); + session.goal_status = Some("running".to_string()); + session.goal_iteration = 0; + session.goal_plan.clear(); + let _ = store.save_session(session); + + let (instance_id, inst_lock) = registry + .checkout_any_idle(&provider, session.session_id) + .ok_or_else(|| { + CoreError::Runner("no idle codex instance available to drive native goal".to_string()) + })?; + // Drive the native goal; the drain task keeps running off Arc clones after + // we drop the instance lock, so other health checks aren't blocked. + let mut event_rx = { + let mut inst = inst_lock.lock().await; + inst.drive_native_goal(goal, cancel.clone()) + .await + .map_err(|e| CoreError::Runner(format!("codex drive_native_goal: {e}")))? + }; + + let mut iteration: u32 = 0; + let mut status = GoalStatus::Stopped; + let mut cur_turn: Option = None; + let mut cur_response = String::new(); + + while let Some(mut pe) = event_rx.recv().await { + pe.provider = provider.clone(); + + // Goal status transitions (the completion signal). + if let Some(status_str) = goal_status_from_event(&pe) { + if let Some(term) = terminal_goal_status(&status_str) { + status = term; + break; + } + continue; + } + + // New codex turn → finalize the previous, start a fresh canonical Turn. + if cur_turn.as_ref().map(|t| t.turn_id) != Some(pe.turn_id) { + if let Some(turn) = cur_turn.take() { + finalize_codex_turn(store, turn, &cur_response)?; + } + iteration += 1; + session.goal_iteration = iteration; + let _ = store.save_session(session); + let display = if iteration == 1 { + format!("🎯 {goal}") + } else { + "🎯 (goal continuation)".to_string() + }; + let mut turn = Turn::new( + session.session_id, + provider.clone(), + TurnRole::Core, + display, + ); + turn.turn_id = pe.turn_id; + turn.status = TurnStatus::Running; + store.append_turn(&turn)?; + cur_turn = Some(turn); + cur_response.clear(); + emit( + runtime_tx, + RuntimeEvent::GoalIterationStarted { + session_id: session.session_id, + provider: provider.clone(), + iteration, + }, + ); + } + + if pe.event_type == EventType::ItemCompleted + && let Some(text) = extract_display_text(&pe.payload) + && !text.trim().is_empty() + { + if !cur_response.is_empty() { + cur_response.push('\n'); + } + cur_response.push_str(text.trim()); + } + + let turn_completed = pe.event_type == EventType::TurnCompleted; + let canonical = map_provider_event(&pe); + store.append_event(&canonical)?; + + if turn_completed && let Some(turn) = cur_turn.take() { + finalize_codex_turn(store, turn, &cur_response)?; + cur_response.clear(); + } + } + + if let Some(turn) = cur_turn.take() { + finalize_codex_turn(store, turn, &cur_response)?; + } + registry.release(instance_id); + + session.goal_status = Some(status.as_str().to_string()); + let _ = store.save_session(session); + emit( + runtime_tx, + RuntimeEvent::GoalCompleted { + session_id: session.session_id, + provider, + status: status.as_str().to_string(), + iterations: iteration, + }, + ); + Ok(GoalOutcome { + status, + iterations: iteration, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::fake_provider::FakeProvider; + + fn temp_store() -> (JsonlStore, tempfile::TempDir) { + let dir = tempfile::tempdir().unwrap(); + (JsonlStore::new(dir.path().to_path_buf()), dir) + } + + #[test] + fn compose_prompt_includes_goal_marker_and_feedback() { + let first = compose_goal_prompt("ship it", 1, None); + assert!(first.contains("ship it") && first.contains(GOAL_DONE_MARKER)); + let later = compose_goal_prompt("ship it", 3, Some("tests still failing")); + assert!(later.contains("iteration 3")); + assert!(later.contains("tests still failing")); + } + + #[test] + fn precheck_stop_defaults_unbounded_but_honors_cancel_and_optin_caps() { + let unbounded = GoalLimits::default(); + // Unbounded: even a huge iteration count does not stop by itself. + assert_eq!(precheck_stop(10_000, 10_000, false, &unbounded), None); + assert_eq!( + precheck_stop(0, 0, true, &unbounded), + Some(GoalStatus::Stopped) + ); + let capped = GoalLimits { + max_iterations: Some(3), + max_wall_secs: Some(60), + max_no_progress: 6, + }; + assert_eq!( + precheck_stop(3, 0, false, &capped), + Some(GoalStatus::Stopped) + ); + assert_eq!( + precheck_stop(1, 60, false, &capped), + Some(GoalStatus::Stopped) + ); + assert_eq!(precheck_stop(1, 1, false, &capped), None); + } + + #[test] + fn parse_judge_verdict_reads_yes_no_and_reason() { + let (met, reason) = parse_judge_verdict("YES\nall tests pass"); + assert!(met); + assert_eq!(reason, "all tests pass"); + let (met, reason) = parse_judge_verdict("no\nlint still fails"); + assert!(!met); + assert_eq!(reason, "lint still fails"); + let (met, _) = parse_judge_verdict(" yes, done "); + assert!(met); + } + + #[test] + fn goal_status_strings() { + assert_eq!(GoalStatus::Achieved.as_str(), "achieved"); + assert_eq!(GoalStatus::Stopped.as_str(), "stopped"); + assert_eq!(GoalStatus::Failed.as_str(), "failed"); + } + + #[tokio::test] + async fn self_signal_marker_achieves_without_a_judge() { + let (mut store, dir) = temp_store(); + let mut session = Session::new("fake".to_string()); + let provider = FakeProvider::success(format!("all set {GOAL_DONE_MARKER}")); + let catalog = PeerCatalog::default(); + let resolve_peer = |_: &str| None; + + let outcome = run_goal_loop( + &mut store, + &mut session, + &provider, + &catalog, + &resolve_peer, + None, + "make it green", + dir.path().to_path_buf(), + None, + None, + CancellationToken::new(), + ExecutionPolicy::read_only(dir.path().to_path_buf()), + GoalLimits::default(), + None, + ) + .await + .unwrap(); + + assert_eq!(outcome.status, GoalStatus::Achieved); + assert_eq!(outcome.iterations, 1); + } + + #[tokio::test] + async fn evaluator_yes_achieves_even_without_self_signal() { + let (mut store, dir) = temp_store(); + let mut session = Session::new("fake".to_string()); + // Worker never self-signals; the judge says YES → achieved via evaluator. + let worker = FakeProvider::success("did the work"); + let judge = FakeProvider::success("YES\nlooks complete"); + let catalog = PeerCatalog::default(); + let resolve_peer = |_: &str| None; + + let outcome = run_goal_loop( + &mut store, + &mut session, + &worker, + &catalog, + &resolve_peer, + None, + "do the thing", + dir.path().to_path_buf(), + None, + None, + CancellationToken::new(), + ExecutionPolicy::read_only(dir.path().to_path_buf()), + GoalLimits::default(), + Some(&judge), + ) + .await + .unwrap(); + + assert_eq!(outcome.status, GoalStatus::Achieved); + assert_eq!(outcome.iterations, 1); + } + + #[tokio::test] + async fn stuck_guard_stops_a_never_completing_loop() { + let (mut store, dir) = temp_store(); + let mut session = Session::new("fake".to_string()); + // Worker keeps producing the SAME output and never self-signals; the + // judge always says NO. Only the stuck-guard (identical output) stops it. + let worker = FakeProvider::success("still working"); + let judge = FakeProvider::success("NO\nnot done yet"); + let catalog = PeerCatalog::default(); + let resolve_peer = |_: &str| None; + let limits = GoalLimits { + max_iterations: None, + max_wall_secs: None, + max_no_progress: 2, + }; + + let outcome = run_goal_loop( + &mut store, + &mut session, + &worker, + &catalog, + &resolve_peer, + None, + "impossible", + dir.path().to_path_buf(), + None, + None, + CancellationToken::new(), + ExecutionPolicy::read_only(dir.path().to_path_buf()), + limits, + Some(&judge), + ) + .await + .unwrap(); + + assert_eq!(outcome.status, GoalStatus::Stopped); + assert_eq!(session.goal_status.as_deref(), Some("stopped")); + } +} diff --git a/crates/switchyard-core/src/lib.rs b/crates/switchyard-core/src/lib.rs index 514b972..cca23aa 100644 --- a/crates/switchyard-core/src/lib.rs +++ b/crates/switchyard-core/src/lib.rs @@ -1,6 +1,7 @@ pub mod error; pub mod event_mapper; pub mod fake_provider; +pub mod goal_runner; pub mod instance; pub mod ipc; pub mod policy; @@ -18,10 +19,15 @@ pub use proxy::PersistentProviderProxy; // execute_delegate; core depends on orchestrator for the call site). pub use event_mapper::map_provider_event; pub use fake_provider::FakeProvider; +pub use goal_runner::{ + GoalLimits, GoalOutcome, GoalStatus, compose_goal_prompt, run_goal_loop, run_native_codex_goal, + run_native_slash_goal, +}; pub use policy::{execution_policy_from_config, execution_policy_from_config_with_overrides}; pub use registry::{ProviderRegistry, build_peer_catalog, build_peer_catalog_probed}; pub use router::{ RoutedTurnOutput, RouterPromptInjection, run_routed_turn, run_routed_turn_observable, + run_routed_turn_observable_with_display_and_prompt_injection, run_routed_turn_observable_with_policy, run_routed_turn_observable_with_policy_and_attachments, run_routed_turn_observable_with_policy_attachments_and_prompt_injection, run_routed_turn_with_archive, run_routed_turn_with_archive_and_policy, diff --git a/crates/switchyard-core/src/router.rs b/crates/switchyard-core/src/router.rs index 3973ec2..e6b02ed 100644 --- a/crates/switchyard-core/src/router.rs +++ b/crates/switchyard-core/src/router.rs @@ -643,6 +643,49 @@ pub async fn run_routed_turn_observable_with_policy_attachments_and_prompt_injec cancel: switchyard_provider_api::CancellationToken, base_policy: ExecutionPolicy, prompt_injection: RouterPromptInjection, +) -> Result { + run_routed_turn_observable_with_display_and_prompt_injection( + store, + session, + core_provider, + peer_catalog, + resolve_peer, + registry, + user_message, + attachments, + cwd, + artifact_dir, + runtime_tx, + cancel, + base_policy, + None, + prompt_injection, + ) + .await +} + +/// Same as the above, but with an optional `display_message`: when `Some`, that +/// string is what's persisted and shown as the turn's user message, while +/// `user_message` is used only to build the actual model prompt. Goal mode uses +/// this so its verbose autonomous-mode instructions drive the model WITHOUT +/// leaking into the user's chat bubble (the bubble shows a clean goal line). +#[allow(clippy::too_many_arguments)] +pub async fn run_routed_turn_observable_with_display_and_prompt_injection( + store: &mut (impl CanonicalStore + ?Sized), + session: &mut Session, + core_provider: &dyn Provider, + peer_catalog: &PeerCatalog, + resolve_peer: &(dyn Fn(&str) -> Option> + Send + Sync), + registry: Option>, + user_message: String, + attachments: Vec, + cwd: PathBuf, + artifact_dir: Option<&std::path::Path>, + runtime_tx: Option<&tokio::sync::mpsc::Sender>, + cancel: switchyard_provider_api::CancellationToken, + base_policy: ExecutionPolicy, + display_message: Option, + prompt_injection: RouterPromptInjection, ) -> Result { let mut last_output: Option = None; let mut delegated = false; @@ -652,7 +695,10 @@ pub async fn run_routed_turn_observable_with_policy_attachments_and_prompt_injec system_prompt: None, attachments: attachments.clone(), }; - let stored_user_message = user_turn_input.user_message_text(); + // `model_base` is the full text used to build the model prompt; the stored/ + // displayed user message can differ (goal mode shows a clean line). + let model_base = user_turn_input.user_message_text(); + let stored_user_message = display_message.unwrap_or_else(|| model_base.clone()); let (continuation_hint, initial_hint, mut callback_receipts) = if prompt_injection .includes_orchestration() { @@ -691,18 +737,14 @@ pub async fn run_routed_turn_observable_with_policy_attachments_and_prompt_injec // First iteration: clean user text, optionally augmented with // peer catalog + HYARD continuity state for orchestration mode. build_first_iteration_router_message( - &stored_user_message, + &model_base, peer_catalog, initial_hint.as_deref(), prompt_injection, ) } else if let Some(ref ctx) = inject_context { // Finalization: original task + delegate result + explicit instruction - build_finalization_router_message( - &stored_user_message, - ctx, - continuation_hint.as_deref(), - ) + build_finalization_router_message(&model_base, ctx, continuation_hint.as_deref()) } else { break; }; diff --git a/crates/switchyard-core/src/runtime_events.rs b/crates/switchyard-core/src/runtime_events.rs index 2b539b4..84de4c5 100644 --- a/crates/switchyard-core/src/runtime_events.rs +++ b/crates/switchyard-core/src/runtime_events.rs @@ -4,6 +4,7 @@ //! so the TUI (or any observer) can render live state without polling the store. use switchyard_provider_api::{ExecutionTelemetry, HyardJobObservation}; +use switchyard_session::GoalPlanItem; use uuid::Uuid; use serde::{Deserialize, Serialize}; @@ -194,4 +195,27 @@ pub enum RuntimeEvent { /// `permanent_death`, `core_reset`, `session_clear`. reason: String, }, + /// A Switchyard-native goal-mode iteration is starting. `iteration` is + /// 1-indexed. Frontends use this to show live goal progress. + GoalIterationStarted { + session_id: Uuid, + provider: String, + iteration: u32, + }, + /// A goal run finished. `status` is `achieved` | `stopped` | `failed`; + /// `iterations` is how many loop passes ran. + GoalCompleted { + session_id: Uuid, + provider: String, + status: String, + iterations: u32, + }, + /// The goal's tracked plan/todo list changed (Switchyard self-driven loop + /// parsed a fresh plan block from the agent). Carries the full current list + /// so the UI can render it outside the chat without re-fetching the session. + GoalPlanUpdated { + session_id: Uuid, + provider: String, + plan: Vec, + }, } diff --git a/crates/switchyard-gui/frontend/src/App.tsx b/crates/switchyard-gui/frontend/src/App.tsx index 7b9250a..27b6846 100644 --- a/crates/switchyard-gui/frontend/src/App.tsx +++ b/crates/switchyard-gui/frontend/src/App.tsx @@ -1872,6 +1872,9 @@ function App() { const [telemetryLogs, setTelemetryLogs] = useState([]); const [sessionEvents, setSessionEvents] = useState([]); const sessionEventsCursorRef = useRef>({}); + // Turns whose events we've already lazily backfilled (keyed `sessionId:turnId`), + // so a historical row that scrolls into view fetches its events at most once. + const ensuredTurnEventsRef = useRef>(new Set()); const [realtimeTerminalLines, setRealtimeTerminalLines] = useState>({}); // Keep the unbounded-ish text accumulator out of React state. Terminal // chunks can arrive dozens of times per second; storing the buffer in state @@ -2932,6 +2935,45 @@ function App() { return eventList; }; + // Lazily backfill a single historical turn's events on demand. The session + // loader caps how many events it pulls up front, so in a long session most + // turns' tool/diff events fall outside that window and their cards can't + // render. When such a row scrolls into view, fetch just that turn's events + // (cheap point lookup on the events(turn_id) index) and merge them in. + const ensureTurnEventsLoaded = useCallback((turnId: string) => { + if (!turnId) return; + const sessionId = selectedSessionRef.current?.session_id; + if (!sessionId) return; + const key = `${sessionId}:${turnId}`; + if (ensuredTurnEventsRef.current.has(key)) return; + ensuredTurnEventsRef.current.add(key); + void (async () => { + try { + const events = await invoke('get_turn_events', { turnId }); + if (!events?.length) return; + // Session may have changed while the fetch was in flight. + if (selectedSessionRef.current?.session_id !== sessionId) return; + setSessionEvents((prev) => { + const seen = new Set(prev.map((e) => e?.event_id).filter(Boolean)); + const fresh = events.filter((e) => e?.event_id && !seen.has(e.event_id)); + if (fresh.length === 0) return prev; + return [...prev, ...fresh]; + }); + } catch (err) { + // Allow a retry on a transient failure. + ensuredTurnEventsRef.current.delete(key); + console.warn('failed to lazy-load turn events', turnId, err); + } + })(); + }, []); + + // Reset the lazy-backfill dedup set when the active session changes, so + // re-opening a session re-backfills any turn whose events aren't in the + // restored snapshot (a redundant fetch is deduped server-side cheaply). + useEffect(() => { + ensuredTurnEventsRef.current.clear(); + }, [selectedSession?.session_id]); + const fetchRuntimeSnapshot = async (sessionId: string, mode: 'full' | 'incremental' = 'incremental') => { const afterEventId = mode === 'incremental' ? (runtimeSnapshotCursorRef.current[sessionId] ?? 0) : 0; const snapshot = await invoke('get_runtime_snapshot', { @@ -3914,6 +3956,44 @@ function App() { } break; } + + case 'GoalIterationStarted': { + const sessionId = normalizedString(data.session_id); + const iteration = typeof data.iteration === 'number' ? data.iteration : undefined; + if (sessionId) { + applyGoalPatch(sessionId, { + goal_status: 'running', + ...(iteration !== undefined ? { goal_iteration: iteration } : {}), + }); + if (iteration !== undefined) { + enqueueLog(now, 'sys', `Goal iteration ${iteration} started (${data.provider})`); + } + } + break; + } + + case 'GoalCompleted': { + const sessionId = normalizedString(data.session_id); + const status = normalizedString(data.status) ?? 'stopped'; + if (sessionId) { + applyGoalPatch(sessionId, { goal_status: status }); + enqueueLog( + now, + 'sys', + `Goal ${status} after ${data.iterations ?? '?'} iteration(s) (${data.provider})`, + ); + } + break; + } + + case 'GoalPlanUpdated': { + const sessionId = normalizedString(data.session_id); + const plan = Array.isArray(data.plan) + ? data.plan.map((p: any) => ({ text: String(p?.text ?? ''), done: Boolean(p?.done) })) + : []; + if (sessionId) applyGoalPatch(sessionId, { goal_plan: plan }); + break; + } } }; @@ -4546,6 +4626,20 @@ function App() { if (res.length > 0 && !selectedSessionRef.current) { selectSession(res[0]); } + // Durable goal mode: resume any goal left "running" when the app last + // closed. Idempotent — the backend skips goals already driven by a live + // loop, so calling this on every workspace load is safe. + void invoke('resume_active_goals') + .then((ids) => { + if (ids.length > 0) { + addLog( + new Date().toLocaleTimeString(), + 'sys', + `Resumed ${ids.length} active goal(s) from a previous session.`, + ); + } + }) + .catch((e) => console.error('resume_active_goals failed', e)); } catch (e) { console.error(e); } @@ -5177,6 +5271,21 @@ function App() { }, openCanvasFile: openFileInCanvas, appendSystemNote, + startGoal: async (goal: string) => { + const target = selectedSessionRef.current; + if (!target) throw new Error('select a session first'); + await handleStartGoal(target.session_id, goal); + }, + stopGoal: async () => { + const target = selectedSessionRef.current; + if (target) await handleStopGoal(target.session_id); + }, + goalStatus: () => { + const s = selectedSessionRef.current; + if (!s?.goal) return null; + const iter = s.goal_iteration ? ` (iter ${s.goal_iteration})` : ''; + return `${s.goal_status ?? 'idle'}${iter}: ${s.goal}`; + }, }); const handleSend = async ( @@ -5625,6 +5734,48 @@ function App() { } }; + // Patch goal fields on a session in both the list and the live selection. + const applyGoalPatch = (sessionId: string, patch: Partial) => { + setSessions((prev) => + prev.map((s) => (s.session_id === sessionId ? { ...s, ...patch } : s)), + ); + if (selectedSessionRef.current?.session_id === sessionId) { + const updated = { ...selectedSessionRef.current, ...patch } as Session; + selectedSessionRef.current = updated; + setSelectedSession(updated); + } + }; + + // Activate Switchyard goal mode. `start_goal` blocks for the whole run (like + // `run_turn` blocks for one turn); each iteration streams as a normal turn via + // runtime events, so the chat shows live progress. We optimistically flag the + // session "running" and set the terminal status from the ":" + // result when it resolves. + const handleStartGoal = async (sessionId: string, goal: string) => { + applyGoalPatch(sessionId, { goal, goal_status: 'running', goal_iteration: 0 }); + addLog(new Date().toLocaleTimeString(), 'sys', `Goal activated: ${goal}`); + try { + const resp = await invoke('start_goal', { sessionId, goal }); + const status = (resp || '').split(':')[0] || 'stopped'; + applyGoalPatch(sessionId, { goal_status: status }); + addLog(new Date().toLocaleTimeString(), 'sys', `Goal ${resp}`); + } catch (e) { + applyGoalPatch(sessionId, { goal_status: 'failed' }); + addLog(new Date().toLocaleTimeString(), 'sys', `Goal failed: ${e}`); + console.error('goal run failed', e); + } + }; + + // Cooperatively stop the active goal. The blocking `start_goal` call then + // resolves with a "stopped:*" status and patches the session. + const handleStopGoal = async (sessionId: string) => { + try { + await invoke('stop_goal', { sessionId }); + } catch (e) { + console.error('stop goal failed', e); + } + }; + const handleUpdateSessionRuntime = async ( sessionId: string, activeCore: string, @@ -5867,6 +6018,7 @@ function App() { renderMessageBody={renderMessageBody} renderTurnEvents={renderTurnEventsWithActions} renderTurnActivitySummary={renderTurnActivitySummaryWithActions} + onEnsureTurnEvents={ensureTurnEventsLoaded} queuedMessages={messageQueue} onClearQueue={handleClearQueue} sandboxMode={config?.sandbox?.mode ?? DEFAULT_SANDBOX_MODE} @@ -6037,6 +6189,8 @@ function App() { onUpdateSessionRuntime={handleUpdateSessionRuntime} onUpdateSessionSummary={handleUpdateSessionSummary} onUpdateSessionChecklist={handleUpdateSessionChecklist} + onStartGoal={handleStartGoal} + onStopGoal={handleStopGoal} /> )} diff --git a/crates/switchyard-gui/frontend/src/components/ChatArea.tsx b/crates/switchyard-gui/frontend/src/components/ChatArea.tsx index 61ad71f..2e0fd49 100644 --- a/crates/switchyard-gui/frontend/src/components/ChatArea.tsx +++ b/crates/switchyard-gui/frontend/src/components/ChatArea.tsx @@ -15,6 +15,16 @@ import { stripAttachmentReferences, } from '../utils/attachments'; +/// Format an elapsed-seconds count as `m:ss` (or `h:mm:ss` past an hour) for the +/// active-goal header timer. +function formatGoalElapsed(secs: number): string { + const h = Math.floor(secs / 3600); + const m = Math.floor((secs % 3600) / 60); + const s = secs % 60; + const pad = (n: number) => String(n).padStart(2, '0'); + return h > 0 ? `${h}:${pad(m)}:${pad(s)}` : `${m}:${pad(s)}`; +} + interface ChatAreaProps { selectedSession: Session | null; isGenerating: boolean; @@ -56,6 +66,9 @@ interface ChatAreaProps { hyardJobs?: Record, options?: RenderTurnEventsOptions, ) => React.ReactNode; + /// Lazily backfill a historical turn's events when its row scrolls into view + /// (its events may be outside the session loader's initial window). + onEnsureTurnEvents?: (turnId: string) => void; queuedMessages: SendPayload[]; onClearQueue: () => void; sandboxMode: SandboxMode; @@ -1114,6 +1127,7 @@ interface HistoricalTurnContentProps { storedAttachments?: InputAttachment[]; hasAssistantActivity: boolean; turnEvents: any[]; + onEnsureTurnEvents?: (turnId: string) => void; relatedTurns: Turn[]; realtimeLines?: string[]; hyardJob?: any; @@ -1139,6 +1153,7 @@ const HistoricalTurnContentBase: React.FC = ({ storedAttachments, hasAssistantActivity, turnEvents, + onEnsureTurnEvents, relatedTurns, realtimeLines, hyardJob, @@ -1157,6 +1172,17 @@ const HistoricalTurnContentBase: React.FC = ({ const assistantContent = t.provider_response || t.error_message; const scopedHyardJobs = t.turn_id && hyardJob ? { [t.turn_id]: hyardJob } : undefined; + // A completed assistant turn that produced activity but has no loaded events + // is almost certainly one whose events fell outside the session loader's + // initial window — lazily backfill them so its tool/diff cards render. The + // parent dedups per turn, so this fires at most once per row. + useEffect(() => { + if (!t.turn_id || !onEnsureTurnEvents) return; + if (turnEvents.length > 0) return; + if (!hasAssistantActivity && !t.provider_response) return; + onEnsureTurnEvents(t.turn_id); + }, [t.turn_id, turnEvents.length, hasAssistantActivity, t.provider_response, onEnsureTurnEvents]); + if (t.origin === 'user') { if (isSystemFeedback) { const parsedResults = cachedParseSystemFeedbackResults(t.user_message); @@ -1588,6 +1614,7 @@ export const ChatArea: React.FC = ({ renderMessageBody, renderTurnEvents, renderTurnActivitySummary, + onEnsureTurnEvents, queuedMessages, onClearQueue, sandboxMode, @@ -1651,6 +1678,29 @@ export const ChatArea: React.FC = ({ const currentSessionId = selectedSession?.session_id ?? null; const sessionCacheKey = virtualSessionKey(currentSessionId); + // Live elapsed timer for an active goal run (the user wants a clock, not an + // iteration counter). Baseline is tracked in the frontend and resets on + // session switch / new run — matching Claude Code's "timer resets on resume". + const goalRunning = selectedSession?.goal_status === 'running'; + const [goalElapsedSecs, setGoalElapsedSecs] = useState(0); + const goalStartRef = useRef(null); + useEffect(() => { + if (!goalRunning) { + goalStartRef.current = null; + setGoalElapsedSecs(0); + return; + } + if (goalStartRef.current == null) goalStartRef.current = Date.now(); + const tick = () => { + if (goalStartRef.current != null) { + setGoalElapsedSecs(Math.floor((Date.now() - goalStartRef.current) / 1000)); + } + }; + tick(); + const id = window.setInterval(tick, 1000); + return () => window.clearInterval(id); + }, [goalRunning, currentSessionId]); + const { lastUserTurnId, delegateTurnsByParent, @@ -2437,6 +2487,21 @@ export const ChatArea: React.FC = ({
+ {goalRunning && ( +
+ 🎯 Goal · {formatGoalElapsed(goalElapsedSecs)} +
+ )} {isGenerating && (
@@ -2577,6 +2642,7 @@ export const ChatArea: React.FC = ({ storedAttachments={turnAttachments[t.turn_id]} hasAssistantActivity={hasAssistantActivity} turnEvents={turnEvents} + onEnsureTurnEvents={onEnsureTurnEvents} relatedTurns={relatedTurns} realtimeLines={realtimeTerminalLines[t.turn_id]} hyardJob={t.turn_id ? hyardJobsByTurnId.get(t.turn_id) : undefined} diff --git a/crates/switchyard-gui/frontend/src/components/ControlCenter.tsx b/crates/switchyard-gui/frontend/src/components/ControlCenter.tsx index 9444f56..3fe7d9f 100644 --- a/crates/switchyard-gui/frontend/src/components/ControlCenter.tsx +++ b/crates/switchyard-gui/frontend/src/components/ControlCenter.tsx @@ -44,6 +44,10 @@ interface ControlCenterProps { ) => Promise; onUpdateSessionSummary: (sessionId: string, summary: string | null) => Promise; onUpdateSessionChecklist: (sessionId: string, checklistJson: string) => Promise; + /// Activate Switchyard goal mode: drive the Core toward `goal` until done. + onStartGoal: (sessionId: string, goal: string) => Promise; + /// Cooperatively stop the active goal run. + onStopGoal: (sessionId: string) => Promise; } type TabType = 'topology' | 'agents' | 'checklist' | 'summary' | 'telemetry'; @@ -107,6 +111,8 @@ export const ControlCenter: React.FC = ({ onUpdateSessionRuntime, onUpdateSessionSummary, onUpdateSessionChecklist, + onStartGoal, + onStopGoal, }) => { const [activeTab, setActiveTab] = useState('agents'); const [logSearch, setLogSearch] = useState(''); @@ -119,6 +125,8 @@ export const ControlCenter: React.FC = ({ const [sessionModel, setSessionModel] = useState(''); const [sessionThinkingLevel, setSessionThinkingLevel] = useState(''); const [runtimeSaving, setRuntimeSaving] = useState(false); + const [goalText, setGoalText] = useState(''); + const [goalBusy, setGoalBusy] = useState(false); // Sync summaryText when selectedSession changes useEffect(() => { @@ -129,11 +137,13 @@ export const ControlCenter: React.FC = ({ setSessionCore(selectedSession?.active_core || ''); setSessionModel(selectedSession?.model || ''); setSessionThinkingLevel(selectedSession?.thinking_level || ''); + setGoalText(selectedSession?.goal || ''); }, [ selectedSession?.session_id, selectedSession?.active_core, selectedSession?.model, selectedSession?.thinking_level, + selectedSession?.goal, ]); const checklistItems: ChecklistItem[] = React.useMemo(() => { @@ -1202,6 +1212,117 @@ export const ControlCenter: React.FC = ({ {selectedRuntimeMapping.thinkingHint}
+
+
Goal Mode (Switchyard)
+
+ 给一个目标,Switchyard 驱动 Core 连续多轮运行直到达成或到上限。对所有 provider 通用。 +
+