From c0f8a49e3e552cdb6471f00c22fa7505ac9a5009 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Tue, 14 Oct 2025 17:03:43 +0100 Subject: [PATCH] P1 --- codex-rs/codex-infty/src/lib.rs | 1 + codex-rs/codex-infty/src/orchestrator.rs | 340 ++++++++--------------- codex-rs/codex-infty/src/session.rs | 134 +++++++++ codex-rs/infty2.md | 325 ++++++++++++++++++++++ 4 files changed, 578 insertions(+), 222 deletions(-) create mode 100644 codex-rs/codex-infty/src/session.rs create mode 100644 codex-rs/infty2.md diff --git a/codex-rs/codex-infty/src/lib.rs b/codex-rs/codex-infty/src/lib.rs index 19d7e346a1..c358e88a6e 100644 --- a/codex-rs/codex-infty/src/lib.rs +++ b/codex-rs/codex-infty/src/lib.rs @@ -4,6 +4,7 @@ mod orchestrator; mod progress; mod prompts; mod run_store; +mod session; mod signals; mod types; diff --git a/codex-rs/codex-infty/src/orchestrator.rs b/codex-rs/codex-infty/src/orchestrator.rs index 6afb72e2fa..afda611769 100644 --- a/codex-rs/codex-infty/src/orchestrator.rs +++ b/codex-rs/codex-infty/src/orchestrator.rs @@ -12,11 +12,8 @@ use anyhow::bail; use codex_core::CodexAuth; use codex_core::CodexConversation; use codex_core::ConversationManager; -use codex_core::CrossSessionSpawnParams; use codex_core::cross_session::AssistantMessage; use codex_core::cross_session::CrossSessionHub; -use codex_core::cross_session::PostUserTurnRequest; -use codex_core::cross_session::RoleOrId; use codex_core::cross_session::SessionEventStream; use codex_core::protocol::EventMsg; use codex_core::protocol::Op; @@ -28,7 +25,6 @@ use serde::de::Error as _; use serde_json::Value; use serde_json::json; use tokio::signal; -use tokio::time::Instant; use tokio_stream::StreamExt; use tracing::warn; @@ -36,6 +32,7 @@ use crate::progress::ProgressReporter; use crate::prompts; 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; @@ -300,7 +297,8 @@ impl InftyOrchestrator { let mut waiting_for_signal = false; let mut pending_solver_turn_completion = false; if let Some(objective) = &options.objective { - self.post_to_role( + session::post_turn( + self.hub.as_ref(), &sessions.run_id, &sessions.solver.role, objective.as_str(), @@ -486,31 +484,39 @@ impl InftyOrchestrator { objective: options.objective.as_deref(), }; let request_text = serde_json::to_string_pretty(&request)?; - let handle = self - .post_to_role( - &sessions.run_id, - &sessions.director.role, - request_text, - Some(directive_response_schema()), - ) - .await?; - let directive = self - .await_first_assistant_idle(&handle, options.director_timeout, Some("director")) - .await?; + 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) .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)?; - let _ = self - .post_to_role( - &sessions.run_id, - &sessions.solver.role, - directive_text, - Some(solver_signal_schema()), - ) - .await?; + session::post_turn( + self.hub.as_ref(), + &sessions.run_id, + &sessions.solver.role, + directive_text, + Some(solver_signal_schema()), + ) + .await?; Ok(()) } @@ -571,16 +577,15 @@ impl InftyOrchestrator { } async fn request_solver_signal(&self, run_id: &str, solver_role: &str) -> Result<()> { - let handle = self - .post_to_role( - run_id, - solver_role, - FINALIZATION_PROMPT, - Some(final_delivery_schema()), - ) - .await?; - let _ = self - .await_first_assistant_idle(&handle, Duration::from_secs(5), None) + 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(()) } @@ -606,17 +611,25 @@ impl InftyOrchestrator { let request_text = serde_json::to_string_pretty(&request)?; let mut collected = Vec::with_capacity(sessions.verifiers.len()); for verifier in &sessions.verifiers { - let handle = self - .post_to_role( - &sessions.run_id, - &verifier.role, - request_text.as_str(), - Some(verifier_verdict_schema()), - ) - .await?; - let response = self - .await_first_assistant_idle(&handle, options.verifier_timeout, Some(&verifier.role)) - .await?; + 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) @@ -642,14 +655,14 @@ impl InftyOrchestrator { summary: &AggregatedVerifierVerdict, ) -> Result<()> { let summary_text = serde_json::to_string_pretty(summary)?; - let _ = self - .post_to_role( - &sessions.run_id, - &sessions.solver.role, - summary_text, - Some(solver_signal_schema()), - ) - .await?; + session::post_turn( + self.hub.as_ref(), + &sessions.run_id, + &sessions.solver.role, + summary_text, + Some(solver_signal_schema()), + ) + .await?; Ok(()) } @@ -686,90 +699,11 @@ impl InftyOrchestrator { } } - pub async fn post_to_role( + pub fn stream_events( &self, - run_id: &str, - role: &str, - text: impl Into, - final_output_json_schema: Option, - ) -> Result { - let handle = self - .hub - .post_user_turn(PostUserTurnRequest { - target: RoleOrId::RunRole { - run_id: run_id.to_string(), - role: role.to_string(), - }, - text: text.into(), - final_output_json_schema, - }) - .await?; - Ok(handle) - } - - pub async fn await_first_assistant( - &self, - handle: &codex_core::cross_session::TurnHandle, - timeout: Duration, - ) -> Result { - let message = self.hub.await_first_assistant(handle, timeout).await?; - Ok(message) - } - - /// Await the first assistant message for this turn, but only time out after a - /// period of inactivity. Any event activity for this submission id resets the timer. - pub async fn await_first_assistant_idle( - &self, - handle: &codex_core::cross_session::TurnHandle, - idle_timeout: Duration, - role_label: Option<&str>, - ) -> Result { - // Subscribe to the session event stream to observe activity for this turn. - let mut events = self - .hub - .stream_events(handle.conversation_id()) - .map_err(|e| anyhow!(e))?; - - // We still rely on the hub's oneshot to capture the first assistant message. - let wait_first = self.hub.await_first_assistant(handle, idle_timeout); - tokio::pin!(wait_first); - - // Idle timer that resets on any event for this submission. - let idle = tokio::time::sleep(idle_timeout); - tokio::pin!(idle); - - let sub_id = handle.submission_id().to_string(); - - loop { - tokio::select! { - res = &mut wait_first => { - return res.map_err(|e| anyhow!(e)); - } - maybe_event = events.next() => { - let Some(ev) = maybe_event else { - // Event stream ended; if the assistant message has not arrived yet, - // treat as session closed. - bail!(codex_core::cross_session::CrossSessionError::SessionClosed); - }; - // Reset idle timer only for events emitted for our submission id. - if ev.event.id == sub_id { - if let Some(progress) = self.progress.as_ref() && let Some(role) = role_label { - progress.role_event(role, &ev.event.msg); - } - // If the session emits an error for this submission, surface it immediately - // rather than waiting for the idle timeout. - if let EventMsg::Error(err) = &ev.event.msg { - bail!(anyhow!(err.message.clone())); - } - idle.as_mut().reset(Instant::now() + idle_timeout); - } - } - _ = &mut idle => { - // No activity for the idle window — return a timeout error. - bail!(codex_core::cross_session::CrossSessionError::AwaitTimeout(idle_timeout)); - } - } - } + conversation_id: ConversationId, + ) -> Result { + self.hub.stream_events(conversation_id) } pub async fn call_role( @@ -780,11 +714,16 @@ impl InftyOrchestrator { timeout: Duration, final_output_json_schema: Option, ) -> Result { - let handle = self - .post_to_role(run_id, role, text, final_output_json_schema) - .await?; - self.await_first_assistant_idle(&handle, timeout, Some(role)) - .await + 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( @@ -795,23 +734,19 @@ impl InftyOrchestrator { timeout: Duration, final_output_json_schema: Option, ) -> Result { - let handle = self - .post_to_role( - run_id, - target_role, - assistant.message.message.clone(), - final_output_json_schema, - ) - .await?; - self.await_first_assistant_idle(&handle, timeout, Some(target_role)) - .await - } - - pub fn stream_events( - &self, - conversation_id: ConversationId, - ) -> Result { - self.hub.stream_events(conversation_id) + 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( @@ -822,9 +757,15 @@ impl InftyOrchestrator { store: &mut RunStore, cleanup: &mut Vec, ) -> Result { - let session = self - .spawn_role_session(run_id, run_path, role_config.clone()) - .await?; + let session = session::spawn_role( + Arc::clone(&self.hub), + &self.conversation_manager, + run_id, + run_path, + role_config.clone(), + prompts::ensure_instructions, + ) + .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() { @@ -840,49 +781,6 @@ impl InftyOrchestrator { 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, - run_path: &Path, - role_config: RoleConfig, - ) -> Result { - let RoleConfig { - role, mut config, .. - } = role_config; - config.cwd = run_path.to_path_buf(); - prompts::ensure_instructions(&role, &mut config); - let session = self - .conversation_manager - .new_conversation_with_cross_session( - config, - CrossSessionSpawnParams { - hub: Arc::clone(&self.hub), - run_id: Some(run_id.to_string()), - role: Some(role.clone()), - }, - ) - .await?; - Ok(RoleSession::from_new(role, session)) - } - - async fn resume_role_session( - &self, - run_id: &str, - run_path: &Path, - role_config: &RoleConfig, - store: &RunStore, ) -> Result { let metadata = store .role_metadata(&role_config.role) @@ -892,24 +790,22 @@ impl InftyOrchestrator { .as_ref() .ok_or_else(|| anyhow!("missing rollout path for role {}", role_config.role))?; - let mut config = role_config.config.clone(); - config.cwd = run_path.to_path_buf(); - prompts::ensure_instructions(&role_config.role, &mut config); - - let session = self - .conversation_manager - .resume_conversation_with_cross_session( - config, - rollout_path.clone(), - CrossSessionSpawnParams { - hub: Arc::clone(&self.hub), - run_id: Some(run_id.to_string()), - role: Some(role_config.role.clone()), - }, - ) - .await?; - - Ok(RoleSession::from_new(role_config.role.clone(), session)) + let session = session::resume_role( + Arc::clone(&self.hub), + &self.conversation_manager, + run_id, + run_path, + role_config, + rollout_path, + prompts::ensure_instructions, + ) + .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) } } diff --git a/codex-rs/codex-infty/src/session.rs b/codex-rs/codex-infty/src/session.rs new file mode 100644 index 0000000000..f8cae90968 --- /dev/null +++ b/codex-rs/codex-infty/src/session.rs @@ -0,0 +1,134 @@ +use std::path::Path; +use std::sync::Arc; +use std::time::Duration; + +use anyhow::Result; +use anyhow::anyhow; +use anyhow::bail; +use codex_core::ConversationManager; +use codex_core::CrossSessionSpawnParams; +use codex_core::config::Config; +use codex_core::cross_session::AssistantMessage; +use codex_core::cross_session::CrossSessionError; +use codex_core::cross_session::CrossSessionHub; +use codex_core::cross_session::PostUserTurnRequest; +use codex_core::cross_session::RoleOrId; +use codex_core::cross_session::TurnHandle; +use serde_json::Value; +use tokio::time::Instant; +use tokio_stream::StreamExt as _; + +use crate::progress::ProgressReporter; +use crate::types::RoleConfig; +use crate::types::RoleSession; + +pub async fn spawn_role( + hub: Arc, + manager: &ConversationManager, + run_id: &str, + run_path: &Path, + role_config: RoleConfig, + ensure_instructions: impl FnOnce(&str, &mut Config), +) -> Result { + let RoleConfig { + role, mut config, .. + } = role_config; + config.cwd = run_path.to_path_buf(); + ensure_instructions(&role, &mut config); + let session = manager + .new_conversation_with_cross_session( + config, + CrossSessionSpawnParams { + hub: Arc::clone(&hub), + run_id: Some(run_id.to_string()), + role: Some(role.clone()), + }, + ) + .await?; + Ok(RoleSession::from_new(role, session)) +} + +pub async fn resume_role( + hub: Arc, + manager: &ConversationManager, + run_id: &str, + run_path: &Path, + role_config: &RoleConfig, + rollout_path: &Path, + ensure_instructions: impl FnOnce(&str, &mut Config), +) -> Result { + let mut config = role_config.config.clone(); + config.cwd = run_path.to_path_buf(); + ensure_instructions(&role_config.role, &mut config); + let session = manager + .resume_conversation_with_cross_session( + config, + rollout_path.to_path_buf(), + CrossSessionSpawnParams { + hub: Arc::clone(&hub), + run_id: Some(run_id.to_string()), + role: Some(role_config.role.clone()), + }, + ) + .await?; + Ok(RoleSession::from_new(role_config.role.clone(), session)) +} + +pub async fn post_turn( + hub: &CrossSessionHub, + run_id: &str, + role: &str, + text: impl Into, + final_output_json_schema: Option, +) -> Result { + hub.post_user_turn(PostUserTurnRequest { + target: RoleOrId::RunRole { + run_id: run_id.to_string(), + role: role.to_string(), + }, + text: text.into(), + final_output_json_schema, + }) + .await +} + +pub async fn await_first_idle( + hub: &CrossSessionHub, + handle: &TurnHandle, + idle_timeout: Duration, + progress: Option<(&dyn ProgressReporter, &str)>, +) -> Result { + let mut events = hub.stream_events(handle.conversation_id())?; + let wait_first = hub.await_first_assistant(handle, idle_timeout); + tokio::pin!(wait_first); + + let idle = tokio::time::sleep(idle_timeout); + tokio::pin!(idle); + + let submission_id = handle.submission_id().to_string(); + + loop { + tokio::select! { + result = &mut wait_first => { + return result.map_err(|err| anyhow!(err)); + } + maybe_event = events.next() => { + let Some(event) = maybe_event else { + bail!(CrossSessionError::SessionClosed); + }; + if event.event.id == submission_id { + if let Some((reporter, role)) = progress { + reporter.role_event(role, &event.event.msg); + } + if let codex_core::protocol::EventMsg::Error(err) = &event.event.msg { + bail!(anyhow!(err.message.clone())); + } + idle.as_mut().reset(Instant::now() + idle_timeout); + } + } + _ = &mut idle => { + bail!(CrossSessionError::AwaitTimeout(idle_timeout)); + } + } + } +} diff --git a/codex-rs/infty2.md b/codex-rs/infty2.md new file mode 100644 index 0000000000..ee9e9645ed --- /dev/null +++ b/codex-rs/infty2.md @@ -0,0 +1,325 @@ +# Infty v2 - Minimal Cross-Session Loop + +Goal: collapse the orchestration to three composable primitives while preserving the existing flow. + +- spawn: create a role session with base instructions + config +- await: wait for the assistant message that ends the user turn +- forward: inject an assistant message as a user message in another session + +The rest of the orchestrator becomes a tiny router that parses the Solver's signal and calls these helpers. + +--- + +## Design Overview + +We build a thin, reusable facade over `codex-core`'s cross-session utilities. This facade is role- and run-aware so callers don't need to handle `ConversationId` bookkeeping. + +Key types from `codex-core::cross_session` that we lean on: + +- `CrossSessionHub` - registers sessions and routes messages across them +- `PostUserTurnRequest` - payload to submit text to a session +- `TurnHandle` - handle for a turn (used to await the assistant) +- `AssistantMessage` - the first assistant message for a turn +- `SessionEventStream` - event stream for activity/idle timeouts + +In `codex-infty`, we expose tiny helpers that wrap these primitives in a role-centric API. + +--- + +## Minimal API (Facade) + +Proposed module: `codex-infty/src/session.rs` (or fold into `orchestrator.rs` if preferred). Names shown here as free functions; methods on a small struct are also fine. + +```rust +use std::sync::Arc; +use std::time::Duration; +use anyhow::Result; +use serde_json::Value; +use codex_core::{ConversationManager, NewConversation}; +use codex_core::config::Config; +use codex_core::cross_session::{ + CrossSessionHub, PostUserTurnRequest, RoleOrId, TurnHandle, AssistantMessage, +}; +use codex_protocol::ConversationId; + +/// Opaque role session reference used by the orchestrator. +#[derive(Clone)] +pub struct RoleSession { + pub role: String, + pub conversation_id: ConversationId, + pub conversation: Arc, +} + +/// 1) Spawn a role session with base instructions applied. +pub async fn spawn( + hub: Arc, + manager: &ConversationManager, + run_id: &str, + role: &str, + mut config: Config, + rollout_dir: impl Into, + ensure_instructions: impl FnOnce(&str, &mut Config), +) -> Result { + config.cwd = rollout_dir.into(); + ensure_instructions(role, &mut config); + let created: NewConversation = manager + .new_conversation_with_cross_session( + config, + codex_core::CrossSessionSpawnParams { + hub: Arc::clone(&hub), + run_id: Some(run_id.to_string()), + role: Some(role.to_string()), + }, + ) + .await?; + Ok(RoleSession { + role: role.to_string(), + conversation_id: created.conversation_id, + conversation: created.conversation, + }) +} + +/// 2a) Post a user turn to a role. +pub async fn post( + hub: &CrossSessionHub, + run_id: &str, + role: &str, + text: impl Into, + final_output_json_schema: Option, +) -> Result { + hub.post_user_turn(PostUserTurnRequest { + target: RoleOrId::RunRole { run_id: run_id.to_string(), role: role.to_string() }, + text: text.into(), + final_output_json_schema, + }).await +} + +/// 2b) Await the first assistant message for this turn. +pub async fn await_first( + hub: &CrossSessionHub, + handle: &TurnHandle, + timeout: Duration, +) -> Result { + hub.await_first_assistant(handle, timeout).await +} + +/// 2c) Await with idle timeout that resets on activity for this submission id. +/// (Move the existing codex-infty implementation here verbatim.) +``` + +```rust +pub async fn await_first_idle( + hub: &CrossSessionHub, + handle: &TurnHandle, + idle_timeout: Duration, +) -> Result { + use anyhow::{anyhow, bail}; + use codex_core::protocol::EventMsg; + use tokio::time::Instant; + use tokio_stream::StreamExt as _; + + let mut events = hub.stream_events(handle.conversation_id())?; + let wait_first = hub.await_first_assistant(handle, idle_timeout); + tokio::pin!(wait_first); + + let idle = tokio::time::sleep(idle_timeout); + tokio::pin!(idle); + + let sub_id = handle.submission_id().to_string(); + + loop { + tokio::select! { + res = &mut wait_first => { return res.map_err(|e| anyhow!(e)); } + maybe_event = events.next() => { + let Some(ev) = maybe_event else { bail!(codex_core::cross_session::CrossSessionError::SessionClosed); }; + if ev.event.id == sub_id { + if let EventMsg::Error(err) = &ev.event.msg { bail!(anyhow!(err.message.clone())); } + idle.as_mut().reset(Instant::now() + idle_timeout); + } + } + _ = &mut idle => { bail!(codex_core::cross_session::CrossSessionError::AwaitTimeout(idle_timeout)); } + } + } +} +``` + +```rust +/// 3) Forward an assistant's content as a user message to another role. +pub async fn forward_assistant( + hub: &CrossSessionHub, + run_id: &str, + target_role: &str, + assistant: &AssistantMessage, + timeout: Duration, + final_output_json_schema: Option, +) -> Result { + let handle = post( + hub, + run_id, + target_role, + assistant.message.message.clone(), + final_output_json_schema, + ).await?; + Ok(await_first(hub, &handle, timeout).await?) +} + +/// Convenience: do both post + await in one call. +pub async fn call( + hub: &CrossSessionHub, + run_id: &str, + role: &str, + text: impl Into, + timeout: Duration, + final_output_json_schema: Option, +) -> Result { + let handle = post(hub, run_id, role, text, final_output_json_schema).await?; + Ok(await_first(hub, &handle, timeout).await?) +} +``` + +Notes: +- `await_first_idle` is the ergonomic default in Infty because it handles streaming with activity-based resets. +- The facade leaves JSON schema optional and role-addressing consistent with `RunRole { run_id, role }`. + +--- + +## Orchestrator Main Loop Becomes Tiny + +Once the three operations exist, the loop reduces to routing: + +```rust +// Pseudocode using the facade +let mut solver_ev = hub.stream_events(sessions.solver.conversation_id)?; + +if let Some(objective) = options.objective.as_deref() { + post(&hub, &run_id, &sessions.solver.role, objective, Some(solver_signal_schema())).await?; +} + +loop { + let ev = solver_ev.next().await.ok_or_else(|| anyhow::anyhow!("solver closed"))?; + if let EventMsg::AgentMessage(agent) = &ev.event.msg { + if let Some(signal) = parse_solver_signal(&agent.message) { + match signal { + SolverSignal::DirectionRequest { prompt: Some(p) } => { + let req = serde_json::to_string(&DirectionRequestPayload { + kind: "direction_request", + prompt: &p, + objective: options.objective.as_deref(), + })?; + let directive = call(&hub, &run_id, &sessions.director.role, req, options.director_timeout, Some(directive_response_schema())).await?; + let _ = forward_assistant(&hub, &run_id, &sessions.solver.role, &directive, std::time::Duration::from_secs(5), Some(solver_signal_schema())).await?; + } + SolverSignal::VerificationRequest { claim_path: Some(path), notes } => { + let req = serde_json::to_string(&VerificationRequestPayload { + kind: "verification_request", + claim_path: &path, + notes: notes.as_deref(), + objective: options.objective.as_deref(), + })?; + let mut verdicts = Vec::new(); + for v in &sessions.verifiers { + let verdict = call(&hub, &run_id, &v.role, &req, options.verifier_timeout, Some(verifier_verdict_schema())).await?; + verdicts.push((v.role.clone(), parse_json_struct::(&verdict.message.message)?)); + } + let summary = aggregate_verdicts(verdicts); + let _ = post(&hub, &run_id, &sessions.solver.role, serde_json::to_string(&summary)?, Some(solver_signal_schema())).await?; + } + SolverSignal::FinalDelivery { deliverable_path: Some(path), summary } => { + let deliverable = resolve_deliverable_path(sessions.store.path(), &path)?; + return Ok(RunOutcome { run_id, deliverable_path: deliverable, summary, raw_message: agent.message.clone() }); + } + _ => {} + } + } + } +} +``` + +Everything above already exists in `codex-infty` today; the facade simply standardizes the small operations so the loop reads linearly. + +--- + +## Implementation Steps + +1) Extract helpers +- Add `session.rs` with `spawn`, `post`, `await_first`, `await_first_idle`, `forward_assistant`, `call`. +- Move the existing `await_first_assistant_idle` body from `orchestrator.rs` to this module (exported). +- Re-export from `lib.rs` if desirable for external callers. + +2) Adopt helpers in `orchestrator.rs` +- Replace `post_to_role`, `await_first_assistant`, `relay_assistant_to_role`, and `call_role` with the facade functions. +- Keep signal parsing and run-store logic; delete glue code that becomes redundant. + +3) Keep role spawn/resume minimal +- Inline `spawn_role_session` and `resume_role_session` to call `session::spawn(...)` with `prompts::ensure_instructions`. +- Preserve persistence of rollout/config paths via `RunStore`. + +4) Preserve JSON schema guarantees +- Pass schemas through `post`/`call`/`forward_assistant` exactly as today: + - Solver outbound: `solver_signal_schema()` + - Director outbound: `directive_response_schema()` + - Verifier outbound: `verifier_verdict_schema()` + - Finalization: `final_delivery_schema()` for the last probe + +5) Progress reporting stays orthogonal +- Where the orchestrator previously called `progress.*`, keep those calls around the facade usage (no change to the trait). + +6) Tests and docs +- Unit-test the facade with a tiny harness that posts to a mock/run role and awaits the first assistant. +- Update README examples to use `call` and `forward_assistant` for clarity. + +--- + +## Snippets to Drop In + +- Posting user input and awaiting the assistant with idle timeout: + +```rust +let handle = session::post(hub, &run_id, &role, user_text, schema).await?; +let assistant = session::await_first_idle(hub, &handle, std::time::Duration::from_secs(120)).await?; +``` + +- Forwarding an assistant to another role: + +```rust +let reply = session::forward_assistant(hub, &run_id, &target_role, &assistant, std::time::Duration::from_secs(60), target_schema).await?; +``` + +- Spawning a session with base instructions: + +```rust +let solver = session::spawn( + Arc::clone(&hub), + &conversation_manager, + &run_id, + "solver", + solver_cfg.clone(), + run_path, // becomes cfg.cwd + |role, cfg| prompts::ensure_instructions(role, cfg), +).await?; +``` + +--- + +## Why This Simplifies Things + +- One mental model: "post -> await -> forward" across roles. +- Orchestrator logic is a small, readable router. +- Cross-session reliability remains in one place (the hub). +- Tests become surgical: assert an assistant message is forwarded or a schema is respected. + +--- + +## Backward Compatibility + +- All current public behavior stays the same. +- `InftyOrchestrator` public methods keep signatures; they are implemented in terms of the facade. +- No changes to `codex-core` types or wire protocol. + +--- + +## Optional Follow-Ups + +- Consider upstreaming `await_first_idle` into `codex-core` so others can reuse it outside Infty. +- Add typed wrappers for JSON payloads (newtypes) to reduce `serde_json::Value` usage at call sites. +- Provide a tiny `SessionRouter` example crate to demonstrate building custom flows with these primitives.