From 119dabd2727a45b85db2a856a8691ea4346e9a21 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 16 Oct 2025 10:30:34 +0100 Subject: [PATCH] R3 --- codex-rs/codex-infty/src/lib.rs | 2 +- codex-rs/codex-infty/src/orchestrator.rs | 57 +++++++++---------- .../codex-infty/src/roles/verifier_pool.rs | 35 +++++------- codex-rs/codex-infty/src/utils.rs | 8 ++- codex-rs/codex-infty/tests/orchestrator.rs | 4 +- 5 files changed, 51 insertions(+), 55 deletions(-) diff --git a/codex-rs/codex-infty/src/lib.rs b/codex-rs/codex-infty/src/lib.rs index 886e50ec1b..af2c66127f 100644 --- a/codex-rs/codex-infty/src/lib.rs +++ b/codex-rs/codex-infty/src/lib.rs @@ -3,12 +3,12 @@ mod orchestrator; mod progress; mod prompts; -pub(crate) mod utils; mod roles; mod run_store; mod session; mod signals; mod types; +pub(crate) mod utils; pub use orchestrator::InftyOrchestrator; pub use progress::ProgressReporter; diff --git a/codex-rs/codex-infty/src/orchestrator.rs b/codex-rs/codex-infty/src/orchestrator.rs index 6480ccf95a..bf747fe96f 100644 --- a/codex-rs/codex-infty/src/orchestrator.rs +++ b/codex-rs/codex-infty/src/orchestrator.rs @@ -42,6 +42,12 @@ use crate::types::RunOutcome; use crate::types::RunParams; use crate::types::RunSessions; +#[derive(Default)] +struct LoopState { + waiting_for_signal: bool, + pending_solver_turn_completion: bool, +} + struct SessionCleanup { conversation_id: ConversationId, conversation: Arc, @@ -64,6 +70,9 @@ pub struct InftyOrchestrator { } impl InftyOrchestrator { + fn progress_ref(&self) -> Option<&dyn ProgressReporter> { + self.progress.as_deref() + } pub fn new(auth: CodexAuth) -> Result { let runs_root = crate::default_runs_root()?; Ok(Self::with_runs_root(auth, runs_root)) @@ -202,17 +211,16 @@ impl InftyOrchestrator { ); let mut solver_events = solver_role.stream_events()?; - let mut waiting_for_signal = false; - let mut pending_solver_turn_completion = false; + let mut state = LoopState::default(); if let Some(objective) = &options.objective { 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() { - progress.objective_posted(objective); - progress.waiting_for_solver(); + state.waiting_for_signal = true; + if let Some(p) = self.progress_ref() { + p.objective_posted(objective); + p.waiting_for_solver(); } } @@ -225,25 +233,19 @@ impl InftyOrchestrator { let Some(event) = maybe_event else { break 'event_loop; }; - if let Some(progress) = self.progress.as_ref() { - progress.solver_event(&event.event.msg); - } + if let Some(p) = self.progress_ref() { p.solver_event(&event.event.msg); } match &event.event.msg { EventMsg::AgentMessage(agent_msg) => { - if let Some(progress) = self.progress.as_ref() { - progress.solver_agent_message(agent_msg); - } + if let Some(p) = self.progress_ref() { p.solver_agent_message(agent_msg); } if let Some(signal) = parse_solver_signal(&agent_msg.message) { - waiting_for_signal = false; + state.waiting_for_signal = false; match signal { SolverSignal::DirectionRequest { prompt } => { 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); - } + if let Some(p) = self.progress_ref() { p.direction_request(&prompt); } self .handle_direction_request( &prompt, @@ -253,19 +255,14 @@ impl InftyOrchestrator { ) .await?; sessions.store.touch()?; - pending_solver_turn_completion = true; + state.pending_solver_turn_completion = true; } SolverSignal::VerificationRequest { claim_path, notes } => { 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, - notes.as_deref(), - ); - } + if let Some(p) = self.progress_ref() { p.verification_request(&claim_path, notes.as_deref()); } let verified = self .handle_verification_request( sessions, @@ -278,7 +275,7 @@ impl InftyOrchestrator { .await?; sessions.store.touch()?; if verified { - pending_solver_turn_completion = true; + state.pending_solver_turn_completion = true; } } SolverSignal::FinalDelivery { @@ -298,9 +295,7 @@ impl InftyOrchestrator { )?; 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); - } + if let Some(p) = self.progress_ref() { p.final_delivery(&resolved, summary_ref); } let verified = self .run_final_verification( sessions, @@ -312,7 +307,7 @@ impl InftyOrchestrator { ) .await?; if !verified { - pending_solver_turn_completion = true; + state.pending_solver_turn_completion = true; continue; } sessions.store.touch()?; @@ -327,12 +322,12 @@ impl InftyOrchestrator { } } EventMsg::TaskComplete(..) => { - if waiting_for_signal { + if state.waiting_for_signal { // The solver completed its turn without issuing a signal; ask for one now. solver_role.request_finalization_signal().await?; - } else if pending_solver_turn_completion { + } else if state.pending_solver_turn_completion { // We handled a signal earlier in the loop; this completion corresponds to it. - pending_solver_turn_completion = false; + state.pending_solver_turn_completion = false; } } _ => {} diff --git a/codex-rs/codex-infty/src/roles/verifier_pool.rs b/codex-rs/codex-infty/src/roles/verifier_pool.rs index dafdc49eae..6ea783abcd 100644 --- a/codex-rs/codex-infty/src/roles/verifier_pool.rs +++ b/codex-rs/codex-infty/src/roles/verifier_pool.rs @@ -1,7 +1,8 @@ use std::sync::Arc; use std::time::Duration; -use anyhow::{Context as _, Result}; +use anyhow::Context as _; +use anyhow::Result; use codex_core::cross_session::CrossSessionHub; use crate::progress::ProgressReporter; @@ -36,7 +37,15 @@ impl VerifierPool { let roles = sessions .verifiers .iter() - .map(|v| VerifierRole::new(Arc::clone(&hub), sessions.run_id.clone(), v.role.clone(), timeout, progress.clone())) + .map(|v| { + VerifierRole::new( + Arc::clone(&hub), + sessions.run_id.clone(), + v.role.clone(), + timeout, + progress.clone(), + ) + }) .collect(); Self { hub, @@ -80,7 +89,10 @@ impl VerifierPool { results.push((name, verdict)); } let summary = aggregate_verdicts(results); - Ok(VerificationRound { summary, passing_roles }) + Ok(VerificationRound { + summary, + passing_roles, + }) } pub fn replace_role(&mut self, role_name: &str) { @@ -94,21 +106,4 @@ impl VerifierPool { ); } } - - 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/utils.rs b/codex-rs/codex-infty/src/utils.rs index 9bf95b773c..44f41821b0 100644 --- a/codex-rs/codex-infty/src/utils.rs +++ b/codex-rs/codex-infty/src/utils.rs @@ -1,6 +1,10 @@ -use std::path::{Path, PathBuf}; +use std::path::Path; +use std::path::PathBuf; -use anyhow::{anyhow, bail, Context, Result}; +use anyhow::Context; +use anyhow::Result; +use anyhow::anyhow; +use anyhow::bail; pub fn trim_to_non_empty(opt: Option) -> Option { opt.and_then(|s| { diff --git a/codex-rs/codex-infty/tests/orchestrator.rs b/codex-rs/codex-infty/tests/orchestrator.rs index 3ef773ea45..39b3a6a631 100644 --- a/codex-rs/codex-infty/tests/orchestrator.rs +++ b/codex-rs/codex-infty/tests/orchestrator.rs @@ -5,8 +5,10 @@ use std::time::Duration; use codex_core::CodexAuth; use codex_core::built_in_model_providers; use codex_core::config::Config; +use codex_core::cross_session::AssistantMessage; +use codex_core::cross_session::PostUserTurnRequest; +use codex_core::cross_session::RoleOrId; 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;