diff --git a/codex-rs/codex-infty/src/lib.rs b/codex-rs/codex-infty/src/lib.rs index c586542b43..396fbaf2dd 100644 --- a/codex-rs/codex-infty/src/lib.rs +++ b/codex-rs/codex-infty/src/lib.rs @@ -185,6 +185,22 @@ struct AggregatedVerifierVerdict { verdicts: Vec, } +#[derive(Serialize)] +struct DirectionRequestPayload<'a> { + #[serde(rename = "type")] + kind: &'static str, + prompt: &'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>, +} + struct SessionCleanup { conversation_id: ConversationId, conversation: Arc, @@ -268,25 +284,10 @@ impl InftyOrchestrator { let mut cleanup = Vec::new(); let solver_session = match self - .spawn_role_session(&run_id, &run_path, solver.clone()) + .spawn_and_register_role(&run_id, &run_path, &solver, &mut store, &mut cleanup) .await { - Ok(session) => { - cleanup.push(SessionCleanup::new(&session)); - if let Err(err) = - store.update_rollout_path(&session.role, session.rollout_path.clone()) - { - self.cleanup_failed_spawn(cleanup, &run_path).await; - return Err(err); - } - if let Some(path) = solver.config_path.clone() { - if let Err(err) = store.set_role_config_path(&session.role, path) { - self.cleanup_failed_spawn(cleanup, &run_path).await; - return Err(err); - } - } - session - } + Ok(session) => session, Err(err) => { self.cleanup_failed_spawn(cleanup, &run_path).await; return Err(err); @@ -294,25 +295,10 @@ impl InftyOrchestrator { }; let director_session = match self - .spawn_role_session(&run_id, &run_path, director.clone()) + .spawn_and_register_role(&run_id, &run_path, &director, &mut store, &mut cleanup) .await { - Ok(session) => { - cleanup.push(SessionCleanup::new(&session)); - if let Err(err) = - store.update_rollout_path(&session.role, session.rollout_path.clone()) - { - self.cleanup_failed_spawn(cleanup, &run_path).await; - return Err(err); - } - if let Some(path) = director.config_path.clone() { - if let Err(err) = store.set_role_config_path(&session.role, path) { - self.cleanup_failed_spawn(cleanup, &run_path).await; - return Err(err); - } - } - session - } + Ok(session) => session, Err(err) => { self.cleanup_failed_spawn(cleanup, &run_path).await; return Err(err); @@ -322,25 +308,10 @@ impl InftyOrchestrator { let mut verifier_sessions = Vec::with_capacity(verifiers.len()); for verifier in verifiers { let session = match self - .spawn_role_session(&run_id, &run_path, verifier.clone()) + .spawn_and_register_role(&run_id, &run_path, &verifier, &mut store, &mut cleanup) .await { - Ok(session) => { - cleanup.push(SessionCleanup::new(&session)); - if let Some(path) = verifier.config_path.clone() { - if let Err(err) = store.set_role_config_path(&session.role, path) { - self.cleanup_failed_spawn(cleanup, &run_path).await; - return Err(err); - } - } - if let Err(err) = - store.update_rollout_path(&session.role, session.rollout_path.clone()) - { - self.cleanup_failed_spawn(cleanup, &run_path).await; - return Err(err); - } - session - } + Ok(session) => session, Err(err) => { self.cleanup_failed_spawn(cleanup, &run_path).await; return Err(err); @@ -373,19 +344,10 @@ impl InftyOrchestrator { let run_path = store.path().to_path_buf(); let solver_session = match self - .resume_role_session(&run_id, &run_path, &solver, &store) + .resume_and_register_role(&run_id, &run_path, &solver, &mut store, &mut cleanup) .await { - Ok(session) => { - cleanup.push(SessionCleanup::new(&session)); - if let Err(err) = - store.update_rollout_path(&session.role, session.rollout_path.clone()) - { - self.cleanup_failed_resume(cleanup).await; - return Err(err); - } - session - } + Ok(session) => session, Err(err) => { self.cleanup_failed_resume(cleanup).await; return Err(err); @@ -393,19 +355,10 @@ impl InftyOrchestrator { }; let director_session = match self - .resume_role_session(&run_id, &run_path, &director, &store) + .resume_and_register_role(&run_id, &run_path, &director, &mut store, &mut cleanup) .await { - Ok(session) => { - cleanup.push(SessionCleanup::new(&session)); - if let Err(err) = - store.update_rollout_path(&session.role, session.rollout_path.clone()) - { - self.cleanup_failed_resume(cleanup).await; - return Err(err); - } - session - } + Ok(session) => session, Err(err) => { self.cleanup_failed_resume(cleanup).await; return Err(err); @@ -415,19 +368,10 @@ impl InftyOrchestrator { let mut verifier_sessions = Vec::with_capacity(verifiers.len()); for verifier in verifiers.iter() { let session = match self - .resume_role_session(&run_id, &run_path, verifier, &store) + .resume_and_register_role(&run_id, &run_path, verifier, &mut store, &mut cleanup) .await { - Ok(session) => { - cleanup.push(SessionCleanup::new(&session)); - if let Err(err) = - store.update_rollout_path(&session.role, session.rollout_path.clone()) - { - self.cleanup_failed_resume(cleanup).await; - return Err(err); - } - session - } + Ok(session) => session, Err(err) => { self.cleanup_failed_resume(cleanup).await; return Err(err); @@ -465,9 +409,14 @@ impl InftyOrchestrator { ) -> Result { let mut solver_events = self.stream_events(sessions.solver.conversation_id)?; - if let Some(objective) = options.objective.clone() { - self.post_to_role(&sessions.run_id, &sessions.solver.role, objective, None) - .await?; + if let Some(objective) = &options.objective { + self.post_to_role( + &sessions.run_id, + &sessions.solver.role, + objective.as_str(), + None, + ) + .await?; sessions.store.touch()?; } @@ -477,21 +426,16 @@ impl InftyOrchestrator { if let Some(signal) = parse_solver_signal(&agent_msg.message) { match signal { SolverSignal::DirectionRequest { prompt } => { - let _ = prompt; - self.handle_direction_request( - sessions, - &agent_msg.message, - options, - ) - .await?; + self.handle_direction_request(sessions, &prompt, options) + .await?; sessions.store.touch()?; } SolverSignal::VerificationRequest { claim_path, notes } => { - let _ = (claim_path, notes); let pass = self .handle_verification_request( sessions, - &agent_msg.message, + &claim_path, + notes.as_deref(), options, ) .await?; @@ -547,14 +491,19 @@ impl InftyOrchestrator { async fn handle_direction_request( &self, sessions: &RunSessions, - raw_message: &str, + prompt: &str, options: &RunExecutionOptions, ) -> Result<()> { + let request = DirectionRequestPayload { + kind: "direction_request", + prompt, + }; + let request_text = serde_json::to_string_pretty(&request)?; let handle = self .post_to_role( &sessions.run_id, &sessions.director.role, - raw_message.to_string(), + request_text, Some(director_schema()), ) .await?; @@ -578,7 +527,8 @@ impl InftyOrchestrator { async fn handle_verification_request( &self, sessions: &RunSessions, - raw_message: &str, + claim_path: &str, + notes: Option<&str>, options: &RunExecutionOptions, ) -> Result { if sessions.verifiers.is_empty() { @@ -590,14 +540,21 @@ impl InftyOrchestrator { return Ok(true); } + let request = VerificationRequestPayload { + kind: "verification_request", + claim_path, + notes, + }; + let request_text = serde_json::to_string_pretty(&request)?; let mut collected = Vec::with_capacity(sessions.verifiers.len()); + let schema = verifier_schema(); for verifier in &sessions.verifiers { let handle = self .post_to_role( &sessions.run_id, &verifier.role, - raw_message.to_string(), - Some(verifier_schema()), + request_text.as_str(), + Some(schema.clone()), ) .await?; let response = self @@ -620,14 +577,14 @@ impl InftyOrchestrator { async fn cleanup_failed_spawn(&self, sessions: Vec, run_path: &Path) { self.shutdown_sessions(sessions).await; - if run_path.exists() { - if let Err(err) = fs::remove_dir_all(run_path) { - warn!( - path = %run_path.display(), - ?err, - "failed to remove run directory after spawn failure" - ); - } + if run_path.exists() + && let Err(err) = fs::remove_dir_all(run_path) + { + warn!( + path = %run_path.display(), + ?err, + "failed to remove run directory after spawn failure" + ); } } @@ -721,6 +678,44 @@ impl InftyOrchestrator { self.hub.stream_events(conversation_id) } + async fn spawn_and_register_role( + &self, + run_id: &str, + run_path: &Path, + role_config: &RoleConfig, + store: &mut RunStore, + cleanup: &mut Vec, + ) -> Result { + let session = self + .spawn_role_session(run_id, run_path, role_config.clone()) + .await?; + cleanup.push(SessionCleanup::new(&session)); + store.update_rollout_path(&session.role, session.rollout_path.clone())?; + if let Some(path) = role_config.config_path.clone() { + store.set_role_config_path(&session.role, path)?; + } + Ok(session) + } + + async fn resume_and_register_role( + &self, + run_id: &str, + run_path: &Path, + role_config: &RoleConfig, + store: &mut RunStore, + cleanup: &mut Vec, + ) -> Result { + let session = self + .resume_role_session(run_id, run_path, role_config, store) + .await?; + cleanup.push(SessionCleanup::new(&session)); + store.update_rollout_path(&session.role, session.rollout_path.clone())?; + if let Some(path) = role_config.config_path.clone() { + store.set_role_config_path(&session.role, path)?; + } + Ok(session) + } + async fn spawn_role_session( &self, run_id: &str, diff --git a/codex-rs/codex-infty/src/prompts.rs b/codex-rs/codex-infty/src/prompts.rs index 0f50d81e67..60decc846e 100644 --- a/codex-rs/codex-infty/src/prompts.rs +++ b/codex-rs/codex-infty/src/prompts.rs @@ -71,10 +71,10 @@ Use `"fail"` when the claim is not ready. Populate `reasons` with concrete block Do not include extra commentary outside the JSON payload."#; pub fn ensure_instructions(role: &str, config: &mut Config) { - if config.base_instructions.is_none() { - if let Some(text) = default_instructions_for_role(role) { - config.base_instructions = Some(text.to_string()); - } + if config.base_instructions.is_none() + && let Some(text) = default_instructions_for_role(role) + { + config.base_instructions = Some(text.to_string()); } } diff --git a/codex-rs/codex-infty/tests/orchestrator.rs b/codex-rs/codex-infty/tests/orchestrator.rs index 122e77fded..1d7c44e55e 100644 --- a/codex-rs/codex-infty/tests/orchestrator.rs +++ b/codex-rs/codex-infty/tests/orchestrator.rs @@ -246,8 +246,10 @@ async fn execute_new_run_drives_to_completion() -> anyhow::Result<()> { let director_config = build_config(&server).await?; let verifier_config = build_config(&server).await?; - let mut options = RunExecutionOptions::default(); - options.objective = Some("Implement feature".to_string()); + let options = RunExecutionOptions { + objective: Some("Implement feature".to_string()), + ..RunExecutionOptions::default() + }; let outcome = orchestrator .execute_new_run(