From c0baaa171bb3d1b1b716f6a908a2a40bebcaab03 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 16 Oct 2025 10:27:49 +0100 Subject: [PATCH] R2 --- codex-rs/codex-infty/src/lib.rs | 2 + codex-rs/codex-infty/src/orchestrator.rs | 620 ++++-------------- codex-rs/codex-infty/src/roles/director.rs | 98 +++ codex-rs/codex-infty/src/roles/mod.rs | 49 ++ codex-rs/codex-infty/src/roles/solver.rs | 216 ++++++ codex-rs/codex-infty/src/roles/verifier.rs | 132 ++++ .../codex-infty/src/roles/verifier_pool.rs | 114 ++++ codex-rs/codex-infty/src/signals.rs | 12 +- codex-rs/codex-infty/src/utils.rs | 56 ++ codex-rs/codex-infty/tests/orchestrator.rs | 91 ++- .../codex-infty/tests/verifier_replacement.rs | 155 +++++ 11 files changed, 1008 insertions(+), 537 deletions(-) create mode 100644 codex-rs/codex-infty/src/roles/director.rs create mode 100644 codex-rs/codex-infty/src/roles/mod.rs create mode 100644 codex-rs/codex-infty/src/roles/solver.rs create mode 100644 codex-rs/codex-infty/src/roles/verifier.rs create mode 100644 codex-rs/codex-infty/src/roles/verifier_pool.rs create mode 100644 codex-rs/codex-infty/src/utils.rs create mode 100644 codex-rs/codex-infty/tests/verifier_replacement.rs diff --git a/codex-rs/codex-infty/src/lib.rs b/codex-rs/codex-infty/src/lib.rs index f66ae69649..886e50ec1b 100644 --- a/codex-rs/codex-infty/src/lib.rs +++ b/codex-rs/codex-infty/src/lib.rs @@ -3,6 +3,8 @@ mod orchestrator; mod progress; mod prompts; +pub(crate) mod utils; +mod roles; mod run_store; mod session; mod signals; diff --git a/codex-rs/codex-infty/src/orchestrator.rs b/codex-rs/codex-infty/src/orchestrator.rs index 46e5d0be3e..6480ccf95a 100644 --- a/codex-rs/codex-infty/src/orchestrator.rs +++ b/codex-rs/codex-infty/src/orchestrator.rs @@ -1,9 +1,7 @@ -use std::any::type_name; use std::fs; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; -use std::time::Duration; use anyhow::Context; use anyhow::Result; @@ -14,33 +12,29 @@ use codex_core::CodexConversation; use codex_core::ConversationManager; use codex_core::config::Config; use codex_core::config::ConfigOverrides; -use codex_core::cross_session::AssistantMessage; use codex_core::cross_session::CrossSessionHub; -use codex_core::cross_session::SessionEventStream; use codex_core::protocol::EventMsg; use codex_core::protocol::Op; use codex_protocol::ConversationId; -use serde::Deserialize; -use serde::Serialize; -use serde::de::DeserializeOwned; -use serde::de::Error as _; -use serde_json::Value; -use serde_json::json; use tokio::signal; use tokio_stream::StreamExt; use tracing::warn; use crate::progress::ProgressReporter; use crate::prompts; +use crate::roles::Role; +use crate::roles::director::DirectionRequestPayload; +use crate::roles::director::DirectorRole; +use crate::roles::solver::SolverRequest; +use crate::roles::solver::SolverRole; +use crate::roles::solver::SolverSignal; +use crate::roles::solver::parse_solver_signal; +use crate::roles::verifier::VerificationRequestPayload; +use crate::roles::verifier_pool::VerifierPool; use crate::run_store::RoleMetadata; use crate::run_store::RunStore; use crate::session; use crate::signals::AggregatedVerifierVerdict; -use crate::signals::DirectiveResponse; -use crate::signals::VerifierDecision; -use crate::signals::VerifierReport; -use crate::signals::VerifierVerdict; -use crate::types::FINALIZATION_PROMPT; use crate::types::RoleConfig; use crate::types::RoleSession; use crate::types::RunExecutionOptions; @@ -48,47 +42,6 @@ use crate::types::RunOutcome; use crate::types::RunParams; use crate::types::RunSessions; -#[derive(Debug, Deserialize)] -#[serde(tag = "type", rename_all = "snake_case")] -enum SolverSignal { - DirectionRequest { - #[serde(default)] - prompt: Option, - }, - VerificationRequest { - #[serde(default)] - claim_path: Option, - #[serde(default)] - notes: Option, - }, - FinalDelivery { - #[serde(default)] - deliverable_path: Option, - #[serde(default)] - summary: Option, - }, -} - -#[derive(Serialize)] -struct DirectionRequestPayload<'a> { - #[serde(rename = "type")] - kind: &'static str, - prompt: &'a str, - #[serde(skip_serializing_if = "Option::is_none")] - objective: Option<&'a str>, -} - -#[derive(Serialize)] -struct VerificationRequestPayload<'a> { - #[serde(rename = "type")] - kind: &'static str, - claim_path: &'a str, - #[serde(skip_serializing_if = "Option::is_none")] - notes: Option<&'a str>, - #[serde(skip_serializing_if = "Option::is_none")] - objective: Option<&'a str>, -} - struct SessionCleanup { conversation_id: ConversationId, conversation: Arc, @@ -227,18 +180,34 @@ impl InftyOrchestrator { sessions: &mut RunSessions, options: &RunExecutionOptions, ) -> Result { - let mut solver_events = self.stream_events(sessions.solver.conversation_id)?; + let solver_role = SolverRole::new( + Arc::clone(&self.hub), + sessions.run_id.clone(), + sessions.solver.role.clone(), + sessions.solver.conversation_id, + self.progress.clone(), + ); + let director_role = DirectorRole::new( + Arc::clone(&self.hub), + sessions.run_id.clone(), + sessions.director.role.clone(), + options.director_timeout, + self.progress.clone(), + ); + let mut verifier_pool = VerifierPool::from_sessions( + Arc::clone(&self.hub), + sessions, + options.verifier_timeout, + self.progress.clone(), + ); + + let mut solver_events = solver_role.stream_events()?; let mut waiting_for_signal = false; let mut pending_solver_turn_completion = false; if let Some(objective) = &options.objective { - session::post_turn( - self.hub.as_ref(), - &sessions.run_id, - &sessions.solver.role, - objective.as_str(), - Some(solver_signal_schema()), - ) - .await?; + solver_role + .post(objective.as_str(), Some(SolverRole::solver_signal_schema())) + .await?; sessions.store.touch()?; waiting_for_signal = true; if let Some(progress) = self.progress.as_ref() { @@ -268,47 +237,29 @@ impl InftyOrchestrator { waiting_for_signal = false; match signal { SolverSignal::DirectionRequest { prompt } => { - let prompt = prompt - .and_then(|p| { - let trimmed = p.trim(); - if trimmed.is_empty() { - None - } else { - Some(trimmed.to_string()) - } - }) - .ok_or_else(|| { - anyhow!( - "solver direction_request missing prompt text" - ) - })?; + let prompt = crate::utils::required_trimmed( + prompt, + "solver direction_request missing prompt text", + )?; if let Some(progress) = self.progress.as_ref() { progress.direction_request(&prompt); } - self.handle_direction_request( - sessions, - &prompt, - options, - ) - .await?; + self + .handle_direction_request( + &prompt, + options, + &director_role, + &solver_role, + ) + .await?; sessions.store.touch()?; pending_solver_turn_completion = true; } SolverSignal::VerificationRequest { claim_path, notes } => { - let claim_path = claim_path - .and_then(|p| { - let trimmed = p.trim(); - if trimmed.is_empty() { - None - } else { - Some(trimmed.to_string()) - } - }) - .ok_or_else(|| { - anyhow!( - "solver verification_request missing claim_path" - ) - })?; + let claim_path = crate::utils::required_trimmed( + claim_path, + "solver verification_request missing claim_path", + )?; if let Some(progress) = self.progress.as_ref() { progress.verification_request( &claim_path, @@ -318,9 +269,11 @@ impl InftyOrchestrator { let verified = self .handle_verification_request( sessions, + &mut verifier_pool, &claim_path, notes.as_deref(), options, + &solver_role, ) .await?; sessions.store.touch()?; @@ -332,35 +285,18 @@ impl InftyOrchestrator { deliverable_path, summary, } => { - let deliverable_path = deliverable_path - .and_then(|p| { - let trimmed = p.trim(); - if trimmed.is_empty() { - None - } else { - Some(trimmed.to_string()) - } - }) - .ok_or_else(|| { - anyhow!( - "solver final_delivery missing deliverable_path" - ) - })?; + let deliverable_path = crate::utils::required_trimmed( + deliverable_path, + "solver final_delivery missing deliverable_path", + )?; if deliverable_path.is_empty() { bail!("solver final_delivery provided empty path"); } - let resolved = resolve_deliverable_path( + let resolved = crate::utils::resolve_deliverable_path( sessions.store.path(), &deliverable_path, )?; - let summary_clean = summary.and_then(|s| { - let trimmed = s.trim(); - if trimmed.is_empty() { - None - } else { - Some(trimmed.to_string()) - } - }); + let summary_clean = crate::utils::trim_to_non_empty(summary); let summary_ref = summary_clean.as_deref(); if let Some(progress) = self.progress.as_ref() { progress.final_delivery(&resolved, summary_ref); @@ -368,9 +304,11 @@ impl InftyOrchestrator { let verified = self .run_final_verification( sessions, + &mut verifier_pool, &resolved, summary_ref, options, + &solver_role, ) .await?; if !verified { @@ -391,8 +329,7 @@ impl InftyOrchestrator { EventMsg::TaskComplete(..) => { if waiting_for_signal { // The solver completed its turn without issuing a signal; ask for one now. - self.request_solver_signal(&sessions.run_id, &sessions.solver.role) - .await?; + solver_role.request_finalization_signal().await?; } else if pending_solver_turn_completion { // We handled a signal earlier in the loop; this completion corresponds to it. pending_solver_turn_completion = false; @@ -405,8 +342,7 @@ impl InftyOrchestrator { if let Some(progress) = self.progress.as_ref() { progress.run_interrupted(); } - let cleanup = collect_session_cleanup(sessions); - self.shutdown_sessions(cleanup).await; + // Cleanup is handled by the caller (drive_run) to avoid double-shutdown bail!("run interrupted by Ctrl+C"); } } @@ -420,80 +356,62 @@ impl InftyOrchestrator { async fn handle_direction_request( &self, - sessions: &RunSessions, prompt: &str, options: &RunExecutionOptions, + director_role: &DirectorRole, + solver_role: &SolverRole, ) -> Result<()> { - let request = DirectionRequestPayload { - kind: "direction_request", - prompt, - objective: options.objective.as_deref(), - }; - let request_text = serde_json::to_string_pretty(&request)?; - let handle = session::post_turn( - self.hub.as_ref(), - &sessions.run_id, - &sessions.director.role, - request_text, - Some(directive_response_schema()), - ) - .await?; - let progress = self - .progress - .as_deref() - .map(|reporter| (reporter, "director")); - let directive = session::await_first_idle( - self.hub.as_ref(), - &handle, - options.director_timeout, - progress, - ) - .await?; - let directive_payload: DirectiveResponse = parse_json_struct(&directive.message.message) + let request = DirectionRequestPayload::new(prompt, options.objective.as_deref()); + let directive_payload = director_role + .call(&request) + .await .context("director response was not valid directive JSON")?; if let Some(progress) = self.progress.as_ref() { progress.director_response(&directive_payload); } - let directive_text = serde_json::to_string_pretty(&directive_payload)?; - session::post_turn( - self.hub.as_ref(), - &sessions.run_id, - &sessions.solver.role, - directive_text, - Some(solver_signal_schema()), - ) - .await?; + let req = SolverRequest::from(directive_payload); + solver_role.call(&req).await?; Ok(()) } async fn handle_verification_request( &self, sessions: &mut RunSessions, + verifier_pool: &mut VerifierPool, claim_path: &str, notes: Option<&str>, options: &RunExecutionOptions, + solver_role: &SolverRole, ) -> Result { - let objective = options - .objective - .as_deref() - .map(str::trim) - .filter(|objective| !objective.is_empty()); + let objective = crate::utils::objective_as_str(options); - let summary = self - .collect_verification_summary(sessions, claim_path, notes, objective, options) - .await?; + let request = VerificationRequestPayload::new(claim_path, notes, objective); + if verifier_pool.is_empty() { + return Ok(true); + } + let round = verifier_pool.collect_round(&request).await?; + for role in &round.passing_roles { + if let Err(err) = self.replace_verifier_session(sessions, role).await { + warn!(role = %role, ?err, "failed to replace verifier session; keeping existing"); + } else { + verifier_pool.replace_role(role); + } + } + let summary = round.summary; self.emit_verification_summary(&summary); - self.post_verification_summary_to_solver(sessions, &summary) - .await?; + let req = SolverRequest::from(&summary); + solver_role.call(&req).await?; Ok(summary.overall.is_pass()) } async fn run_final_verification( &self, sessions: &mut RunSessions, + verifier_pool: &mut VerifierPool, deliverable_path: &Path, summary: Option<&str>, options: &RunExecutionOptions, + solver_role: &SolverRole, ) -> Result { let relative = deliverable_path .strip_prefix(sessions.store.path()) @@ -501,112 +419,25 @@ impl InftyOrchestrator { .and_then(|p| p.to_str().map(|s| s.to_string())); let claim_path = relative.unwrap_or_else(|| deliverable_path.display().to_string()); - let objective = options - .objective - .as_deref() - .map(str::trim) - .filter(|objective| !objective.is_empty()); + let objective = crate::utils::objective_as_str(options); - let summary_result = self - .collect_verification_summary( - sessions, - claim_path.as_str(), - summary, - objective, - options, - ) - .await?; - self.emit_verification_summary(&summary_result); - self.post_verification_summary_to_solver(sessions, &summary_result) - .await?; - Ok(summary_result.overall.is_pass()) - } - - async fn request_solver_signal(&self, run_id: &str, solver_role: &str) -> Result<()> { - let handle = session::post_turn( - self.hub.as_ref(), - run_id, - solver_role, - FINALIZATION_PROMPT, - Some(final_delivery_schema()), - ) - .await?; - let _ = session::await_first_idle(self.hub.as_ref(), &handle, Duration::from_secs(5), None) - .await?; - Ok(()) - } - - async fn collect_verification_summary( - &self, - sessions: &mut RunSessions, - claim_path: &str, - notes: Option<&str>, - objective: Option<&str>, - options: &RunExecutionOptions, - ) -> Result { - if sessions.verifiers.is_empty() { - return Ok(aggregate_verdicts(Vec::new())); + let request = VerificationRequestPayload::new(claim_path.as_str(), summary, objective); + if verifier_pool.is_empty() { + return Ok(true); } - - let request = VerificationRequestPayload { - kind: "verification_request", - claim_path, - notes, - objective, - }; - let request_text = serde_json::to_string_pretty(&request)?; - let mut results: Vec<(String, VerifierVerdict)> = - Vec::with_capacity(sessions.verifiers.len()); - for verifier in &sessions.verifiers { - let handle = session::post_turn( - self.hub.as_ref(), - &sessions.run_id, - &verifier.role, - request_text.clone(), - Some(verifier_verdict_schema()), - ) - .await?; - let progress = self - .progress - .as_deref() - .map(|reporter| (reporter, verifier.role.as_str())); - let response = session::await_first_idle( - self.hub.as_ref(), - &handle, - options.verifier_timeout, - progress, - ) - .await?; - let verdict: VerifierVerdict = parse_json_struct(&response.message.message) - .with_context(|| { - format!("verifier {} returned invalid verdict JSON", verifier.role) - })?; - if let Some(progress) = self.progress.as_ref() { - progress.verifier_verdict(&verifier.role, &verdict); - } - results.push((verifier.role.clone(), verdict)); - } - - // Replace any verifier that passed with a fresh session; keep failures. - // Build a set of roles to replace to avoid borrowing issues while mutating. - let to_replace: Vec = results - .iter() - .filter_map(|(role, verdict)| { - if verdict.verdict.is_pass() { - Some(role.clone()) - } else { - None - } - }) - .collect(); - for role in to_replace { - if let Err(err) = self.replace_verifier_session(sessions, &role).await { + let round = verifier_pool.collect_round(&request).await?; + for role in &round.passing_roles { + if let Err(err) = self.replace_verifier_session(sessions, role).await { warn!(role = %role, ?err, "failed to replace verifier session; keeping existing"); + } else { + verifier_pool.replace_role(role); } } - - // Aggregate directly from the collected results - Ok(aggregate_verdicts(results)) + let summary_result = round.summary; + self.emit_verification_summary(&summary_result); + let req = SolverRequest::from(&summary_result); + solver_role.call(&req).await?; + Ok(summary_result.overall.is_pass()) } async fn replace_verifier_session(&self, sessions: &mut RunSessions, role: &str) -> Result<()> { @@ -656,23 +487,6 @@ impl InftyOrchestrator { } } - async fn post_verification_summary_to_solver( - &self, - sessions: &RunSessions, - summary: &AggregatedVerifierVerdict, - ) -> Result<()> { - let summary_text = serde_json::to_string_pretty(summary)?; - session::post_turn( - self.hub.as_ref(), - &sessions.run_id, - &sessions.solver.role, - summary_text, - Some(solver_signal_schema()), - ) - .await?; - Ok(()) - } - async fn cleanup_failed_spawn(&self, sessions: Vec, run_path: &Path) { self.shutdown_sessions(sessions).await; if run_path.exists() @@ -704,56 +518,6 @@ impl InftyOrchestrator { } } - pub fn stream_events( - &self, - conversation_id: ConversationId, - ) -> Result { - self.hub.stream_events(conversation_id) - } - - pub async fn call_role( - &self, - run_id: &str, - role: &str, - text: impl Into, - timeout: Duration, - final_output_json_schema: Option, - ) -> Result { - let handle = session::post_turn( - self.hub.as_ref(), - run_id, - role, - text, - final_output_json_schema, - ) - .await?; - let progress = self.progress.as_deref().map(|reporter| (reporter, role)); - session::await_first_idle(self.hub.as_ref(), &handle, timeout, progress).await - } - - pub async fn relay_assistant_to_role( - &self, - run_id: &str, - target_role: &str, - assistant: &AssistantMessage, - timeout: Duration, - final_output_json_schema: Option, - ) -> Result { - let handle = session::post_turn( - self.hub.as_ref(), - run_id, - target_role, - assistant.message.message.clone(), - final_output_json_schema, - ) - .await?; - let progress = self - .progress - .as_deref() - .map(|reporter| (reporter, target_role)); - session::await_first_idle(self.hub.as_ref(), &handle, timeout, progress).await - } - async fn spawn_and_register_role( &self, run_id: &str, @@ -791,8 +555,15 @@ impl InftyOrchestrator { claim_path: &str, options: &RunExecutionOptions, ) -> Result { - self.collect_verification_summary(sessions, claim_path, None, None, options) - .await + let pool = VerifierPool::from_sessions( + Arc::clone(&self.hub), + sessions, + options.verifier_timeout, + self.progress.clone(), + ); + let req = VerificationRequestPayload::new(claim_path, None, None); + let round = pool.collect_round(&req).await?; + Ok(round.summary) } } @@ -833,168 +604,3 @@ fn collect_session_cleanup(sessions: &RunSessions) -> Vec { cleanup.extend(sessions.verifiers.iter().map(SessionCleanup::new)); cleanup } - -fn parse_solver_signal(message: &str) -> Option { - let trimmed = message.trim(); - if trimmed.is_empty() { - return None; - } - - serde_json::from_str(trimmed) - .or_else(|_| { - strip_json_code_fence(trimmed) - .map(|inner| serde_json::from_str(inner.trim())) - .unwrap_or_else(|| Err(serde_json::Error::custom("invalid payload"))) - }) - .ok() -} - -fn strip_json_code_fence(text: &str) -> Option<&str> { - let trimmed = text.trim(); - if let Some(rest) = trimmed.strip_prefix("```json") { - return rest.strip_suffix("```").map(str::trim); - } - if let Some(rest) = trimmed.strip_prefix("```JSON") { - return rest.strip_suffix("```").map(str::trim); - } - if let Some(rest) = trimmed.strip_prefix("```") { - return rest.strip_suffix("```").map(str::trim); - } - None -} - -fn parse_json_struct(message: &str) -> Result -where - T: DeserializeOwned, -{ - let trimmed = message.trim(); - if trimmed.is_empty() { - return Err(anyhow!("message was empty")); - } - - serde_json::from_str(trimmed) - .or_else(|err| { - strip_json_code_fence(trimmed) - .map(|inner| serde_json::from_str(inner)) - .unwrap_or_else(|| Err(err)) - }) - .map_err(|err| anyhow!(err)) - .with_context(|| format!("failed to parse message as {}", type_name::())) -} - -fn aggregate_verdicts(items: Vec<(String, VerifierVerdict)>) -> AggregatedVerifierVerdict { - let mut overall = VerifierDecision::Pass; - let mut verdicts = Vec::with_capacity(items.len()); - - for (role, verdict) in items { - if !verdict.verdict.is_pass() { - overall = VerifierDecision::Fail; - } - verdicts.push(VerifierReport { - role, - verdict: verdict.verdict, - reasons: verdict.reasons, - suggestions: verdict.suggestions, - }); - } - - AggregatedVerifierVerdict { - kind: "verification_feedback", - overall, - verdicts, - } -} - -fn solver_signal_schema() -> Value { - json!({ - "type": "object", - "properties": { - "type": { - "type": "string", - "enum": ["direction_request", "verification_request", "final_delivery"] - }, - "prompt": { "type": ["string", "null"] }, - "claim_path": { "type": ["string", "null"] }, - "notes": { "type": ["string", "null"] }, - "deliverable_path": { "type": ["string", "null"] }, - "summary": { "type": ["string", "null"] } - }, - "required": [ - "type", - "prompt", - "claim_path", - "notes", - "deliverable_path", - "summary" - ], - "additionalProperties": false - }) -} - -fn final_delivery_schema() -> Value { - json!({ - "type": "object", - "required": ["type", "deliverable_path", "summary"], - "properties": { - "type": { "const": "final_delivery" }, - "deliverable_path": { "type": "string" }, - "summary": { "type": ["string", "null"] } - }, - "additionalProperties": false - }) -} - -fn directive_response_schema() -> Value { - json!({ - "type": "object", - "required": ["directive", "rationale"], - "properties": { - "directive": { "type": "string" }, - "rationale": { "type": ["string", "null"] } - }, - "additionalProperties": false - }) -} - -fn verifier_verdict_schema() -> Value { - json!({ - "type": "object", - "required": ["verdict", "reasons", "suggestions"], - "properties": { - "verdict": { "type": "string", "enum": ["pass", "fail"] }, - "reasons": { "type": "array", "items": { "type": "string" } }, - "suggestions": { "type": "array", "items": { "type": "string" } } - }, - "additionalProperties": false - }) -} - -fn resolve_deliverable_path(base: &Path, candidate: &str) -> Result { - let base_abs = base - .canonicalize() - .with_context(|| format!("failed to canonicalize run store {}", base.display()))?; - - let candidate_path = Path::new(candidate); - let joined = if candidate_path.is_absolute() { - candidate_path.to_path_buf() - } else { - base_abs.join(candidate_path) - }; - - let resolved = joined.canonicalize().with_context(|| { - format!( - "failed to canonicalize deliverable path {}", - joined.display() - ) - })?; - - if !resolved.starts_with(&base_abs) { - bail!( - "deliverable path {} escapes run store {}", - resolved.display(), - base_abs.display() - ); - } - - Ok(resolved) -} diff --git a/codex-rs/codex-infty/src/roles/director.rs b/codex-rs/codex-infty/src/roles/director.rs new file mode 100644 index 0000000000..07abd59d4e --- /dev/null +++ b/codex-rs/codex-infty/src/roles/director.rs @@ -0,0 +1,98 @@ +use std::sync::Arc; +use std::time::Duration; + +use anyhow::Result; +use codex_core::cross_session::AssistantMessage; +use codex_core::cross_session::CrossSessionHub; +use serde::Serialize; +use serde_json::Value; + +use crate::progress::ProgressReporter; +use crate::roles::Role; +use crate::roles::parse_json_struct; +use crate::session; +use crate::signals::DirectiveResponse; + +#[derive(Serialize)] +pub struct DirectionRequestPayload<'a> { + #[serde(rename = "type")] + kind: &'static str, + pub prompt: &'a str, + #[serde(skip_serializing_if = "Option::is_none")] + pub objective: Option<&'a str>, +} + +impl<'a> DirectionRequestPayload<'a> { + pub fn new(prompt: &'a str, objective: Option<&'a str>) -> Self { + Self { + kind: "direction_request", + prompt, + objective, + } + } +} + +pub struct DirectorRole { + hub: Arc, + run_id: String, + role: String, + timeout: Duration, + progress: Option>, +} + +impl DirectorRole { + pub fn new( + hub: Arc, + run_id: impl Into, + role: impl Into, + timeout: Duration, + progress: Option>, + ) -> Self { + Self { + hub, + run_id: run_id.into(), + role: role.into(), + timeout, + progress, + } + } + + pub fn response_schema() -> Value { + serde_json::json!({ + "type": "object", + "required": ["directive", "rationale"], + "properties": { + "directive": { "type": "string" }, + "rationale": { "type": ["string", "null"] } + }, + "additionalProperties": false + }) + } +} + +impl Role, DirectiveResponse> for DirectorRole { + fn call<'a>( + &'a self, + req: &'a DirectionRequestPayload<'a>, + ) -> futures::future::BoxFuture<'a, Result> { + Box::pin(async move { + let request_text = serde_json::to_string_pretty(req)?; + let handle = session::post_turn( + self.hub.as_ref(), + &self.run_id, + &self.role, + request_text, + Some(Self::response_schema()), + ) + .await?; + let progress = self + .progress + .as_deref() + .map(|reporter| (reporter, self.role.as_str())); + let response: AssistantMessage = + session::await_first_idle(self.hub.as_ref(), &handle, self.timeout, progress) + .await?; + parse_json_struct(&response.message.message) + }) + } +} diff --git a/codex-rs/codex-infty/src/roles/mod.rs b/codex-rs/codex-infty/src/roles/mod.rs new file mode 100644 index 0000000000..55fa022f72 --- /dev/null +++ b/codex-rs/codex-infty/src/roles/mod.rs @@ -0,0 +1,49 @@ +use anyhow::Result; +use futures::future::BoxFuture; + +pub mod director; +pub mod solver; +pub mod verifier; +pub mod verifier_pool; + +pub trait Role { + fn call<'a>(&'a self, req: &'a Req) -> BoxFuture<'a, Result>; +} + +// Shared helpers used by role implementations +use anyhow::Context as _; +use anyhow::anyhow; +use std::any::type_name; + +pub(crate) fn strip_json_code_fence(text: &str) -> Option<&str> { + let trimmed = text.trim(); + if let Some(rest) = trimmed.strip_prefix("```json") { + return rest.strip_suffix("```").map(str::trim); + } + if let Some(rest) = trimmed.strip_prefix("```JSON") { + return rest.strip_suffix("```").map(str::trim); + } + if let Some(rest) = trimmed.strip_prefix("```") { + return rest.strip_suffix("```").map(str::trim); + } + None +} + +pub(crate) fn parse_json_struct(message: &str) -> Result +where + T: serde::de::DeserializeOwned, +{ + let trimmed = message.trim(); + if trimmed.is_empty() { + return Err(anyhow!("message was empty")); + } + + serde_json::from_str(trimmed) + .or_else(|err| { + strip_json_code_fence(trimmed) + .map(|inner| serde_json::from_str(inner)) + .unwrap_or_else(|| Err(err)) + }) + .map_err(|err| anyhow!(err)) + .with_context(|| format!("failed to parse message as {}", type_name::())) +} diff --git a/codex-rs/codex-infty/src/roles/solver.rs b/codex-rs/codex-infty/src/roles/solver.rs new file mode 100644 index 0000000000..812d7e5207 --- /dev/null +++ b/codex-rs/codex-infty/src/roles/solver.rs @@ -0,0 +1,216 @@ +use std::sync::Arc; +use std::time::Duration; + +use anyhow::Result; +use codex_core::cross_session::AssistantMessage; +use codex_core::cross_session::CrossSessionHub; +use codex_core::cross_session::SessionEventStream; +use codex_protocol::ConversationId; +use serde::de::Error as _; +use serde_json::Value; + +use crate::progress::ProgressReporter; +use crate::roles::Role; +use crate::session; +use crate::signals::AggregatedVerifierVerdict; +use crate::signals::DirectiveResponse; + +pub struct SolverRole { + hub: Arc, + run_id: String, + role: String, + conversation_id: ConversationId, + progress: Option>, +} + +impl SolverRole { + pub fn new( + hub: Arc, + run_id: impl Into, + role: impl Into, + conversation_id: ConversationId, + progress: Option>, + ) -> Self { + Self { + hub, + run_id: run_id.into(), + role: role.into(), + conversation_id, + progress, + } + } + + pub fn solver_signal_schema() -> Value { + serde_json::json!({ + "type": "object", + "properties": { + "type": { + "type": "string", + "enum": ["direction_request", "verification_request", "final_delivery"] + }, + "prompt": { "type": ["string", "null"] }, + "claim_path": { "type": ["string", "null"] }, + "notes": { "type": ["string", "null"] }, + "deliverable_path": { "type": ["string", "null"] }, + "summary": { "type": ["string", "null"] } + }, + "required": [ + "type", + "prompt", + "claim_path", + "notes", + "deliverable_path", + "summary" + ], + "additionalProperties": false + }) + } + + pub fn final_delivery_schema() -> Value { + serde_json::json!({ + "type": "object", + "required": ["type", "deliverable_path", "summary"], + "properties": { + "type": { "const": "final_delivery" }, + "deliverable_path": { "type": "string" }, + "summary": { "type": ["string", "null"] } + }, + "additionalProperties": false + }) + } + + pub async fn post( + &self, + text: impl Into, + final_output_json_schema: Option, + ) -> Result<()> { + let _ = session::post_turn( + self.hub.as_ref(), + &self.run_id, + &self.role, + text, + final_output_json_schema, + ) + .await?; + Ok(()) + } + + pub fn stream_events( + &self, + ) -> Result { + self.hub.stream_events(self.conversation_id) + } + + pub async fn request_finalization_signal(&self) -> Result<()> { + let handle = session::post_turn( + self.hub.as_ref(), + &self.run_id, + &self.role, + crate::types::FINALIZATION_PROMPT, + Some(Self::final_delivery_schema()), + ) + .await?; + let _ = session::await_first_idle(self.hub.as_ref(), &handle, Duration::from_secs(5), None) + .await?; + Ok(()) + } +} + +pub struct SolverPost { + pub text: String, + pub final_output_json_schema: Option, + pub timeout: Duration, +} + +pub enum SolverRequest { + Directive(DirectiveResponse), + VerificationSummary(AggregatedVerifierVerdict), +} + +impl From for SolverRequest { + fn from(d: DirectiveResponse) -> Self { + SolverRequest::Directive(d) + } +} + +impl From<&AggregatedVerifierVerdict> for SolverRequest { + fn from(v: &AggregatedVerifierVerdict) -> Self { + SolverRequest::VerificationSummary(v.clone()) + } +} + +impl SolverRequest { + fn to_text(&self) -> Result { + match self { + SolverRequest::Directive(d) => Ok(serde_json::to_string_pretty(d)?), + SolverRequest::VerificationSummary(s) => Ok(serde_json::to_string_pretty(s)?), + } + } +} + +impl Role for SolverRole { + fn call<'a>( + &'a self, + req: &'a SolverPost, + ) -> futures::future::BoxFuture<'a, Result> { + Box::pin(async move { + let handle = session::post_turn( + self.hub.as_ref(), + &self.run_id, + &self.role, + req.text.clone(), + req.final_output_json_schema.clone(), + ) + .await?; + let progress = self + .progress + .as_deref() + .map(|reporter| (reporter, self.role.as_str())); + session::await_first_idle(self.hub.as_ref(), &handle, req.timeout, progress).await + }) + } +} + +impl Role for SolverRole { + fn call<'a>(&'a self, req: &'a SolverRequest) -> futures::future::BoxFuture<'a, Result<()>> { + Box::pin(async move { + let text = req.to_text()?; + self.post(text, Some(Self::solver_signal_schema())).await + }) + } +} + +#[derive(Debug, serde::Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum SolverSignal { + DirectionRequest { + #[serde(default)] + prompt: Option, + }, + VerificationRequest { + #[serde(default)] + claim_path: Option, + #[serde(default)] + notes: Option, + }, + FinalDelivery { + #[serde(default)] + deliverable_path: Option, + #[serde(default)] + summary: Option, + }, +} + +pub fn parse_solver_signal(message: &str) -> Option { + let trimmed = message.trim(); + if trimmed.is_empty() { + return None; + } + serde_json::from_str(trimmed) + .or_else(|_| { + crate::roles::strip_json_code_fence(trimmed) + .map(|inner| serde_json::from_str(inner.trim())) + .unwrap_or_else(|| Err(serde_json::Error::custom("invalid payload"))) + }) + .ok() +} diff --git a/codex-rs/codex-infty/src/roles/verifier.rs b/codex-rs/codex-infty/src/roles/verifier.rs new file mode 100644 index 0000000000..c19a87e817 --- /dev/null +++ b/codex-rs/codex-infty/src/roles/verifier.rs @@ -0,0 +1,132 @@ +use std::sync::Arc; +use std::time::Duration; + +use anyhow::Result; +use codex_core::cross_session::AssistantMessage; +use codex_core::cross_session::CrossSessionHub; +use serde::Serialize; +use serde_json::Value; + +use crate::progress::ProgressReporter; +use crate::roles::Role; +use crate::roles::parse_json_struct; +use crate::session; +use crate::signals::VerifierVerdict; + +#[derive(Serialize)] +pub struct VerificationRequestPayload<'a> { + #[serde(rename = "type")] + kind: &'static str, + pub claim_path: &'a str, + #[serde(skip_serializing_if = "Option::is_none")] + pub notes: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + pub objective: Option<&'a str>, +} + +impl<'a> VerificationRequestPayload<'a> { + pub fn new(claim_path: &'a str, notes: Option<&'a str>, objective: Option<&'a str>) -> Self { + Self { + kind: "verification_request", + claim_path, + notes, + objective, + } + } +} + +pub struct VerifierRole { + hub: Arc, + run_id: String, + role: String, + timeout: Duration, + progress: Option>, +} + +impl VerifierRole { + pub fn new( + hub: Arc, + run_id: impl Into, + role: impl Into, + timeout: Duration, + progress: Option>, + ) -> Self { + Self { + hub, + run_id: run_id.into(), + role: role.into(), + timeout, + progress, + } + } + + pub fn role(&self) -> &str { + &self.role + } + + pub fn response_schema() -> Value { + serde_json::json!({ + "type": "object", + "required": ["verdict", "reasons", "suggestions"], + "properties": { + "verdict": { "type": "string", "enum": ["pass", "fail"] }, + "reasons": { "type": "array", "items": { "type": "string" } }, + "suggestions": { "type": "array", "items": { "type": "string" } } + }, + "additionalProperties": false + }) + } +} + +impl Role, VerifierVerdict> for VerifierRole { + fn call<'a>( + &'a self, + req: &'a VerificationRequestPayload<'a>, + ) -> futures::future::BoxFuture<'a, Result> { + Box::pin(async move { + let request_text = serde_json::to_string_pretty(req)?; + let handle = session::post_turn( + self.hub.as_ref(), + &self.run_id, + &self.role, + request_text, + Some(Self::response_schema()), + ) + .await?; + let progress = self + .progress + .as_deref() + .map(|reporter| (reporter, self.role.as_str())); + let response: AssistantMessage = + session::await_first_idle(self.hub.as_ref(), &handle, self.timeout, progress) + .await?; + parse_json_struct(&response.message.message) + }) + } +} + +pub fn aggregate_verdicts(items: Vec<(String, VerifierVerdict)>) -> AggregatedVerifierVerdict { + let mut overall = VerifierDecision::Pass; + let mut verdicts = Vec::with_capacity(items.len()); + + for (role, verdict) in items { + if !verdict.verdict.is_pass() { + overall = VerifierDecision::Fail; + } + verdicts.push(VerifierReport { + role, + verdict: verdict.verdict, + reasons: verdict.reasons, + suggestions: verdict.suggestions, + }); + } + + AggregatedVerifierVerdict { + kind: "verification_feedback", + overall, + verdicts, + } +} +use crate::signals::AggregatedVerifierVerdict; +use crate::signals::VerifierDecision; +use crate::signals::VerifierReport; diff --git a/codex-rs/codex-infty/src/roles/verifier_pool.rs b/codex-rs/codex-infty/src/roles/verifier_pool.rs new file mode 100644 index 0000000000..dafdc49eae --- /dev/null +++ b/codex-rs/codex-infty/src/roles/verifier_pool.rs @@ -0,0 +1,114 @@ +use std::sync::Arc; +use std::time::Duration; + +use anyhow::{Context as _, Result}; +use codex_core::cross_session::CrossSessionHub; + +use crate::progress::ProgressReporter; +use crate::roles::Role; +use crate::roles::verifier::VerificationRequestPayload; +use crate::roles::verifier::VerifierRole; +use crate::roles::verifier::aggregate_verdicts; +use crate::signals::AggregatedVerifierVerdict; +use crate::signals::VerifierVerdict; +use crate::types::RunSessions; + +pub struct VerificationRound { + pub summary: AggregatedVerifierVerdict, + pub passing_roles: Vec, +} + +pub struct VerifierPool { + hub: Arc, + run_id: String, + timeout: Duration, + progress: Option>, + roles: Vec, +} + +impl VerifierPool { + pub fn from_sessions( + hub: Arc, + sessions: &RunSessions, + timeout: Duration, + progress: Option>, + ) -> Self { + let roles = sessions + .verifiers + .iter() + .map(|v| VerifierRole::new(Arc::clone(&hub), sessions.run_id.clone(), v.role.clone(), timeout, progress.clone())) + .collect(); + Self { + hub, + run_id: sessions.run_id.clone(), + timeout, + progress, + roles, + } + } + + pub fn is_empty(&self) -> bool { + self.roles.is_empty() + } + + pub async fn collect_round( + &self, + request: &VerificationRequestPayload<'_>, + ) -> Result { + let futures = self + .roles + .iter() + .map(|role| async { + let name = role.role().to_string(); + let verdict = role.call(request).await; + (name, verdict) + }) + .collect::>(); + let joined = futures::future::join_all(futures).await; + + let mut results: Vec<(String, VerifierVerdict)> = Vec::with_capacity(joined.len()); + let mut passing_roles: Vec = Vec::new(); + for (name, verdict_res) in joined.into_iter() { + let verdict = verdict_res + .with_context(|| format!("verifier {} returned invalid verdict JSON", name))?; + if let Some(progress) = self.progress.as_ref() { + progress.verifier_verdict(&name, &verdict); + } + if verdict.verdict.is_pass() { + passing_roles.push(name.clone()); + } + results.push((name, verdict)); + } + let summary = aggregate_verdicts(results); + Ok(VerificationRound { summary, passing_roles }) + } + + pub fn replace_role(&mut self, role_name: &str) { + if let Some(idx) = self.roles.iter().position(|v| v.role() == role_name) { + self.roles[idx] = VerifierRole::new( + Arc::clone(&self.hub), + self.run_id.clone(), + role_name.to_string(), + self.timeout, + self.progress.clone(), + ); + } + } + + pub async fn rotate_passing_with( + &mut self, + sessions: &mut RunSessions, + passing_roles: &[String], + mut respawn_fn: F, + ) -> Result<()> + where + F: FnMut(&str, &mut RunSessions) -> Fut, + Fut: std::future::Future>, + { + for role in passing_roles { + respawn_fn(role.as_str(), sessions).await?; + self.replace_role(role); + } + Ok(()) + } +} diff --git a/codex-rs/codex-infty/src/signals.rs b/codex-rs/codex-infty/src/signals.rs index f2d26bc618..a606309e1b 100644 --- a/codex-rs/codex-infty/src/signals.rs +++ b/codex-rs/codex-infty/src/signals.rs @@ -21,7 +21,7 @@ impl VerifierDecision { } } -#[derive(Debug, Deserialize, Serialize)] +#[derive(Debug, Deserialize, Serialize, Clone)] pub struct VerifierVerdict { pub verdict: VerifierDecision, #[serde(default)] @@ -30,7 +30,7 @@ pub struct VerifierVerdict { pub suggestions: Vec, } -#[derive(Debug, Serialize)] +#[derive(Debug, Serialize, Clone)] pub struct VerifierReport { pub role: String, pub verdict: VerifierDecision, @@ -40,10 +40,16 @@ pub struct VerifierReport { pub suggestions: Vec, } -#[derive(Debug, Serialize)] +#[derive(Debug, Serialize, Clone)] pub struct AggregatedVerifierVerdict { #[serde(rename = "type")] pub kind: &'static str, pub overall: VerifierDecision, pub verdicts: Vec, } + +impl From<&AggregatedVerifierVerdict> for String { + fn from(value: &AggregatedVerifierVerdict) -> Self { + serde_json::to_string_pretty(value).unwrap_or_else(|_| "{}".to_string()) + } +} diff --git a/codex-rs/codex-infty/src/utils.rs b/codex-rs/codex-infty/src/utils.rs new file mode 100644 index 0000000000..9bf95b773c --- /dev/null +++ b/codex-rs/codex-infty/src/utils.rs @@ -0,0 +1,56 @@ +use std::path::{Path, PathBuf}; + +use anyhow::{anyhow, bail, Context, Result}; + +pub fn trim_to_non_empty(opt: Option) -> Option { + opt.and_then(|s| { + let trimmed = s.trim(); + if trimmed.is_empty() { + None + } else { + Some(trimmed.to_string()) + } + }) +} + +pub fn required_trimmed(opt: Option, err_msg: &str) -> Result { + trim_to_non_empty(opt).ok_or_else(|| anyhow!(err_msg.to_string())) +} + +pub fn resolve_deliverable_path(base: &Path, candidate: &str) -> Result { + let base_abs = base + .canonicalize() + .with_context(|| format!("failed to canonicalize run store {}", base.display()))?; + + let candidate_path = Path::new(candidate); + let joined = if candidate_path.is_absolute() { + candidate_path.to_path_buf() + } else { + base_abs.join(candidate_path) + }; + + let resolved = joined.canonicalize().with_context(|| { + format!( + "failed to canonicalize deliverable path {}", + joined.display() + ) + })?; + + if !resolved.starts_with(&base_abs) { + bail!( + "deliverable path {} escapes run store {}", + resolved.display(), + base_abs.display() + ); + } + + Ok(resolved) +} + +pub fn objective_as_str(options: &crate::types::RunExecutionOptions) -> Option<&str> { + options + .objective + .as_deref() + .map(str::trim) + .filter(|s| !s.is_empty()) +} diff --git a/codex-rs/codex-infty/tests/orchestrator.rs b/codex-rs/codex-infty/tests/orchestrator.rs index 00d3eee735..3ef773ea45 100644 --- a/codex-rs/codex-infty/tests/orchestrator.rs +++ b/codex-rs/codex-infty/tests/orchestrator.rs @@ -6,6 +6,7 @@ use codex_core::CodexAuth; use codex_core::built_in_model_providers; use codex_core::config::Config; use codex_core::protocol::Op; +use codex_core::cross_session::{AssistantMessage, PostUserTurnRequest, RoleOrId}; use codex_infty::InftyOrchestrator; use codex_infty::RoleConfig; use codex_infty::RunExecutionOptions; @@ -58,37 +59,34 @@ async fn orchestrator_routes_between_roles_and_records_store() -> anyhow::Result }) .await?; - let solver_message = orchestrator - .call_role( - &sessions.run_id, - "solver", - "kick off plan", - Duration::from_secs(1), - None, - ) - .await?; + let solver_message = call_role( + &orchestrator, + &sessions.run_id, + "solver", + "kick off plan", + Duration::from_secs(1), + ) + .await?; assert_eq!(solver_message.message.message, "Need direction"); - let director_message = orchestrator - .relay_assistant_to_role( - &sessions.run_id, - "director", - &solver_message, - Duration::from_secs(1), - None, - ) - .await?; + let director_message = relay_assistant_to_role( + &orchestrator, + &sessions.run_id, + "director", + &solver_message, + Duration::from_secs(1), + ) + .await?; assert_eq!(director_message.message.message, "Proceed iteratively"); - let solver_reply = orchestrator - .relay_assistant_to_role( - &sessions.run_id, - "solver", - &director_message, - Duration::from_secs(1), - None, - ) - .await?; + let solver_reply = relay_assistant_to_role( + &orchestrator, + &sessions.run_id, + "solver", + &director_message, + Duration::from_secs(1), + ) + .await?; assert_eq!(solver_reply.message.message, "Acknowledged"); assert_eq!(response_mock.requests().len(), 3); @@ -303,3 +301,42 @@ async fn build_config(server: &MockServer) -> anyhow::Result { config.model_provider = provider; Ok(config) } + +async fn call_role( + orchestrator: &InftyOrchestrator, + run_id: &str, + role: &str, + text: &str, + timeout: Duration, +) -> anyhow::Result { + let hub = orchestrator.hub(); + let handle = hub + .post_user_turn(PostUserTurnRequest { + target: RoleOrId::RunRole { + run_id: run_id.to_string(), + role: role.to_string(), + }, + text: text.to_string(), + final_output_json_schema: None, + }) + .await?; + let reply = hub.await_first_assistant(&handle, timeout).await?; + Ok(reply) +} + +async fn relay_assistant_to_role( + orchestrator: &InftyOrchestrator, + run_id: &str, + target_role: &str, + assistant: &AssistantMessage, + timeout: Duration, +) -> anyhow::Result { + call_role( + orchestrator, + run_id, + target_role, + &assistant.message.message, + timeout, + ) + .await +} diff --git a/codex-rs/codex-infty/tests/verifier_replacement.rs b/codex-rs/codex-infty/tests/verifier_replacement.rs new file mode 100644 index 0000000000..f39a6ca413 --- /dev/null +++ b/codex-rs/codex-infty/tests/verifier_replacement.rs @@ -0,0 +1,155 @@ +#![cfg(not(target_os = "windows"))] + +use std::time::Duration; + +use codex_core::CodexAuth; +use codex_core::built_in_model_providers; +use codex_core::config::Config; +use codex_infty::InftyOrchestrator; +use codex_infty::RoleConfig; +use codex_infty::RunExecutionOptions; +use codex_infty::RunParams; +use core_test_support::load_default_config_for_test; +use core_test_support::responses; +use core_test_support::skip_if_no_network; +use tempfile::TempDir; +use wiremock::MockServer; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn replaces_passing_verifiers_and_keeps_failing() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + + let server = responses::start_mock_server().await; + + // Round 1: alpha passes, beta fails + let body_verifier_alpha_r1 = responses::sse(vec![ + responses::ev_response_created("verifier-alpha-r1"), + responses::ev_assistant_message( + "verifier-alpha-msg-r1", + r#"{"verdict":"pass","reasons":[],"suggestions":[]}"#, + ), + responses::ev_completed("verifier-alpha-r1"), + ]); + let body_verifier_beta_r1 = responses::sse(vec![ + responses::ev_response_created("verifier-beta-r1"), + responses::ev_assistant_message( + "verifier-beta-msg-r1", + r#"{"verdict":"fail","reasons":["missing"],"suggestions":[]}"#, + ), + responses::ev_completed("verifier-beta-r1"), + ]); + + // Round 2: both pass + let body_verifier_alpha_r2 = responses::sse(vec![ + responses::ev_response_created("verifier-alpha-r2"), + responses::ev_assistant_message( + "verifier-alpha-msg-r2", + r#"{"verdict":"pass","reasons":[],"suggestions":[]}"#, + ), + responses::ev_completed("verifier-alpha-r2"), + ]); + let body_verifier_beta_r2 = responses::sse(vec![ + responses::ev_response_created("verifier-beta-r2"), + responses::ev_assistant_message( + "verifier-beta-msg-r2", + r#"{"verdict":"pass","reasons":[],"suggestions":[]}"#, + ), + responses::ev_completed("verifier-beta-r2"), + ]); + + // Mount verifier SSE bodies in the exact order collect_verification_summary posts to verifiers. + // The implementation posts sequentially in the order of sessions.verifiers. + let _m1 = responses::mount_sse_once(&server, body_verifier_alpha_r1).await; + let _m2 = responses::mount_sse_once(&server, body_verifier_beta_r1).await; + let _m3 = responses::mount_sse_once(&server, body_verifier_alpha_r2).await; + let _m4 = responses::mount_sse_once(&server, body_verifier_beta_r2).await; + + let runs_root = TempDir::new()?; + let orchestrator = + InftyOrchestrator::with_runs_root(CodexAuth::from_api_key("dummy-key"), runs_root.path()); + let run_id = "run-verifier-replacement".to_string(); + + let solver_config = build_config(&server).await?; + let director_config = build_config(&server).await?; + let verifier_config = build_config(&server).await?; + + // Spawn run with two verifiers in known order. + let mut sessions = orchestrator + .spawn_run(RunParams { + run_id: run_id.clone(), + run_root: Some(runs_root.path().join("runs").join(&run_id)), + solver: RoleConfig::new("solver", solver_config), + director: RoleConfig::new("director", director_config), + verifiers: vec![ + RoleConfig::new("verifier-alpha", verifier_config.clone()), + RoleConfig::new("verifier-beta", verifier_config), + ], + }) + .await?; + + let alpha_initial = sessions + .store + .role_metadata("verifier-alpha") + .and_then(|m| m.rollout_path.clone()) + .expect("alpha initial rollout path"); + let beta_initial = sessions + .store + .role_metadata("verifier-beta") + .and_then(|m| m.rollout_path.clone()) + .expect("beta initial rollout path"); + + let mut options = RunExecutionOptions::default(); + options.verifier_timeout = Duration::from_secs(2); + + // Round 1: alpha pass (should be replaced), beta fail (should be kept) + let _summary1 = orchestrator + .verify_round_for_test(&mut sessions, "memory/claims/c1.json", &options) + .await?; + + let alpha_after_r1 = sessions + .store + .role_metadata("verifier-alpha") + .and_then(|m| m.rollout_path.clone()) + .expect("alpha rollout after r1"); + let beta_after_r1 = sessions + .store + .role_metadata("verifier-beta") + .and_then(|m| m.rollout_path.clone()) + .expect("beta rollout after r1"); + + assert_ne!( + alpha_initial, alpha_after_r1, + "alpha should be replaced after pass" + ); + assert_eq!( + beta_initial, beta_after_r1, + "beta should be kept after fail" + ); + + // Round 2: both pass; beta should be replaced now. + let _summary2 = orchestrator + .verify_round_for_test(&mut sessions, "memory/claims/c2.json", &options) + .await?; + let beta_after_r2 = sessions + .store + .role_metadata("verifier-beta") + .and_then(|m| m.rollout_path.clone()) + .expect("beta rollout after r2"); + assert_ne!( + beta_initial, beta_after_r2, + "beta should be replaced after pass in r2" + ); + + Ok(()) +} + +async fn build_config(server: &MockServer) -> anyhow::Result { + let home = TempDir::new()?; + let cwd = TempDir::new()?; + let mut config = load_default_config_for_test(&home); + config.cwd = cwd.path().to_path_buf(); + let mut provider = built_in_model_providers()["openai"].clone(); + provider.base_url = Some(format!("{}/v1", server.uri())); + config.model_provider = provider; + Ok(config) +}