diff --git a/codex-rs/cli/src/infty/progress.rs b/codex-rs/cli/src/infty/progress.rs index 8f68a23369..1c5f0af341 100644 --- a/codex-rs/cli/src/infty/progress.rs +++ b/codex-rs/cli/src/infty/progress.rs @@ -61,7 +61,7 @@ impl ProgressReporter for TerminalProgressReporter { fn solver_event(&self, event: &EventMsg) { match serde_json::to_string_pretty(event) { Ok(json) => { - tracing::trace!("[solver:event]\n{json}"); + tracing::debug!("[solver:event]\n{json}"); } Err(err) => { tracing::warn!("[solver:event] (failed to serialize: {err}) {event:?}"); @@ -69,6 +69,17 @@ impl ProgressReporter for TerminalProgressReporter { } } + fn role_event(&self, role: &str, event: &EventMsg) { + match serde_json::to_string_pretty(event) { + Ok(json) => { + tracing::debug!("[{role}:event]\n{json}"); + } + Err(err) => { + tracing::warn!("[{role}:event] (failed to serialize: {err}) {event:?}"); + } + } + } + fn solver_agent_message(&self, agent_msg: &AgentMessageEvent) { let prefix = if self.color_enabled { format!("{}", "[solver]".magenta().bold()) diff --git a/codex-rs/codex-infty/README.md b/codex-rs/codex-infty/README.md new file mode 100644 index 0000000000..ec8f7a3e27 --- /dev/null +++ b/codex-rs/codex-infty/README.md @@ -0,0 +1,229 @@ +# Codex Infty + +Codex Infty is a small orchestration layer that coordinates multiple Codex roles (Solver, Director, Verifier(s)) to drive longer, multi‑step objectives with minimal human intervention. It provides: + +- A run orchestrator that routes messages between roles and advances the workflow. +- A durable run store on disk with metadata and standard subfolders. +- Default role prompts for Solver/Director/Verifier. +- A lightweight progress reporting hook for UIs/CLIs. + +The crate is designed to be embedded (via the library API) and also powers the `codex infty` CLI commands. + +## High‑Level Flow + +``` +objective → Solver + Solver → direction_request → Director → directive → Solver + Solver → verification_request → Verifier(s) → aggregated summary → Solver + … (iterate) … + Solver → final_delivery → Orchestrator returns RunOutcome +``` + +- The Solver always speaks structured JSON. The orchestrator parses those messages and decides the next hop. +- The Director provides crisp guidance (also JSON) that is forwarded back to the Solver. +- One or more Verifiers assess claims and return verdicts; the orchestrator aggregates results and reports a summary to the Solver. +- On final_delivery, the orchestrator resolves and validates the deliverable path and returns the `RunOutcome`. + +## Directory Layout (Run Store) + +When a run is created, a directory is initialized with this structure: + +``` +// + artifacts/ # long‑lived artifacts produced by the Solver + memory/ # durable notes, claims, context + index/ # indexes and caches + deliverable/ # final output(s) assembled by the Solver + run.json # run metadata (id, timestamps, roles) +``` + +See: `codex-infty/src/run_store.rs`. + +- The orchestrator persists rollout paths and optional config paths for each role into `run.json`. +- Metadata timestamps are updated on significant events (role spawns, handoffs, final delivery). +- Final deliverables must remain within the run directory. Paths are canonicalized and validated. + +## Roles and Prompts + +Default base instructions are injected per role if the provided `Config` has none: + +- Solver: `codex-infty/src/prompts/solver.md` +- Director: `codex-infty/src/prompts/director.md` +- Verifier: `codex-infty/src/prompts/verifier.md` + +You can provide your own instructions by pre‑populating `Config.base_instructions`. + +## Solver Signal Contract + +The Solver communicates intent using JSON messages (possibly wrapped in a fenced block). The orchestrator accepts three shapes: + +- Direction request (sent to Director): + +```json +{"type":"direction_request","prompt":""} +``` + +- Verification request (sent to Verifier(s)): + +```json +{"type":"verification_request","claim_path":"memory/claims/.json","notes":null} +``` + +- Final delivery (completes the run): + +```json +{"type":"final_delivery","deliverable_path":"deliverable/summary.txt","summary":""} +``` + +JSON may be fenced as ```json … ```; the orchestrator will strip the fence. + +## Key Types and Modules + +- Orchestrator: `codex-infty/src/orchestrator.rs` + - `InftyOrchestrator`: spawns/resumes role sessions, drives the event loop, and routes signals. + - `execute_new_run` / `execute_existing_run`: one‑shot helpers that spawn/resume and then drive. + - `spawn_run` / `resume_run`: set up sessions and the run store. + - `call_role`, `relay_assistant_to_role`, `post_to_role`, `await_first_assistant`, `stream_events`: utilities when integrating custom flows. + +- Run store: `codex-infty/src/run_store.rs` + - `RunStore`, `RunMetadata`, `RoleMetadata`: metadata and persistence helpers. + +- Types: `codex-infty/src/types.rs` + - `RoleConfig`: wraps a `Config` and sets sensible defaults for autonomous flows (no approvals, full sandbox access). Also used to persist optional config paths. + - `RunParams`, `ResumeParams`: input to spawn/resume runs. + - `RunExecutionOptions`: per‑run options (objective, timeouts). + - `RunOutcome`: returned on successful final delivery. + +- Signals: `codex-infty/src/signals.rs` + - DTOs for director responses and verifier verdicts, and the aggregated summary type. + +- Progress: `codex-infty/src/progress.rs` + - `ProgressReporter` trait: hook for UIs/CLIs to observe solver/director/verifier activity. + +## Orchestrator Workflow (Details) + +1. Spawn or resume role sessions (Solver, Director, and zero or more Verifiers). Default prompts are applied if the role’s `Config` has no base instructions. +2. Optionally post an `objective` to the Solver. The progress reporter is notified and the orchestrator waits for the first Solver signal. +3. On `direction_request`: + - Post a structured request to the Director and await the first assistant message. + - Parse it into a `DirectiveResponse` and forward the normalized JSON to the Solver. +4. On `verification_request`: + - Send structured requests to each Verifier and await each first assistant message. + - Aggregate verdicts into an `AggregatedVerifierVerdict` and post the summary back to the Solver. +5. On `final_delivery`: + - Canonicalize and validate that `deliverable_path` stays within the run directory. + - Notify the progress reporter, touch the run store, and return `RunOutcome`. + +Note: The orchestrator does not re‑run verification after `final_delivery`. Verification is performed whenever the Solver issues a `verification_request` during the run. + +## Library Usage + +```rust +use std::sync::Arc; +use codex_core::{CodexAuth, config::Config}; +use codex_infty::{InftyOrchestrator, RoleConfig, RunParams, RunExecutionOptions}; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + // 1) Load or build a Config for each role + let solver_cfg: Config = load_config(); + let mut director_cfg = solver_cfg.clone(); + director_cfg.model = "o4-mini".into(); + + // 2) Build role configs + let solver = RoleConfig::new("solver", solver_cfg.clone()); + let director = RoleConfig::new("director", director_cfg); + let verifiers = vec![RoleConfig::new("verifier-alpha", solver_cfg.clone())]; + + // 3) Create an orchestrator (using default runs root) + let auth = CodexAuth::from_api_key("sk-…"); + let orchestrator = InftyOrchestrator::new(auth)?; + + // 4) Execute a new run with an objective + let params = RunParams { + run_id: "my-run".into(), + run_root: None, // use default ~/.codex/infty/ + solver, + director, + verifiers, + }; + let mut opts = RunExecutionOptions::default(); + opts.objective = Some("Implement feature X".into()); + + let outcome = orchestrator.execute_new_run(params, opts).await?; + println!("deliverable: {}", outcome.deliverable_path.display()); + Ok(()) +} +# fn load_config() -> codex_core::config::Config { codex_core::config::Config::default() } +``` + +Resuming an existing run: + +```rust +use codex_infty::{InftyOrchestrator, ResumeParams, RoleConfig}; + +async fn resume_example(orchestrator: &InftyOrchestrator) -> anyhow::Result<()> { + let solver = RoleConfig::new("solver", load_config()); + let director = RoleConfig::new("director", load_config()); + let verifiers = vec![]; + + let resume = ResumeParams { + run_path: std::path::PathBuf::from("/path/to/run"), + solver, + director, + verifiers, + }; + let outcome = orchestrator.execute_existing_run(resume, Default::default()).await?; + println!("{}", outcome.run_id); + Ok(()) +} +# fn load_config() -> codex_core::config::Config { codex_core::config::Config::default() } +``` + +## CLI Quickstart + +The CLI (`codex`) exposes Infty helpers under the `infty` subcommand. Examples: + +```bash +# Create a run and immediately drive toward completion +codex infty create --run-id demo --objective "Build and test feature" + +# Inspect runs +codex infty list +codex infty show demo + +# Send a one-off message to a role in a running/resumable run +codex infty drive demo solver "Summarize progress" +``` + +Flags allow customizing the Director’s model and reasoning effort; see `codex infty create --help`. + +## Progress Reporting + +Integrate your UI by implementing `ProgressReporter` and attaching it with `InftyOrchestrator::with_progress(...)`. You’ll receive callbacks on key milestones (objective posted, solver messages, director response, verification summaries, final delivery, etc.). + +## Safety and Guardrails + +- `RoleConfig::new` sets `SandboxPolicy::DangerFullAccess` and `AskForApproval::Never` to support autonomous flows. Adjust if your environment requires stricter policies. +- Deliverable paths are validated to stay inside the run directory and are fully canonicalized. +- JSON payloads are schema‑checked where applicable (e.g., solver signals and final delivery shape). + +## Tests + +Run the crate’s tests: + +```bash +cargo test -p codex-infty +``` + +Many tests rely on mocked SSE streams and will auto‑skip in sandboxes where network is disabled. + +## When to Use This Crate + +Use `codex-infty` when you want a minimal, pragmatic multi‑role loop with: + +- Clear role separation and routing. +- Durable, restart‑resilient state on disk. +- Simple integration points (progress hooks and helper APIs). + +It’s intentionally small and focused so it can be embedded into larger tools or extended to meet your workflows. diff --git a/codex-rs/codex-infty/src/orchestrator.rs b/codex-rs/codex-infty/src/orchestrator.rs index 59919ba0d6..b6b4b5480d 100644 --- a/codex-rs/codex-infty/src/orchestrator.rs +++ b/codex-rs/codex-infty/src/orchestrator.rs @@ -28,6 +28,7 @@ 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; @@ -491,7 +492,7 @@ impl InftyOrchestrator { ) .await?; let directive = self - .await_first_assistant(&handle, options.director_timeout) + .await_first_assistant_idle(&handle, options.director_timeout, Some("director")) .await?; let directive_payload: DirectiveResponse = parse_json_struct(&directive.message.message) .context("director response was not valid directive JSON")?; @@ -576,7 +577,7 @@ impl InftyOrchestrator { ) .await?; let _ = self - .await_first_assistant(&handle, Duration::from_secs(5)) + .await_first_assistant_idle(&handle, Duration::from_secs(5), None) .await?; Ok(()) } @@ -611,7 +612,7 @@ impl InftyOrchestrator { ) .await?; let response = self - .await_first_assistant(&handle, options.verifier_timeout) + .await_first_assistant_idle(&handle, options.verifier_timeout, Some(&verifier.role)) .await?; let verdict: VerifierVerdict = parse_json_struct(&response.message.message) .with_context(|| { @@ -712,6 +713,59 @@ impl InftyOrchestrator { 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); + } + 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)); + } + } + } + } + pub async fn call_role( &self, run_id: &str, @@ -723,7 +777,8 @@ impl InftyOrchestrator { let handle = self .post_to_role(run_id, role, text, final_output_json_schema) .await?; - self.await_first_assistant(&handle, timeout).await + self.await_first_assistant_idle(&handle, timeout, Some(role)) + .await } pub async fn relay_assistant_to_role( @@ -742,7 +797,8 @@ impl InftyOrchestrator { final_output_json_schema, ) .await?; - self.await_first_assistant(&handle, timeout).await + self.await_first_assistant_idle(&handle, timeout, Some(target_role)) + .await } pub fn stream_events( diff --git a/codex-rs/codex-infty/src/progress.rs b/codex-rs/codex-infty/src/progress.rs index 117ca98bc5..143a1ead98 100644 --- a/codex-rs/codex-infty/src/progress.rs +++ b/codex-rs/codex-infty/src/progress.rs @@ -11,6 +11,7 @@ pub trait ProgressReporter: Send + Sync { fn objective_posted(&self, _objective: &str) {} fn waiting_for_solver(&self) {} fn solver_event(&self, _event: &EventMsg) {} + fn role_event(&self, _role: &str, _event: &EventMsg) {} fn solver_agent_message(&self, _message: &AgentMessageEvent) {} fn direction_request(&self, _prompt: &str) {} fn director_response(&self, _directive: &DirectiveResponse) {}