From 8245a4f53b1b9e363dc723a79d175e670e135ca6 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Mon, 13 Oct 2025 17:56:44 +0100 Subject: [PATCH] V1 --- codex-rs/Cargo.lock | 24 + codex-rs/Cargo.toml | 1 + codex-rs/cli/Cargo.toml | 3 + codex-rs/cli/src/infty.rs | 450 ++++++++++ codex-rs/cli/src/main.rs | 13 + codex-rs/codex-infty/Cargo.toml | 22 + codex-rs/codex-infty/src/lib.rs | 977 +++++++++++++++++++++ codex-rs/codex-infty/src/prompts.rs | 147 ++++ codex-rs/codex-infty/src/run_store.rs | 211 +++++ codex-rs/codex-infty/tests/orchestrator.rs | 358 ++++++++ codex-rs/core/Cargo.toml | 1 + codex-rs/core/src/conversation_manager.rs | 178 +++- codex-rs/core/src/cross_session.rs | 607 +++++++++++++ codex-rs/core/src/lib.rs | 2 + codex-rs/core/tests/suite/cross_session.rs | 171 ++++ codex-rs/core/tests/suite/mod.rs | 1 + codex-rs/infty.md | 189 ++++ 17 files changed, 3334 insertions(+), 21 deletions(-) create mode 100644 codex-rs/cli/src/infty.rs create mode 100644 codex-rs/codex-infty/Cargo.toml create mode 100644 codex-rs/codex-infty/src/lib.rs create mode 100644 codex-rs/codex-infty/src/prompts.rs create mode 100644 codex-rs/codex-infty/src/run_store.rs create mode 100644 codex-rs/codex-infty/tests/orchestrator.rs create mode 100644 codex-rs/core/src/cross_session.rs create mode 100644 codex-rs/core/tests/suite/cross_session.rs create mode 100644 codex-rs/infty.md diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 0c12880c94..989709fdf8 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -941,6 +941,7 @@ dependencies = [ "anyhow", "assert_cmd", "assert_matches", + "chrono", "clap", "clap_complete", "codex-app-server", @@ -951,6 +952,7 @@ dependencies = [ "codex-common", "codex-core", "codex-exec", + "codex-infty", "codex-login", "codex-mcp-server", "codex-process-hardening", @@ -963,6 +965,7 @@ dependencies = [ "owo-colors", "predicates", "pretty_assertions", + "serde", "serde_json", "supports-color", "tempfile", @@ -1076,6 +1079,7 @@ dependencies = [ "thiserror 2.0.16", "time", "tokio", + "tokio-stream", "tokio-test", "tokio-util", "toml", @@ -1177,6 +1181,25 @@ dependencies = [ "walkdir", ] +[[package]] +name = "codex-infty" +version = "0.0.0" +dependencies = [ + "anyhow", + "chrono", + "codex-core", + "codex-protocol", + "core_test_support", + "dirs", + "serde", + "serde_json", + "tempfile", + "tokio", + "tokio-stream", + "tracing", + "wiremock", +] + [[package]] name = "codex-linux-sandbox" version = "0.0.0" @@ -5998,6 +6021,7 @@ dependencies = [ "futures-core", "pin-project-lite", "tokio", + "tokio-util", ] [[package]] diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index ca40b1a536..4b324c4b54 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -6,6 +6,7 @@ members = [ "app-server-protocol", "apply-patch", "arg0", + "codex-infty", "codex-backend-openapi-models", "cloud-tasks", "cloud-tasks-client", diff --git a/codex-rs/cli/Cargo.toml b/codex-rs/cli/Cargo.toml index 453ab80771..ca76472d46 100644 --- a/codex-rs/cli/Cargo.toml +++ b/codex-rs/cli/Cargo.toml @@ -45,6 +45,9 @@ tokio = { workspace = true, features = [ "rt-multi-thread", "signal", ] } +codex-infty = { path = "../codex-infty" } +chrono = { workspace = true } +serde = { workspace = true, features = ["derive"] } [dev-dependencies] assert_matches = { workspace = true } diff --git a/codex-rs/cli/src/infty.rs b/codex-rs/cli/src/infty.rs new file mode 100644 index 0000000000..befe4e9533 --- /dev/null +++ b/codex-rs/cli/src/infty.rs @@ -0,0 +1,450 @@ +use std::fs; +use std::io; +use std::path::Path; +use std::path::PathBuf; +use std::time::Duration; + +use anyhow::Context; +use anyhow::Result; +use anyhow::anyhow; +use anyhow::bail; +use chrono::SecondsFormat; +use chrono::Utc; +use clap::Parser; +use clap::Subcommand; +use codex_common::CliConfigOverrides; +use codex_core::CodexAuth; +use codex_core::auth::read_codex_api_key_from_env; +use codex_core::auth::read_openai_api_key_from_env; +use codex_core::config::Config; +use codex_core::config::ConfigOverrides; +use codex_infty::InftyOrchestrator; +use codex_infty::ResumeParams; +use codex_infty::RoleConfig; +use codex_infty::RunExecutionOptions; +use codex_infty::RunParams; +use codex_infty::RunStore; +use serde::Serialize; + +const DEFAULT_TIMEOUT_SECS: u64 = 60; + +#[derive(Debug, Parser)] +pub struct InftyCli { + #[clap(flatten)] + pub config_overrides: CliConfigOverrides, + + /// Override the default runs root (`~/.codex/infty`). + #[arg(long = "runs-root", value_name = "DIR")] + pub runs_root: Option, + + #[command(subcommand)] + command: InftyCommand, +} + +#[derive(Debug, Subcommand)] +enum InftyCommand { + /// Create a new run store and spawn solver/director sessions. + Create(CreateArgs), + + /// List stored runs. + List(ListArgs), + + /// Show metadata for a stored run. + Show(ShowArgs), + + /// Send a message to a role within a run and print the first reply. + Drive(DriveArgs), +} + +#[derive(Debug, Parser)] +struct CreateArgs { + /// Explicit run id. If omitted, a timestamp-based id is generated. + #[arg(long = "run-id", value_name = "RUN_ID")] + run_id: Option, + + /// Optional objective to send to the solver immediately after creation. + #[arg(long)] + objective: Option, + + /// Timeout in seconds when waiting for the solver reply to --objective. + #[arg(long = "timeout-secs", default_value_t = DEFAULT_TIMEOUT_SECS)] + timeout_secs: u64, +} + +#[derive(Debug, Parser)] +struct ListArgs { + /// Emit JSON describing the stored runs. + #[arg(long)] + json: bool, +} + +#[derive(Debug, Parser)] +struct ShowArgs { + /// Run id to display. + run_id: String, + + /// Emit JSON metadata instead of human-readable text. + #[arg(long)] + json: bool, +} + +#[derive(Debug, Parser)] +struct DriveArgs { + /// Run id to resume. + run_id: String, + + /// Role to address (e.g. solver, director). + role: String, + + /// Message to send to the role. + message: String, + + /// Timeout in seconds to await the first assistant message. + #[arg(long = "timeout-secs", default_value_t = DEFAULT_TIMEOUT_SECS)] + timeout_secs: u64, +} + +#[derive(Debug, Serialize)] +struct RunSummary { + run_id: String, + path: String, + created_at: String, + updated_at: String, + roles: Vec, +} + +impl InftyCli { + pub async fn run(self) -> Result<()> { + let InftyCli { + config_overrides, + runs_root, + command, + } = self; + + match command { + InftyCommand::Create(args) => { + run_create(config_overrides, runs_root, args).await?; + } + InftyCommand::List(args) => run_list(runs_root, args)?, + InftyCommand::Show(args) => run_show(runs_root, args)?, + InftyCommand::Drive(args) => { + run_drive(config_overrides, runs_root, args).await?; + } + } + + Ok(()) + } +} + +async fn run_create( + config_overrides: CliConfigOverrides, + runs_root_override: Option, + args: CreateArgs, +) -> Result<()> { + let config = load_config(config_overrides).await?; + let auth = load_auth(&config)?; + let runs_root = resolve_runs_root(runs_root_override)?; + + let mut run_id = if let Some(id) = args.run_id { + id + } else { + generate_run_id() + }; + run_id = run_id.trim().to_string(); + validate_run_id(&run_id)?; + + let run_path = runs_root.join(&run_id); + if run_path.exists() { + bail!("run {run_id} already exists at {}", run_path.display()); + } + + let orchestrator = InftyOrchestrator::with_runs_root(auth, runs_root.clone()); + let run_params = RunParams { + run_id: run_id.clone(), + run_root: Some(run_path.clone()), + solver: RoleConfig::new("solver", config.clone()), + director: RoleConfig::new("director", config.clone()), + verifiers: Vec::new(), + }; + + if let Some(objective) = args.objective { + let mut options = RunExecutionOptions::default(); + options.objective = Some(objective); + let timeout = Duration::from_secs(args.timeout_secs); + options.director_timeout = timeout; + options.verifier_timeout = timeout; + + let outcome = orchestrator + .execute_new_run(run_params, options) + .await + .with_context(|| format!("failed to execute run {run_id}"))?; + println!( + "Run {run_id} completed. Deliverable: {}", + outcome.deliverable_path.display() + ); + if let Some(summary) = outcome.summary { + println!("Summary: {summary}"); + } + } else { + let sessions = orchestrator + .spawn_run(run_params) + .await + .with_context(|| format!("failed to create run {run_id}"))?; + + println!( + "Created run {run_id} at {}", + sessions.store.path().display() + ); + } + + Ok(()) +} + +fn run_list(runs_root_override: Option, args: ListArgs) -> Result<()> { + let runs_root = resolve_runs_root(runs_root_override)?; + let listings = collect_run_summaries(&runs_root)?; + + if args.json { + println!("{}", serde_json::to_string_pretty(&listings)?); + return Ok(()); + } + + if listings.is_empty() { + println!("No runs found under {}", runs_root.display()); + return Ok(()); + } + + println!("Runs in {}", runs_root.display()); + for summary in listings { + println!( + "{}\t{}\t{}", + summary.run_id, summary.updated_at, summary.path + ); + } + + Ok(()) +} + +fn run_show(runs_root_override: Option, args: ShowArgs) -> Result<()> { + validate_run_id(&args.run_id)?; + let runs_root = resolve_runs_root(runs_root_override)?; + let run_path = runs_root.join(&args.run_id); + let store = + RunStore::load(&run_path).with_context(|| format!("failed to load run {}", args.run_id))?; + let metadata = store.metadata(); + + let summary = RunSummary { + run_id: metadata.run_id.clone(), + path: run_path.display().to_string(), + created_at: metadata + .created_at + .to_rfc3339_opts(SecondsFormat::Secs, true), + updated_at: metadata + .updated_at + .to_rfc3339_opts(SecondsFormat::Secs, true), + roles: metadata + .roles + .iter() + .map(|role| role.role.clone()) + .collect(), + }; + + if args.json { + println!("{}", serde_json::to_string_pretty(&summary)?); + return Ok(()); + } + + println!("Run: {}", summary.run_id); + println!("Path: {}", summary.path); + println!("Created: {}", summary.created_at); + println!("Updated: {}", summary.updated_at); + println!("Roles: {}", summary.roles.join(", ")); + + Ok(()) +} + +async fn run_drive( + config_overrides: CliConfigOverrides, + runs_root_override: Option, + args: DriveArgs, +) -> Result<()> { + validate_run_id(&args.run_id)?; + let config = load_config(config_overrides).await?; + let auth = load_auth(&config)?; + let runs_root = resolve_runs_root(runs_root_override)?; + let run_path = runs_root.join(&args.run_id); + let store = + RunStore::load(&run_path).with_context(|| format!("failed to load run {}", args.run_id))?; + + let solver_role = store + .role_metadata("solver") + .ok_or_else(|| anyhow!("run {} is missing solver role", args.run_id))?; + let director_role = store + .role_metadata("director") + .ok_or_else(|| anyhow!("run {} is missing director role", args.run_id))?; + + let verifiers: Vec<_> = store + .metadata() + .roles + .iter() + .filter(|role| role.role != solver_role.role && role.role != director_role.role) + .map(|role| RoleConfig::new(role.role.clone(), config.clone())) + .collect(); + + let orchestrator = InftyOrchestrator::with_runs_root(auth, runs_root); + let sessions = orchestrator + .resume_run(ResumeParams { + run_path: run_path.clone(), + solver: RoleConfig::new(solver_role.role.clone(), config.clone()), + director: RoleConfig::new(director_role.role.clone(), config.clone()), + verifiers, + }) + .await + .with_context(|| format!("failed to resume run {}", args.run_id))?; + + let timeout = Duration::from_secs(args.timeout_secs); + let reply = orchestrator + .call_role(&sessions.run_id, &args.role, args.message, timeout, None) + .await + .with_context(|| { + format!( + "failed to deliver message to role {} in run {}", + args.role, sessions.run_id + ) + })?; + + println!("{}", reply.message.message); + Ok(()) +} + +fn validate_run_id(run_id: &str) -> Result<()> { + if run_id.is_empty() { + bail!("run id must not be empty"); + } + if run_id.starts_with('.') || run_id.ends_with('.') { + bail!("run id must not begin or end with '.'"); + } + if run_id + .chars() + .any(|c| !(c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.'))) + { + bail!("run id may only contain ASCII alphanumerics, '-', '_', or '.'"); + } + Ok(()) +} + +fn generate_run_id() -> String { + let timestamp = Utc::now().format("run-%Y%m%d-%H%M%S"); + format!("{timestamp}") +} + +async fn load_config(cli_overrides: CliConfigOverrides) -> Result { + let overrides = cli_overrides + .parse_overrides() + .map_err(|err| anyhow!("failed to parse -c overrides: {err}"))?; + Config::load_with_cli_overrides(overrides, ConfigOverrides::default()) + .await + .context("failed to load Codex configuration") +} + +fn load_auth(config: &Config) -> Result { + if let Some(auth) = + CodexAuth::from_codex_home(&config.codex_home).context("failed to read auth.json")? + { + return Ok(auth); + } + if let Some(api_key) = read_codex_api_key_from_env() { + return Ok(CodexAuth::from_api_key(&api_key)); + } + if let Some(api_key) = read_openai_api_key_from_env() { + return Ok(CodexAuth::from_api_key(&api_key)); + } + bail!("no Codex authentication found. Run `codex login` or set OPENAI_API_KEY."); +} + +fn resolve_runs_root(override_path: Option) -> Result { + if let Some(path) = override_path { + return Ok(path); + } + Ok(codex_infty::default_runs_root()?) +} + +fn collect_run_summaries(root: &Path) -> Result> { + let mut summaries = Vec::new(); + let iter = match fs::read_dir(root) { + Ok(read_dir) => read_dir, + Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(summaries), + Err(err) => { + return Err( + anyhow!(err).context(format!("failed to read runs root {}", root.display())) + ); + } + }; + + for entry in iter { + let entry = entry?; + if !entry.file_type()?.is_dir() { + continue; + } + let run_path = entry.path(); + let store = match RunStore::load(&run_path) { + Ok(store) => store, + Err(err) => { + eprintln!( + "Skipping {}: failed to load run metadata: {err}", + run_path.display() + ); + continue; + } + }; + let metadata = store.metadata(); + summaries.push(RunSummary { + run_id: metadata.run_id.clone(), + path: run_path.display().to_string(), + created_at: metadata + .created_at + .to_rfc3339_opts(SecondsFormat::Secs, true), + updated_at: metadata + .updated_at + .to_rfc3339_opts(SecondsFormat::Secs, true), + roles: metadata + .roles + .iter() + .map(|role| role.role.clone()) + .collect(), + }); + } + + summaries.sort_by(|a, b| b.updated_at.cmp(&a.updated_at)); + Ok(summaries) +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::TempDir; + + #[test] + fn validates_run_ids() { + assert!(validate_run_id("run-20241030-123000").is_ok()); + assert!(validate_run_id("run.alpha").is_ok()); + assert!(validate_run_id("").is_err()); + assert!(validate_run_id("..bad").is_err()); + assert!(validate_run_id("bad/value").is_err()); + } + + #[test] + fn generates_timestamped_run_id() { + let run_id = generate_run_id(); + assert!(run_id.starts_with("run-")); + assert_eq!(run_id.len(), "run-YYYYMMDD-HHMMSS".len()); + } + + #[test] + fn collect_summaries_returns_empty_for_missing_root() { + let temp = TempDir::new().expect("temp dir"); + let missing = temp.path().join("not-present"); + let summaries = collect_run_summaries(&missing).expect("collect"); + assert!(summaries.is_empty()); + } +} diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index cd43041ab0..79ddb58888 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -23,8 +23,10 @@ use owo_colors::OwoColorize; use std::path::PathBuf; use supports_color::Stream; +mod infty; mod mcp_cmd; +use crate::infty::InftyCli; use crate::mcp_cmd::McpCli; /// Codex CLI @@ -97,6 +99,10 @@ enum Subcommand { /// Internal: run the responses API proxy. #[clap(hide = true)] ResponsesApiProxy(ResponsesApiProxyArgs), + + /// [experimental] Manage Codex Infty long-running task runs. + #[clap(name = "infty")] + Infty(InftyCli), } #[derive(Debug, Parser)] @@ -343,6 +349,13 @@ async fn cli_main(codex_linux_sandbox_exe: Option) -> anyhow::Result<() ); codex_cloud_tasks::run_main(cloud_cli, codex_linux_sandbox_exe).await?; } + Some(Subcommand::Infty(mut infty_cli)) => { + prepend_config_flags( + &mut infty_cli.config_overrides, + root_config_overrides.clone(), + ); + infty_cli.run().await?; + } Some(Subcommand::Sandbox(sandbox_args)) => match sandbox_args.cmd { SandboxCommand::Macos(mut seatbelt_cli) => { prepend_config_flags( diff --git a/codex-rs/codex-infty/Cargo.toml b/codex-rs/codex-infty/Cargo.toml new file mode 100644 index 0000000000..65e4b38d49 --- /dev/null +++ b/codex-rs/codex-infty/Cargo.toml @@ -0,0 +1,22 @@ +[package] +name = "codex-infty" +version = { workspace = true } +edition = "2024" + +[dependencies] +anyhow = { workspace = true } +chrono = { workspace = true, features = ["serde"] } +codex-core = { path = "../core" } +codex-protocol = { path = "../protocol" } +dirs = { workspace = true } +serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } +tempfile = { workspace = true } +tokio = { workspace = true, features = ["macros", "rt", "rt-multi-thread"] } +tokio-stream = { workspace = true } +tracing = { workspace = true, features = ["log"] } + +[dev-dependencies] +core_test_support = { path = "../core/tests/common" } +tempfile = { workspace = true } +wiremock = { workspace = true } diff --git a/codex-rs/codex-infty/src/lib.rs b/codex-rs/codex-infty/src/lib.rs new file mode 100644 index 0000000000..c586542b43 --- /dev/null +++ b/codex-rs/codex-infty/src/lib.rs @@ -0,0 +1,977 @@ +#![deny(clippy::print_stdout, clippy::print_stderr)] + +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; +use anyhow::anyhow; +use anyhow::bail; +use codex_core::CodexAuth; +use codex_core::CodexConversation; +use codex_core::ConversationManager; +use codex_core::CrossSessionSpawnParams; +use codex_core::NewConversation; +use codex_core::config::Config; +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; +use codex_core::protocol::SessionConfiguredEvent; +use codex_protocol::ConversationId; +use dirs::home_dir; +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_stream::StreamExt; +use tracing::warn; + +mod prompts; +mod run_store; + +pub use run_store::RoleMetadata; +pub use run_store::RunMetadata; +pub use run_store::RunStore; + +#[derive(Clone)] +pub struct RoleConfig { + pub role: String, + pub config: Config, + pub config_path: Option, +} + +impl RoleConfig { + pub fn new(role: impl Into, config: Config) -> Self { + Self { + role: role.into(), + config, + config_path: None, + } + } + + pub fn with_path(role: impl Into, config: Config, config_path: PathBuf) -> Self { + Self { + role: role.into(), + config, + config_path: Some(config_path), + } + } +} + +pub struct RunParams { + pub run_id: String, + pub run_root: Option, + pub solver: RoleConfig, + pub director: RoleConfig, + pub verifiers: Vec, +} + +pub struct ResumeParams { + pub run_path: PathBuf, + pub solver: RoleConfig, + pub director: RoleConfig, + pub verifiers: Vec, +} + +const DEFAULT_DIRECTOR_TIMEOUT: Duration = Duration::from_secs(120); +const DEFAULT_VERIFIER_TIMEOUT: Duration = Duration::from_secs(180); +const FINALIZATION_PROMPT: &str = "Create deliverable/: include compiled artifacts or scripts, usage docs, and tests. Write deliverable/README.md with overview, manifest (paths and sizes), verification steps, and limitations. Remove scratch files. Reply with JSON: {\"type\":\"final_delivery\",\"deliverable_path\":\"\",\"summary\":\"\"}."; + +#[derive(Clone)] +pub struct RunExecutionOptions { + pub objective: Option, + pub director_timeout: Duration, + pub verifier_timeout: Duration, +} + +impl Default for RunExecutionOptions { + fn default() -> Self { + Self { + objective: None, + director_timeout: DEFAULT_DIRECTOR_TIMEOUT, + verifier_timeout: DEFAULT_VERIFIER_TIMEOUT, + } + } +} + +pub struct RunOutcome { + pub run_id: String, + pub deliverable_path: PathBuf, + pub summary: Option, + pub raw_message: String, +} + +pub struct RoleSession { + pub role: String, + pub conversation_id: ConversationId, + pub conversation: Arc, + pub session_configured: SessionConfiguredEvent, + pub rollout_path: PathBuf, +} + +#[derive(Debug, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +enum SolverSignal { + DirectionRequest { + #[serde(default)] + prompt: String, + }, + VerificationRequest { + claim_path: String, + #[serde(default)] + notes: Option, + }, + FinalDelivery { + deliverable_path: String, + #[serde(default)] + summary: Option, + }, +} + +#[derive(Debug, Deserialize, Serialize)] +struct DirectiveResponse { + directive: String, + #[serde(default)] + rationale: Option, +} + +#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +enum VerifierDecision { + Pass, + Fail, +} + +impl VerifierDecision { + fn is_pass(self) -> bool { + matches!(self, VerifierDecision::Pass) + } +} + +#[derive(Debug, Deserialize, Serialize)] +struct VerifierVerdict { + verdict: VerifierDecision, + #[serde(default)] + reasons: Vec, + #[serde(default)] + suggestions: Vec, +} + +#[derive(Debug, Serialize)] +struct VerifierReport { + role: String, + verdict: VerifierDecision, + #[serde(default)] + reasons: Vec, + #[serde(default)] + suggestions: Vec, +} + +#[derive(Debug, Serialize)] +struct AggregatedVerifierVerdict { + #[serde(rename = "type")] + kind: &'static str, + overall: VerifierDecision, + verdicts: Vec, +} + +struct SessionCleanup { + conversation_id: ConversationId, + conversation: Arc, +} + +impl SessionCleanup { + fn new(session: &RoleSession) -> Self { + Self { + conversation_id: session.conversation_id, + conversation: Arc::clone(&session.conversation), + } + } +} + +pub struct RunSessions { + pub run_id: String, + pub solver: RoleSession, + pub director: RoleSession, + pub verifiers: Vec, + pub store: RunStore, +} + +pub struct InftyOrchestrator { + hub: Arc, + conversation_manager: ConversationManager, + runs_root: PathBuf, +} + +impl InftyOrchestrator { + pub fn new(auth: CodexAuth) -> Result { + let runs_root = default_runs_root()?; + Ok(Self::with_runs_root(auth, runs_root)) + } + + pub fn with_runs_root(auth: CodexAuth, runs_root: impl Into) -> Self { + Self { + hub: Arc::new(CrossSessionHub::new()), + conversation_manager: ConversationManager::with_auth(auth), + runs_root: runs_root.into(), + } + } + + pub fn runs_root(&self) -> &PathBuf { + &self.runs_root + } + + pub fn hub(&self) -> Arc { + Arc::clone(&self.hub) + } + + pub async fn execute_new_run( + &self, + params: RunParams, + options: RunExecutionOptions, + ) -> Result { + let sessions = self.spawn_run(params).await?; + self.drive_run(sessions, options).await + } + + pub async fn execute_existing_run( + &self, + params: ResumeParams, + options: RunExecutionOptions, + ) -> Result { + let sessions = self.resume_run(params).await?; + self.drive_run(sessions, options).await + } + + pub async fn spawn_run(&self, params: RunParams) -> Result { + let RunParams { + run_id, + run_root, + solver, + director, + verifiers, + } = params; + + let run_path = run_root.unwrap_or_else(|| self.runs_root.join(&run_id)); + let role_metadata = collect_role_metadata(&solver, &director, &verifiers); + let mut store = RunStore::initialize(&run_path, &run_id, &role_metadata)?; + let mut cleanup = Vec::new(); + + let solver_session = match self + .spawn_role_session(&run_id, &run_path, solver.clone()) + .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 + } + Err(err) => { + self.cleanup_failed_spawn(cleanup, &run_path).await; + return Err(err); + } + }; + + let director_session = match self + .spawn_role_session(&run_id, &run_path, director.clone()) + .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 + } + Err(err) => { + self.cleanup_failed_spawn(cleanup, &run_path).await; + return Err(err); + } + }; + + 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()) + .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 + } + Err(err) => { + self.cleanup_failed_spawn(cleanup, &run_path).await; + return Err(err); + } + }; + verifier_sessions.push(session); + } + + Ok(RunSessions { + run_id, + solver: solver_session, + director: director_session, + verifiers: verifier_sessions, + store, + }) + } + + pub async fn resume_run(&self, params: ResumeParams) -> Result { + let ResumeParams { + run_path, + solver, + director, + verifiers, + } = params; + + let mut store = RunStore::load(&run_path)?; + let run_id = store.metadata().run_id.clone(); + let mut cleanup = Vec::new(); + + let run_path = store.path().to_path_buf(); + + let solver_session = match self + .resume_role_session(&run_id, &run_path, &solver, &store) + .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 + } + Err(err) => { + self.cleanup_failed_resume(cleanup).await; + return Err(err); + } + }; + + let director_session = match self + .resume_role_session(&run_id, &run_path, &director, &store) + .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 + } + Err(err) => { + self.cleanup_failed_resume(cleanup).await; + return Err(err); + } + }; + + 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) + .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 + } + Err(err) => { + self.cleanup_failed_resume(cleanup).await; + return Err(err); + } + }; + verifier_sessions.push(session); + } + + store.touch()?; + + Ok(RunSessions { + run_id, + solver: solver_session, + director: director_session, + verifiers: verifier_sessions, + store, + }) + } + + async fn drive_run( + &self, + mut sessions: RunSessions, + options: RunExecutionOptions, + ) -> Result { + let result = self.inner_drive_run(&mut sessions, &options).await; + let cleanup = collect_session_cleanup(&sessions); + self.shutdown_sessions(cleanup).await; + result + } + + async fn inner_drive_run( + &self, + sessions: &mut RunSessions, + options: &RunExecutionOptions, + ) -> 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?; + sessions.store.touch()?; + } + + while let Some(event) = solver_events.next().await { + match event.event.msg { + EventMsg::AgentMessage(agent_msg) => { + 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?; + sessions.store.touch()?; + } + SolverSignal::VerificationRequest { claim_path, notes } => { + let _ = (claim_path, notes); + let pass = self + .handle_verification_request( + sessions, + &agent_msg.message, + options, + ) + .await?; + sessions.store.touch()?; + if pass { + self.post_to_role( + &sessions.run_id, + &sessions.solver.role, + FINALIZATION_PROMPT.to_string(), + Some(final_delivery_schema()), + ) + .await?; + sessions.store.touch()?; + } + } + SolverSignal::FinalDelivery { + deliverable_path: candidate_path, + summary, + } => { + sessions.store.touch()?; + let deliverable_path = resolve_deliverable_path( + sessions.store.path(), + &candidate_path, + ) + .with_context(|| { + format!( + "invalid deliverable path reported by solver: {candidate_path}" + ) + })?; + return Ok(RunOutcome { + run_id: sessions.run_id.clone(), + deliverable_path, + summary, + raw_message: agent_msg.message, + }); + } + } + } + } + EventMsg::ShutdownComplete => { + break; + } + _ => {} + } + } + + Err(anyhow!( + "run {} ended before emitting final_delivery message", + sessions.run_id + )) + } + + async fn handle_direction_request( + &self, + sessions: &RunSessions, + raw_message: &str, + options: &RunExecutionOptions, + ) -> Result<()> { + let handle = self + .post_to_role( + &sessions.run_id, + &sessions.director.role, + raw_message.to_string(), + Some(director_schema()), + ) + .await?; + let directive = self + .await_first_assistant(&handle, options.director_timeout) + .await?; + let directive_payload: DirectiveResponse = parse_json_struct(&directive.message.message) + .context("director response was not valid directive JSON")?; + let directive_text = serde_json::to_string_pretty(&directive_payload)?; + let _ = self + .post_to_role( + &sessions.run_id, + &sessions.solver.role, + directive_text, + None, + ) + .await?; + Ok(()) + } + + async fn handle_verification_request( + &self, + sessions: &RunSessions, + raw_message: &str, + options: &RunExecutionOptions, + ) -> Result { + if sessions.verifiers.is_empty() { + let summary = aggregate_verdicts(Vec::new()); + let summary_text = serde_json::to_string_pretty(&summary)?; + let _ = self + .post_to_role(&sessions.run_id, &sessions.solver.role, summary_text, None) + .await?; + return Ok(true); + } + + 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, + raw_message.to_string(), + Some(verifier_schema()), + ) + .await?; + let response = self + .await_first_assistant(&handle, options.verifier_timeout) + .await?; + let verdict: VerifierVerdict = parse_json_struct(&response.message.message) + .with_context(|| { + format!("verifier {} returned invalid verdict JSON", verifier.role) + })?; + collected.push((verifier.role.clone(), verdict)); + } + + let summary = aggregate_verdicts(collected); + let summary_text = serde_json::to_string_pretty(&summary)?; + let _ = self + .post_to_role(&sessions.run_id, &sessions.solver.role, summary_text, None) + .await?; + Ok(summary.overall.is_pass()) + } + + 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" + ); + } + } + } + + async fn cleanup_failed_resume(&self, sessions: Vec) { + self.shutdown_sessions(sessions).await; + } + + async fn shutdown_sessions(&self, sessions: Vec) { + for session in sessions { + if let Err(err) = session.conversation.submit(Op::Shutdown).await { + warn!( + %session.conversation_id, + ?err, + "failed to shutdown session during cleanup" + ); + } + let _ = self + .conversation_manager + .remove_conversation(&session.conversation_id) + .await; + } + } + + pub async fn post_to_role( + &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) + } + + pub async fn call_role( + &self, + run_id: &str, + role: &str, + text: impl Into, + 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(&handle, timeout).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 = self + .post_to_role( + run_id, + target_role, + assistant.message.message.clone(), + final_output_json_schema, + ) + .await?; + self.await_first_assistant(&handle, timeout).await + } + + pub fn stream_events( + &self, + conversation_id: ConversationId, + ) -> Result { + self.hub.stream_events(conversation_id) + } + + 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) + .ok_or_else(|| anyhow!("role {} not found in run metadata", role_config.role))?; + let rollout_path = metadata + .rollout_path + .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)) + } +} + +impl RoleSession { + fn from_new(role: String, session: NewConversation) -> Self { + Self { + role, + conversation_id: session.conversation_id, + conversation: session.conversation, + session_configured: session.session_configured.clone(), + rollout_path: session.session_configured.rollout_path.clone(), + } + } +} + +fn collect_role_metadata( + solver: &RoleConfig, + director: &RoleConfig, + verifiers: &[RoleConfig], +) -> Vec { + solver_and_director_metadata(solver, director) + .into_iter() + .chain(verifiers.iter().map(|verifier| RoleMetadata { + role: verifier.role.clone(), + rollout_path: None, + config_path: verifier.config_path.clone(), + })) + .collect() +} + +fn solver_and_director_metadata(solver: &RoleConfig, director: &RoleConfig) -> Vec { + vec![ + RoleMetadata { + role: solver.role.clone(), + rollout_path: None, + config_path: solver.config_path.clone(), + }, + RoleMetadata { + role: director.role.clone(), + rollout_path: None, + config_path: director.config_path.clone(), + }, + ] +} + +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 director_schema() -> Value { + json!({ + "type": "object", + "required": ["directive"], + "properties": { + "directive": { "type": "string" }, + "rationale": { "type": "string" } + }, + "additionalProperties": false + }) +} + +fn verifier_schema() -> Value { + json!({ + "type": "object", + "required": ["verdict"], + "properties": { + "verdict": { "type": "string", "enum": ["pass", "fail"] }, + "reasons": { "type": "array", "items": { "type": "string" } }, + "suggestions": { "type": "array", "items": { "type": "string" } } + }, + "additionalProperties": false + }) +} + +fn final_delivery_schema() -> Value { + json!({ + "type": "object", + "required": ["type", "deliverable_path"], + "properties": { + "type": { "const": "final_delivery" }, + "deliverable_path": { "type": "string" }, + "summary": { "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) +} + +fn collect_session_cleanup(sessions: &RunSessions) -> Vec { + let mut cleanup = Vec::with_capacity(2 + sessions.verifiers.len()); + cleanup.push(SessionCleanup::new(&sessions.solver)); + cleanup.push(SessionCleanup::new(&sessions.director)); + cleanup.extend(sessions.verifiers.iter().map(SessionCleanup::new)); + cleanup +} + +pub fn default_runs_root() -> Result { + let home = home_dir().ok_or_else(|| anyhow!("failed to determine home directory"))?; + Ok(home.join(".codex").join("infty")) +} diff --git a/codex-rs/codex-infty/src/prompts.rs b/codex-rs/codex-infty/src/prompts.rs new file mode 100644 index 0000000000..0f50d81e67 --- /dev/null +++ b/codex-rs/codex-infty/src/prompts.rs @@ -0,0 +1,147 @@ +use codex_core::config::Config; + +const SOLVER_PROMPT: &str = r#"# Codex Infty Solver + +You are the **Solver** role in a Codex Infty run. Drive the engagement end to end without waiting for humans. Maintain momentum for multi-hour or multi-day efforts. + +Responsibilities: +- Understand the objective and break it into a living execution plan. Refine plans with `update_plan` and keep the run store up to date. +- Produce artifacts under `artifacts/`, durable notes under `memory/`, and supporting indexes under `index/`. Prefer `apply_patch` for text edits and use `shell` for other filesystem work. +- When you exit a task or take a dependency on external evidence, write JSON notes in `memory/claims/` that link to the supporting artifacts. +- Run verification steps (tests, linters, proofs) under the sandbox before claiming completion. + +Available Codex tools mirror standard Codex sessions (e.g. `shell`, `apply_patch`, `read_file`, `list_dir`, `grep_files`). Assume all filesystem paths are relative to the current run store directory unless stated otherwise. + +## Communication contract +The orchestrator routes your structured messages to the Director or Verifier roles. Respond with **JSON only**—no leading prose or trailing commentary. Wrap JSON in a fenced block only if the agent policy forces it. + +- Direction request (send to Director): + ```json + {"type":"direction_request","prompt":""} + ``` +- Verification request (send to Verifier): + ```json + {"type":"verification_request","claim_path":"memory/claims/.json","notes":""} + ``` +- Final delivery (after receiving the finalization instruction): + ```json + {"type":"final_delivery","deliverable_path":"deliverable","summary":""} + ``` + +If you have nothing to add for `notes`, omit the field. + +## Operating rhythm +- Never ask humans for approval to continue; the orchestrator supplies direction via the Director role. +- Keep the run resilient to restarts: document intent, intermediate results, and follow-up tasks in `memory/`. +- Prefer concrete evidence (tests, diffs, logs). Link every claim to artifacts or durable notes so the Verifier can reproduce your reasoning. +- On failure feedback from a Verifier, update artifacts/notes/tests, then issue a new verification request referencing the superseding claim. +- When the orchestrator instructs you to finalize, build the `deliverable/` directory exactly as requested, summarise the outcome, and respond with the `final_delivery` JSON."#; + +const DIRECTOR_PROMPT: &str = r#"# Codex Infty Director + +You are the **Director** role. The Solver routes direction questions to you. Provide crisp guidance that keeps the run aligned with the objective, risks, and verification needs. + +Guidelines: +- Read Solver context from the question, referenced notes, and run store artifacts. +- Fill gaps in requirements, adjust strategy, or re-prioritize tasks when the plan drifts. +- Highlight mandatory verification or documentation steps the Solver must complete. + +Respond **only** with JSON in this exact shape: +```json +{"directive":"","rationale":""} +``` + +Keep `directive` actionable and concise. Use `rationale` for supporting detail. Leave `rationale` empty if it adds no value."#; + +const VERIFIER_PROMPT: &str = r#"# Codex Infty Verifier + +You are a **Verifier**. Assess Solver completion claims objectively. + +Process: +1. Inspect the referenced claim JSON and any linked artifacts, tests, or logs inside the run store. +2. Reproduce evidence when feasible (e.g. run tests via `shell`). Exit early if sandbox restrictions apply and explain the limitation. +3. Evaluate correctness, completeness, and policy alignment. Look for missing tests, undocumented gaps, regressions, or unverifiable assertions. + +Respond **only** with JSON in this form: +```json +{"verdict":"pass","reasons":[],"suggestions":[]} +``` +Use `"fail"` when the claim is not ready. Populate `reasons` with concrete blocking issues. Provide actionable `suggestions` for remediation. Omit entries when not needed. + +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()); + } + } +} + +fn default_instructions_for_role(role: &str) -> Option<&'static str> { + let normalized = role.to_ascii_lowercase(); + if normalized == "solver" { + Some(SOLVER_PROMPT) + } else if normalized == "director" { + Some(DIRECTOR_PROMPT) + } else if normalized.starts_with("verifier") { + Some(VERIFIER_PROMPT) + } else { + None + } +} + +#[cfg(test)] +mod tests { + use super::*; + use core_test_support::load_default_config_for_test; + use tempfile::TempDir; + + #[test] + fn provides_prompts_for_known_roles() { + let home = TempDir::new().unwrap(); + let mut config = load_default_config_for_test(&home); + config.base_instructions = None; + ensure_instructions("solver", &mut config); + assert!( + config + .base_instructions + .as_ref() + .unwrap() + .contains("Codex Infty Solver") + ); + + let home = TempDir::new().unwrap(); + let mut config = load_default_config_for_test(&home); + config.base_instructions = None; + ensure_instructions("director", &mut config); + assert!( + config + .base_instructions + .as_ref() + .unwrap() + .contains("Codex Infty Director") + ); + + let home = TempDir::new().unwrap(); + let mut config = load_default_config_for_test(&home); + config.base_instructions = None; + ensure_instructions("verifier-alpha", &mut config); + assert!( + config + .base_instructions + .as_ref() + .unwrap() + .contains("Codex Infty Verifier") + ); + } + + #[test] + fn does_not_override_existing_instructions() { + let home = TempDir::new().unwrap(); + let mut config = load_default_config_for_test(&home); + config.base_instructions = Some("custom".to_string()); + ensure_instructions("solver", &mut config); + assert_eq!(config.base_instructions.as_deref(), Some("custom")); + } +} diff --git a/codex-rs/codex-infty/src/run_store.rs b/codex-rs/codex-infty/src/run_store.rs new file mode 100644 index 0000000000..a286c34e13 --- /dev/null +++ b/codex-rs/codex-infty/src/run_store.rs @@ -0,0 +1,211 @@ +use std::fs; +use std::io::Write; +use std::path::Path; +use std::path::PathBuf; + +use anyhow::Context; +use anyhow::Result; +use anyhow::anyhow; +use chrono::DateTime; +use chrono::Utc; +use serde::Deserialize; +use serde::Serialize; +use tempfile::NamedTempFile; + +const ARTIFACTS_DIR: &str = "artifacts"; +const MEMORY_DIR: &str = "memory"; +const INDEX_DIR: &str = "index"; +const DELIVERABLE_DIR: &str = "deliverable"; +const METADATA_FILE: &str = "run.json"; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RoleMetadata { + pub role: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub rollout_path: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub config_path: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RunMetadata { + pub run_id: String, + pub created_at: DateTime, + pub updated_at: DateTime, + pub roles: Vec, +} + +#[derive(Debug, Clone)] +pub struct RunStore { + path: PathBuf, + metadata: RunMetadata, +} + +impl RunStore { + pub fn initialize( + run_path: impl AsRef, + run_id: &str, + roles: &[RoleMetadata], + ) -> Result { + let run_path = run_path.as_ref().to_path_buf(); + fs::create_dir_all(&run_path) + .with_context(|| format!("failed to create run directory {}", run_path.display()))?; + + for child in [ARTIFACTS_DIR, MEMORY_DIR, INDEX_DIR, DELIVERABLE_DIR] { + fs::create_dir_all(run_path.join(child)) + .with_context(|| format!("failed to create subdirectory {child}"))?; + } + + let metadata_path = run_path.join(METADATA_FILE); + if metadata_path.exists() { + return Err(anyhow!( + "run metadata already exists at {}", + metadata_path.display() + )); + } + + let now = Utc::now(); + let metadata = RunMetadata { + run_id: run_id.to_string(), + created_at: now, + updated_at: now, + roles: roles.to_vec(), + }; + write_metadata(&metadata_path, &metadata)?; + + Ok(Self { + path: run_path, + metadata, + }) + } + + pub fn load(run_path: impl AsRef) -> Result { + let run_path = run_path.as_ref().to_path_buf(); + let metadata_path = run_path.join(METADATA_FILE); + let metadata: RunMetadata = serde_json::from_slice( + &fs::read(&metadata_path) + .with_context(|| format!("failed to read {}", metadata_path.display()))?, + ) + .with_context(|| format!("failed to parse {}", metadata_path.display()))?; + + Ok(Self { + path: run_path, + metadata, + }) + } + + pub fn path(&self) -> &Path { + &self.path + } + + pub fn metadata(&self) -> &RunMetadata { + &self.metadata + } + + pub fn role_metadata(&self, role: &str) -> Option<&RoleMetadata> { + self.metadata.roles.iter().find(|meta| meta.role == role) + } + + pub fn update_rollout_path(&mut self, role: &str, rollout_path: PathBuf) -> Result<()> { + if let Some(meta) = self + .metadata + .roles + .iter_mut() + .find(|meta| meta.role == role) + { + meta.rollout_path = Some(rollout_path); + self.commit_metadata() + } else { + Err(anyhow!("role {role} not found in run store")) + } + } + + pub fn set_role_config_path(&mut self, role: &str, path: PathBuf) -> Result<()> { + if let Some(meta) = self + .metadata + .roles + .iter_mut() + .find(|meta| meta.role == role) + { + meta.config_path = Some(path); + self.commit_metadata() + } else { + Err(anyhow!("role {role} not found in run store")) + } + } + + pub fn touch(&mut self) -> Result<()> { + self.metadata.updated_at = Utc::now(); + self.commit_metadata() + } + + fn commit_metadata(&mut self) -> Result<()> { + self.metadata.updated_at = Utc::now(); + let metadata_path = self.path.join(METADATA_FILE); + write_metadata(&metadata_path, &self.metadata) + } +} + +fn write_metadata(path: &Path, metadata: &RunMetadata) -> Result<()> { + let parent = path + .parent() + .ok_or_else(|| anyhow!("metadata path must have parent"))?; + let mut temp = NamedTempFile::new_in(parent) + .with_context(|| format!("failed to create temp file in {}", parent.display()))?; + serde_json::to_writer_pretty(&mut temp, metadata)?; + temp.flush()?; + temp.persist(path) + .with_context(|| format!("failed to persist metadata to {}", path.display()))?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::TempDir; + + #[test] + fn initialize_creates_directories_and_metadata() { + let temp = TempDir::new().unwrap(); + let run_path = temp.path().join("run_1"); + let roles = vec![ + RoleMetadata { + role: "solver".into(), + rollout_path: None, + config_path: None, + }, + RoleMetadata { + role: "director".into(), + rollout_path: None, + config_path: None, + }, + ]; + + let store = RunStore::initialize(&run_path, "run_1", &roles).unwrap(); + assert!(store.path().join(ARTIFACTS_DIR).is_dir()); + assert!(store.path().join(MEMORY_DIR).is_dir()); + assert!(store.path().join(INDEX_DIR).is_dir()); + assert!(store.path().join(DELIVERABLE_DIR).is_dir()); + assert_eq!(store.metadata().roles.len(), 2); + } + + #[test] + fn update_rollout_persists_metadata() { + let temp = TempDir::new().unwrap(); + let run_path = temp.path().join("run_2"); + let roles = vec![RoleMetadata { + role: "solver".into(), + rollout_path: None, + config_path: None, + }]; + let mut store = RunStore::initialize(&run_path, "run_2", &roles).unwrap(); + let rollout = PathBuf::from("/tmp/rollout.jsonl"); + store + .update_rollout_path("solver", rollout.clone()) + .unwrap(); + + let loaded = RunStore::load(&run_path).unwrap(); + let solver = loaded.role_metadata("solver").unwrap(); + assert_eq!(solver.rollout_path.as_ref().unwrap(), &rollout); + } +} diff --git a/codex-rs/codex-infty/tests/orchestrator.rs b/codex-rs/codex-infty/tests/orchestrator.rs new file mode 100644 index 0000000000..122e77fded --- /dev/null +++ b/codex-rs/codex-infty/tests/orchestrator.rs @@ -0,0 +1,358 @@ +#![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_core::protocol::Op; +use codex_infty::InftyOrchestrator; +use codex_infty::ResumeParams; +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 orchestrator_routes_between_roles_and_records_store() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + + let server = responses::start_mock_server().await; + let bodies = vec![ + responses::sse(vec![ + responses::ev_response_created("solver-resp-1"), + responses::ev_assistant_message("solver-msg-1", "Need direction"), + responses::ev_completed("solver-resp-1"), + ]), + responses::sse(vec![ + responses::ev_response_created("director-resp-1"), + responses::ev_assistant_message("director-msg-1", "Proceed iteratively"), + responses::ev_completed("director-resp-1"), + ]), + responses::sse(vec![ + responses::ev_response_created("solver-resp-2"), + responses::ev_assistant_message("solver-msg-2", "Acknowledged"), + responses::ev_completed("solver-resp-2"), + ]), + ]; + let response_mock = responses::mount_sse_sequence(&server, bodies).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-orchestrator".to_string(); + + let solver_config = build_config(&server).await?; + let director_config = build_config(&server).await?; + + let 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.clone()), + director: RoleConfig::new("director", director_config.clone()), + verifiers: Vec::new(), + }) + .await?; + + let solver_message = orchestrator + .call_role( + &sessions.run_id, + "solver", + "kick off plan", + Duration::from_secs(1), + None, + ) + .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?; + 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?; + assert_eq!(solver_reply.message.message, "Acknowledged"); + + assert_eq!(response_mock.requests().len(), 3); + let first_request = response_mock.requests().first().unwrap().body_json(); + let instructions = first_request["instructions"] + .as_str() + .expect("request should set instructions"); + assert!( + instructions.contains("Codex Infty Solver"), + "missing solver prompt: {instructions}" + ); + assert!(sessions.store.path().is_dir()); + let solver_meta = sessions.store.role_metadata("solver").unwrap(); + assert!(solver_meta.rollout_path.is_some()); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn orchestrator_resumes_existing_run() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + + let server = responses::start_mock_server().await; + let bodies = vec![ + responses::sse(vec![ + responses::ev_response_created("solver-resp-1"), + responses::ev_assistant_message("solver-msg-1", "Need direction"), + responses::ev_completed("solver-resp-1"), + ]), + responses::sse(vec![ + responses::ev_response_created("director-resp-1"), + responses::ev_assistant_message("director-msg-1", "Proceed iteratively"), + responses::ev_completed("director-resp-1"), + ]), + responses::sse(vec![ + responses::ev_response_created("solver-resp-2"), + responses::ev_assistant_message("solver-msg-2", "Acknowledged"), + responses::ev_completed("solver-resp-2"), + ]), + ]; + responses::mount_sse_sequence(&server, bodies).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-resume".to_string(); + + let solver_config = build_config(&server).await?; + let director_config = build_config(&server).await?; + + let 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.clone()), + director: RoleConfig::new("director", director_config.clone()), + verifiers: Vec::new(), + }) + .await?; + + let resume = orchestrator + .resume_run(ResumeParams { + run_path: sessions.store.path().to_path_buf(), + solver: RoleConfig::new("solver", solver_config), + director: RoleConfig::new("director", director_config), + verifiers: Vec::new(), + }) + .await?; + + assert_eq!(resume.run_id, run_id); + assert!( + resume + .store + .role_metadata("solver") + .unwrap() + .rollout_path + .is_some() + ); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn execute_new_run_drives_to_completion() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + + let server = responses::start_mock_server().await; + let bodies = vec![ + responses::sse(vec![ + responses::ev_response_created("solver-resp-1"), + responses::ev_assistant_message( + "solver-msg-1", + r#"{"type":"direction_request","prompt":"Need directive"}"#, + ), + responses::ev_completed("solver-resp-1"), + ]), + responses::sse(vec![ + responses::ev_response_created("director-resp-1"), + responses::ev_assistant_message( + "director-msg-1", + r#"{"directive":"Proceed","rationale":"Follow the plan"}"#, + ), + responses::ev_completed("director-resp-1"), + ]), + responses::sse(vec![ + responses::ev_response_created("solver-resp-2"), + responses::ev_assistant_message( + "solver-msg-2", + r#"{"type":"verification_request","claim_path":"memory/claims/attempt1.json"}"#, + ), + responses::ev_completed("solver-resp-2"), + ]), + responses::sse(vec![ + responses::ev_response_created("verifier-resp-1"), + responses::ev_assistant_message( + "verifier-msg-1", + r#"{"verdict":"fail","reasons":["Missing tests"],"suggestions":["Add regression tests"]}"#, + ), + responses::ev_completed("verifier-resp-1"), + ]), + responses::sse(vec![ + responses::ev_response_created("solver-resp-3"), + responses::ev_assistant_message( + "solver-msg-3", + r#"{"type":"verification_request","claim_path":"memory/claims/attempt2.json"}"#, + ), + responses::ev_completed("solver-resp-3"), + ]), + responses::sse(vec![ + responses::ev_response_created("verifier-resp-2"), + responses::ev_assistant_message( + "verifier-msg-2", + r#"{"verdict":"pass","reasons":[],"suggestions":[]}"#, + ), + responses::ev_completed("verifier-resp-2"), + ]), + responses::sse(vec![ + responses::ev_response_created("solver-resp-4"), + responses::ev_assistant_message( + "solver-msg-4", + r#"{"type":"final_delivery","deliverable_path":"deliverable","summary":"done"}"#, + ), + responses::ev_completed("solver-resp-4"), + ]), + ]; + responses::mount_sse_sequence(&server, bodies).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-auto".to_string(); + let run_root = runs_root.path().join("runs").join(&run_id); + + let solver_config = build_config(&server).await?; + 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 outcome = orchestrator + .execute_new_run( + RunParams { + run_id: run_id.clone(), + run_root: Some(run_root.clone()), + solver: RoleConfig::new("solver", solver_config), + director: RoleConfig::new("director", director_config), + verifiers: vec![RoleConfig::new("verifier", verifier_config)], + }, + options, + ) + .await?; + + assert_eq!(outcome.run_id, run_id); + assert_eq!(outcome.summary.as_deref(), Some("done")); + assert!(outcome.raw_message.contains("final_delivery")); + assert!(outcome.deliverable_path.starts_with(&run_root)); + assert_eq!(outcome.deliverable_path, run_root.join("deliverable")); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn spawn_run_cleans_up_on_failure() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + + let server = responses::start_mock_server().await; + let bodies = vec![ + responses::sse(vec![ + responses::ev_response_created("solver-resp-1"), + responses::ev_completed("solver-resp-1"), + ]), + responses::sse(vec![ + responses::ev_response_created("director-resp-1"), + responses::ev_completed("director-resp-1"), + ]), + responses::sse(vec![ + responses::ev_response_created("dup-resp"), + responses::ev_completed("dup-resp"), + ]), + ]; + responses::mount_sse_sequence(&server, bodies).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-cleanup".to_string(); + let run_path = runs_root.path().join("runs").join(&run_id); + + let solver_config = build_config(&server).await?; + let director_config = build_config(&server).await?; + + let result = orchestrator + .spawn_run(RunParams { + run_id: run_id.clone(), + run_root: Some(run_path.clone()), + solver: RoleConfig::new("solver", solver_config.clone()), + director: RoleConfig::new("director", director_config.clone()), + verifiers: vec![RoleConfig::new("solver", solver_config.clone())], + }) + .await; + assert!(result.is_err()); + assert!(!run_path.exists(), "failed run should remove run directory"); + + let bodies = vec![ + responses::sse(vec![ + responses::ev_response_created("solver-resp-2"), + responses::ev_completed("solver-resp-2"), + ]), + responses::sse(vec![ + responses::ev_response_created("director-resp-2"), + responses::ev_completed("director-resp-2"), + ]), + ]; + responses::mount_sse_sequence(&server, bodies).await; + + let sessions = orchestrator + .spawn_run(RunParams { + run_id: run_id.clone(), + run_root: Some(run_path.clone()), + solver: RoleConfig::new("solver", solver_config), + director: RoleConfig::new("director", director_config), + verifiers: Vec::new(), + }) + .await?; + + sessions.solver.conversation.submit(Op::Shutdown).await.ok(); + sessions + .director + .conversation + .submit(Op::Shutdown) + .await + .ok(); + + 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) +} diff --git a/codex-rs/core/Cargo.toml b/codex-rs/core/Cargo.toml index 4259e64fc4..219455eee2 100644 --- a/codex-rs/core/Cargo.toml +++ b/codex-rs/core/Cargo.toml @@ -62,6 +62,7 @@ tokio = { workspace = true, features = [ "signal", ] } tokio-util = { workspace = true, features = ["rt"] } +tokio-stream = { workspace = true, features = ["sync"] } toml = { workspace = true } toml_edit = { workspace = true } tracing = { workspace = true, features = ["log"] } diff --git a/codex-rs/core/src/conversation_manager.rs b/codex-rs/core/src/conversation_manager.rs index aeb0780725..513935afd7 100644 --- a/codex-rs/core/src/conversation_manager.rs +++ b/codex-rs/core/src/conversation_manager.rs @@ -7,10 +7,16 @@ use crate::codex::compact::content_items_to_text; use crate::codex::compact::is_session_prefix_message; use crate::codex_conversation::CodexConversation; use crate::config::Config; +use crate::cross_session::CrossSessionError; +use crate::cross_session::CrossSessionHub; +use crate::cross_session::RegisteredSession; +use crate::cross_session::SessionDefaults; +use crate::cross_session::SessionRegistration; use crate::error::CodexErr; use crate::error::Result as CodexResult; use crate::protocol::Event; use crate::protocol::EventMsg; +use crate::protocol::Op; use crate::protocol::SessionConfiguredEvent; use crate::rollout::RolloutRecorder; use codex_protocol::ConversationId; @@ -22,6 +28,7 @@ use std::collections::HashMap; use std::path::PathBuf; use std::sync::Arc; use tokio::sync::RwLock; +use tracing::warn; /// Represents a newly created Codex conversation, including the first event /// (which is [`EventMsg::SessionConfigured`]). @@ -31,10 +38,17 @@ pub struct NewConversation { pub session_configured: SessionConfiguredEvent, } +pub struct CrossSessionSpawnParams { + pub hub: Arc, + pub run_id: Option, + pub role: Option, +} + /// [`ConversationManager`] is responsible for creating conversations and /// maintaining them in memory. pub struct ConversationManager { conversations: Arc>>>, + cross_session_registrations: Arc>>, auth_manager: Arc, session_source: SessionSource, } @@ -43,6 +57,7 @@ impl ConversationManager { pub fn new(auth_manager: Arc, session_source: SessionSource) -> Self { Self { conversations: Arc::new(RwLock::new(HashMap::new())), + cross_session_registrations: Arc::new(RwLock::new(HashMap::new())), auth_manager, session_source, } @@ -58,26 +73,108 @@ impl ConversationManager { } pub async fn new_conversation(&self, config: Config) -> CodexResult { - self.spawn_conversation(config, self.auth_manager.clone()) - .await + self.spawn_conversation_with_history( + config, + self.auth_manager.clone(), + InitialHistory::New, + None, + ) + .await } - async fn spawn_conversation( + pub async fn new_conversation_with_cross_session( + &self, + config: Config, + params: CrossSessionSpawnParams, + ) -> CodexResult { + self.spawn_conversation_with_history( + config, + self.auth_manager.clone(), + InitialHistory::New, + Some(params), + ) + .await + } + + async fn spawn_conversation_with_history( &self, config: Config, auth_manager: Arc, + initial_history: InitialHistory, + cross_session: Option, ) -> CodexResult { + let cross_session = + cross_session.map(|params| (SessionDefaults::from_config(&config), params)); + let CodexSpawnOk { codex, conversation_id, - } = Codex::spawn( - config, - auth_manager, - InitialHistory::New, - self.session_source, - ) - .await?; - self.finalize_spawn(codex, conversation_id).await + } = Codex::spawn(config, auth_manager, initial_history, self.session_source).await?; + + let new_conversation = self.finalize_spawn(codex, conversation_id).await?; + + if let Some((defaults, params)) = cross_session { + if let Err(err) = self + .register_cross_session( + conversation_id, + defaults, + params, + Arc::clone(&new_conversation.conversation), + ) + .await + { + self.abort_conversation( + conversation_id, + Arc::clone(&new_conversation.conversation), + ) + .await; + return Err(CodexErr::Fatal(format!( + "failed to register cross-session for conversation {conversation_id}: {err}" + ))); + } + } + + Ok(new_conversation) + } + + async fn register_cross_session( + &self, + conversation_id: ConversationId, + defaults: SessionDefaults, + params: CrossSessionSpawnParams, + conversation: Arc, + ) -> Result<(), CrossSessionError> { + let CrossSessionSpawnParams { hub, run_id, role } = params; + + let registration = SessionRegistration { + conversation_id, + conversation, + defaults, + run_id, + role, + }; + + let guard = hub.register_session(registration)?; + self.cross_session_registrations + .write() + .await + .insert(conversation_id, guard); + Ok(()) + } + + async fn abort_conversation( + &self, + conversation_id: ConversationId, + conversation: Arc, + ) { + let _ = self.remove_conversation(&conversation_id).await; + if let Err(err) = conversation.submit(Op::Shutdown).await { + warn!( + %conversation_id, + ?err, + "failed to shutdown conversation after cross-session registration error" + ); + } } async fn finalize_spawn( @@ -130,11 +227,35 @@ impl ConversationManager { auth_manager: Arc, ) -> CodexResult { let initial_history = RolloutRecorder::get_rollout_history(&rollout_path).await?; - let CodexSpawnOk { - codex, - conversation_id, - } = Codex::spawn(config, auth_manager, initial_history, self.session_source).await?; - self.finalize_spawn(codex, conversation_id).await + self.spawn_conversation_with_history(config, auth_manager, initial_history, None) + .await + } + + pub async fn resume_conversation_from_rollout_with_cross_session( + &self, + config: Config, + rollout_path: PathBuf, + auth_manager: Arc, + params: CrossSessionSpawnParams, + ) -> CodexResult { + let initial_history = RolloutRecorder::get_rollout_history(&rollout_path).await?; + self.spawn_conversation_with_history(config, auth_manager, initial_history, Some(params)) + .await + } + + pub async fn resume_conversation_with_cross_session( + &self, + config: Config, + rollout_path: PathBuf, + params: CrossSessionSpawnParams, + ) -> CodexResult { + self.resume_conversation_from_rollout_with_cross_session( + config, + rollout_path, + self.auth_manager.clone(), + params, + ) + .await } /// Removes the conversation from the manager's internal map, though the @@ -145,6 +266,10 @@ impl ConversationManager { &self, conversation_id: &ConversationId, ) -> Option> { + self.cross_session_registrations + .write() + .await + .remove(conversation_id); self.conversations.write().await.remove(conversation_id) } @@ -164,12 +289,23 @@ impl ConversationManager { // Spawn a new conversation with the computed initial history. let auth_manager = self.auth_manager.clone(); - let CodexSpawnOk { - codex, - conversation_id, - } = Codex::spawn(config, auth_manager, history, self.session_source).await?; + self.spawn_conversation_with_history(config, auth_manager, history, None) + .await + } - self.finalize_spawn(codex, conversation_id).await + pub async fn fork_conversation_with_cross_session( + &self, + nth_user_message: usize, + config: Config, + path: PathBuf, + params: CrossSessionSpawnParams, + ) -> CodexResult { + let history = RolloutRecorder::get_rollout_history(&path).await?; + let history = truncate_before_nth_user_message(history, nth_user_message); + + let auth_manager = self.auth_manager.clone(); + self.spawn_conversation_with_history(config, auth_manager, history, Some(params)) + .await } } diff --git a/codex-rs/core/src/cross_session.rs b/codex-rs/core/src/cross_session.rs new file mode 100644 index 0000000000..b55ca34c9c --- /dev/null +++ b/codex-rs/core/src/cross_session.rs @@ -0,0 +1,607 @@ +use std::collections::HashMap; +use std::fmt; +use std::path::PathBuf; +use std::pin::Pin; +use std::sync::Arc; +use std::sync::Mutex as StdMutex; +use std::sync::RwLock; +use std::sync::RwLockReadGuard; +use std::sync::RwLockWriteGuard; +use std::time::Duration; + +use futures::Stream; +use serde_json::Value; +use tokio::sync::Mutex as TokioMutex; +use tokio::sync::broadcast; +use tokio::sync::oneshot; +use tokio::time; +use tokio_stream::wrappers::BroadcastStream; +use tokio_stream::wrappers::errors::BroadcastStreamRecvError; +use tracing::debug; +use tracing::error; + +use crate::codex_conversation::CodexConversation; +use crate::config::Config; +use crate::error::CodexErr; +use crate::protocol::AgentMessageEvent; +use crate::protocol::AskForApproval; +use crate::protocol::Event; +use crate::protocol::EventMsg; +use crate::protocol::InputItem; +use crate::protocol::Op; +use crate::protocol::SandboxPolicy; +use crate::protocol_config_types::ReasoningEffort as ReasoningEffortConfig; +use crate::protocol_config_types::ReasoningSummary as ReasoningSummaryConfig; +use codex_protocol::ConversationId; + +/// Default capacity for broadcast channels that fan out session events. +const EVENT_BUFFER_LEN: usize = 256; + +/// Encapsulates the defaults needed to submit a new `Op::UserTurn`. +#[derive(Debug, Clone)] +pub struct SessionDefaults { + pub cwd: PathBuf, + pub approval_policy: AskForApproval, + pub sandbox_policy: SandboxPolicy, + pub model: String, + pub effort: Option, + pub summary: ReasoningSummaryConfig, +} + +impl SessionDefaults { + pub fn from_config(config: &Config) -> Self { + Self { + cwd: config.cwd.clone(), + approval_policy: config.approval_policy, + sandbox_policy: config.sandbox_policy.clone(), + model: config.model.clone(), + effort: config.model_reasoning_effort, + summary: config.model_reasoning_summary, + } + } +} + +/// Request payload for posting a user turn to a session. +#[derive(Debug, Clone)] +pub struct PostUserTurnRequest { + pub target: RoleOrId, + pub text: String, + pub final_output_json_schema: Option, +} + +/// Identifier used when targeting sessions for cross-session routing. +#[derive(Debug, Clone)] +pub enum RoleOrId { + Session(ConversationId), + RunRole { run_id: String, role: String }, +} + +/// Handle returned by [`CrossSessionHub::post_user_turn`]. +pub struct TurnHandle { + conversation_id: ConversationId, + submission_id: String, + receiver: TokioMutex>>, +} + +impl TurnHandle { + pub fn conversation_id(&self) -> ConversationId { + self.conversation_id + } + + pub fn submission_id(&self) -> &str { + &self.submission_id + } +} + +impl fmt::Debug for TurnHandle { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("TurnHandle") + .field("conversation_id", &self.conversation_id) + .field("submission_id", &self.submission_id) + .finish() + } +} + +/// First assistant message emitted for a bridged turn. +#[derive(Debug, Clone)] +pub struct AssistantMessage { + pub conversation_id: ConversationId, + pub submission_id: String, + pub message: AgentMessageEvent, +} + +/// Wrapper around a session event tagged with its conversation id. +#[derive(Debug, Clone)] +pub struct SessionEvent { + pub conversation_id: ConversationId, + pub event: Event, +} + +/// Stream of [`SessionEvent`] instances for a particular session. +pub struct SessionEventStream { + inner: BroadcastStream, +} + +impl SessionEventStream { + fn new(receiver: broadcast::Receiver) -> Self { + Self { + inner: BroadcastStream::new(receiver), + } + } +} + +impl Stream for SessionEventStream { + type Item = SessionEvent; + + fn poll_next( + mut self: Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + loop { + match Pin::new(&mut self.inner).poll_next(cx) { + std::task::Poll::Ready(Some(Ok(event))) => { + return std::task::Poll::Ready(Some(event)); + } + std::task::Poll::Ready(Some(Err(BroadcastStreamRecvError::Lagged(_)))) => continue, + std::task::Poll::Ready(None) => return std::task::Poll::Ready(None), + std::task::Poll::Pending => return std::task::Poll::Pending, + } + } + } +} + +#[derive(Clone)] +struct RoleKey { + run_id: Arc, + role: Arc, +} + +impl RoleKey { + fn new(run_id: String, role: String) -> Self { + Self { + run_id: Arc::::from(run_id), + role: Arc::::from(role), + } + } +} + +impl PartialEq for RoleKey { + fn eq(&self, other: &Self) -> bool { + self.run_id.as_ref() == other.run_id.as_ref() && self.role.as_ref() == other.role.as_ref() + } +} + +impl Eq for RoleKey {} + +impl std::hash::Hash for RoleKey { + fn hash(&self, state: &mut H) { + std::hash::Hash::hash(self.run_id.as_ref(), state); + std::hash::Hash::hash(self.role.as_ref(), state); + } +} + +struct SessionEntry { + conversation_id: ConversationId, + conversation: Arc, + defaults: SessionDefaults, + role_key: Option, + event_tx: broadcast::Sender, + turn_watchers: TokioMutex>>, + pending_messages: TokioMutex>, + shutdown_tx: StdMutex>>, +} + +impl SessionEntry { + fn new( + conversation_id: ConversationId, + conversation: Arc, + defaults: SessionDefaults, + role_key: Option, + event_tx: broadcast::Sender, + shutdown_tx: oneshot::Sender<()>, + ) -> Self { + Self { + conversation_id, + conversation, + defaults, + role_key, + event_tx, + turn_watchers: TokioMutex::new(HashMap::new()), + pending_messages: TokioMutex::new(HashMap::new()), + shutdown_tx: StdMutex::new(Some(shutdown_tx)), + } + } + + async fn register_waiter( + &self, + submission_id: String, + sender: oneshot::Sender, + ) { + { + let mut watchers = self.turn_watchers.lock().await; + if let Some(message) = { + let mut pending = self.pending_messages.lock().await; + pending.remove(&submission_id) + } { + drop(watchers); + let _ = sender.send(message); + return; + } + watchers.insert(submission_id, sender); + } + } + + async fn notify_assistant_message(&self, message: AssistantMessage) { + let submission_id = message.submission_id.clone(); + let sender_opt = { + let mut watchers = self.turn_watchers.lock().await; + watchers.remove(&submission_id) + }; + + if let Some(sender) = sender_opt { + let _ = sender.send(message); + } else { + let mut pending = self.pending_messages.lock().await; + pending.entry(submission_id).or_insert(message); + } + } + + fn subscribe(&self) -> broadcast::Receiver { + self.event_tx.subscribe() + } + + fn close(&self) { + if let Ok(mut guard) = self.shutdown_tx.lock() + && let Some(tx) = guard.take() + { + let _ = tx.send(()); + } + } + + fn role_key(&self) -> Option { + self.role_key.clone() + } +} + +/// Input for registering a session with the hub. +pub struct SessionRegistration { + pub conversation_id: ConversationId, + pub conversation: Arc, + pub defaults: SessionDefaults, + pub run_id: Option, + pub role: Option, +} + +/// Guard that unregisters the session on drop. +pub struct RegisteredSession { + inner: Arc, + conversation_id: ConversationId, +} + +impl RegisteredSession { + pub fn conversation_id(&self) -> ConversationId { + self.conversation_id + } +} + +impl Drop for RegisteredSession { + fn drop(&mut self) { + self.inner.unregister(self.conversation_id); + } +} + +#[derive(Default)] +struct Inner { + sessions: RwLock>>, + roles: RwLock>, +} + +impl Inner { + fn sessions_read( + &self, + ) -> Result>>, CrossSessionError> + { + self.sessions + .read() + .map_err(|_| CrossSessionError::LockPoisoned("sessions")) + } + + fn sessions_write( + &self, + ) -> Result>>, CrossSessionError> + { + self.sessions + .write() + .map_err(|_| CrossSessionError::LockPoisoned("sessions")) + } + + fn roles_read( + &self, + ) -> Result>, CrossSessionError> { + self.roles + .read() + .map_err(|_| CrossSessionError::LockPoisoned("roles")) + } + + fn roles_write( + &self, + ) -> Result>, CrossSessionError> { + self.roles + .write() + .map_err(|_| CrossSessionError::LockPoisoned("roles")) + } + + fn insert(&self, entry: Arc) -> Result<(), CrossSessionError> { + { + let mut sessions = self.sessions_write()?; + if sessions + .insert(entry.conversation_id, entry.clone()) + .is_some() + { + return Err(CrossSessionError::SessionAlreadyRegistered( + entry.conversation_id, + )); + } + } + + if let Some(role_key) = entry.role_key() { + let mut roles = self.roles_write()?; + if roles.contains_key(&role_key) { + self.sessions_write()?.remove(&entry.conversation_id); + return Err(CrossSessionError::RoleAlreadyRegistered { + run_id: role_key.run_id.to_string(), + role: role_key.role.to_string(), + }); + } + roles.insert(role_key, entry.conversation_id); + } + + Ok(()) + } + + fn unregister(&self, conversation_id: ConversationId) { + if let Some(entry) = self.remove_internal(conversation_id) { + entry.close(); + } + } + + fn remove_internal(&self, conversation_id: ConversationId) -> Option> { + let (entry, role_key) = { + let mut sessions = self.sessions.write().ok()?; + let entry = sessions.remove(&conversation_id)?; + let role_key = entry.role_key(); + (entry, role_key) + }; + + if let Some(role_key) = role_key + && let Ok(mut roles) = self.roles.write() + { + roles.remove(&role_key); + } + + Some(entry) + } + + fn resolve_session( + &self, + conversation_id: ConversationId, + ) -> Result, CrossSessionError> { + self.sessions_read()? + .get(&conversation_id) + .cloned() + .ok_or(CrossSessionError::SessionNotFound(conversation_id)) + } + + fn resolve_target(&self, target: &RoleOrId) -> Result, CrossSessionError> { + match target { + RoleOrId::Session(id) => self.resolve_session(*id), + RoleOrId::RunRole { run_id, role } => { + let conversation_id = { + let roles = self.roles_read()?; + let key = RoleKey::new(run_id.clone(), role.clone()); + roles + .get(&key) + .copied() + .ok_or_else(|| CrossSessionError::RoleNotFound { + run_id: run_id.clone(), + role: role.clone(), + })? + }; + self.resolve_session(conversation_id) + } + } + } +} + +/// Cross-session coordination hub. +#[derive(Default, Clone)] +pub struct CrossSessionHub { + inner: Arc, +} + +impl CrossSessionHub { + pub fn new() -> Self { + Self::default() + } + + pub fn register_session( + &self, + registration: SessionRegistration, + ) -> Result { + let SessionRegistration { + conversation_id, + conversation, + defaults, + run_id, + role, + } = registration; + + let role_key = match (run_id, role) { + (Some(run_id), Some(role)) => Some(RoleKey::new(run_id, role)), + (None, None) => None, + _ => { + return Err(CrossSessionError::IncompleteRoleRegistration); + } + }; + + let (event_tx, _) = broadcast::channel(EVENT_BUFFER_LEN); + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + let entry = Arc::new(SessionEntry::new( + conversation_id, + Arc::clone(&conversation), + defaults, + role_key, + event_tx, + shutdown_tx, + )); + + self.inner.insert(entry.clone())?; + + self.spawn_event_forwarder(entry, conversation, shutdown_rx); + + Ok(RegisteredSession { + inner: Arc::clone(&self.inner), + conversation_id, + }) + } + + pub async fn post_user_turn( + &self, + request: PostUserTurnRequest, + ) -> Result { + let entry = self.inner.resolve_target(&request.target)?; + + let items = vec![InputItem::Text { text: request.text }]; + + let defaults = &entry.defaults; + let submission_id = entry + .conversation + .submit(Op::UserTurn { + items, + cwd: defaults.cwd.clone(), + approval_policy: defaults.approval_policy, + sandbox_policy: defaults.sandbox_policy.clone(), + model: defaults.model.clone(), + effort: defaults.effort, + summary: defaults.summary, + final_output_json_schema: request.final_output_json_schema, + }) + .await + .map_err(CrossSessionError::from)?; + + let (tx, rx) = oneshot::channel(); + + entry.register_waiter(submission_id.clone(), tx).await; + + Ok(TurnHandle { + conversation_id: entry.conversation_id, + submission_id, + receiver: TokioMutex::new(Some(rx)), + }) + } + + pub async fn await_first_assistant( + &self, + handle: &TurnHandle, + timeout: Duration, + ) -> Result { + let receiver = { + let mut guard = handle.receiver.lock().await; + guard.take().ok_or(CrossSessionError::TurnHandleConsumed)? + }; + + match time::timeout(timeout, receiver).await { + Ok(Ok(message)) => Ok(message), + Ok(Err(_)) => Err(CrossSessionError::SessionClosed), + Err(_) => Err(CrossSessionError::AwaitTimeout(timeout)), + } + } + + pub fn stream_events( + &self, + conversation_id: ConversationId, + ) -> Result { + let entry = self.inner.resolve_session(conversation_id)?; + Ok(SessionEventStream::new(entry.subscribe())) + } + + fn spawn_event_forwarder( + &self, + entry: Arc, + conversation: Arc, + mut shutdown_rx: oneshot::Receiver<()>, + ) { + let conversation_id = entry.conversation_id; + let event_tx = entry.event_tx.clone(); + let inner = Arc::clone(&self.inner); + + tokio::spawn(async move { + loop { + tokio::select! { + _ = &mut shutdown_rx => { + debug!("CrossSessionHub received shutdown for session {conversation_id}"); + break; + } + event = conversation.next_event() => { + match event { + Ok(event) => { + if let EventMsg::AgentMessage(agent_message) = &event.msg { + let message = AssistantMessage { + conversation_id, + submission_id: event.id.clone(), + message: agent_message.clone(), + }; + entry.notify_assistant_message(message).await; + } + + if let Err(err) = event_tx.send(SessionEvent { + conversation_id, + event: event.clone(), + }) { + debug!( + "CrossSessionHub dropped event for session {conversation_id}: {err}" + ); + } + + if matches!(event.msg, EventMsg::ShutdownComplete) { + break; + } + } + Err(err) => { + error!("CrossSessionHub event loop error for session {conversation_id}: {err:#?}"); + break; + } + } + } + } + } + + inner.unregister(conversation_id); + }); + } +} + +/// Errors surfaced by cross-session orchestration. +#[derive(thiserror::Error, Debug)] +pub enum CrossSessionError { + #[error("session {0} is already registered with the hub")] + SessionAlreadyRegistered(ConversationId), + #[error("run {run_id} already has a {role} session registered")] + RoleAlreadyRegistered { run_id: String, role: String }, + #[error("session {0} does not exist")] + SessionNotFound(ConversationId), + #[error("no session registered for run {run_id} role {role}")] + RoleNotFound { run_id: String, role: String }, + #[error("session role registration must set both run_id and role")] + IncompleteRoleRegistration, + #[error("turn handle has already been awaited")] + TurnHandleConsumed, + #[error("session closed before an assistant message was emitted")] + SessionClosed, + #[error("timed out waiting {0:?} for assistant response")] + AwaitTimeout(Duration), + #[error("internal lock poisoned: {0}")] + LockPoisoned(&'static str), + #[error("submit failed: {0}")] + SubmitFailed(#[from] CodexErr), +} diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 201d8feb4d..816beeb336 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -13,6 +13,7 @@ mod client; mod client_common; pub mod codex; mod codex_conversation; +pub mod cross_session; pub mod token_data; pub use codex_conversation::CodexConversation; mod command_safety; @@ -51,6 +52,7 @@ mod event_mapping; pub mod review_format; pub use codex_protocol::protocol::InitialHistory; pub use conversation_manager::ConversationManager; +pub use conversation_manager::CrossSessionSpawnParams; pub use conversation_manager::NewConversation; // Re-export common auth types for workspace consumers pub use auth::AuthManager; diff --git a/codex-rs/core/tests/suite/cross_session.rs b/codex-rs/core/tests/suite/cross_session.rs new file mode 100644 index 0000000000..2cc855f9af --- /dev/null +++ b/codex-rs/core/tests/suite/cross_session.rs @@ -0,0 +1,171 @@ +#![cfg(not(target_os = "windows"))] + +use std::sync::Arc; +use std::time::Duration; + +use codex_core::CodexAuth; +use codex_core::ConversationManager; +use codex_core::CrossSessionSpawnParams; +use codex_core::built_in_model_providers; +use codex_core::config::Config; +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 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 tokio_stream::StreamExt; +use wiremock::MockServer; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn cross_session_hub_routes_between_roles() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + + let server = responses::start_mock_server().await; + let bodies = vec![ + responses::sse(vec![ + responses::ev_response_created("solver-resp-1"), + responses::ev_assistant_message("solver-msg-1", "Need direction"), + responses::ev_completed("solver-resp-1"), + ]), + responses::sse(vec![ + responses::ev_response_created("director-resp-1"), + responses::ev_assistant_message("director-msg-1", "Proceed iteratively"), + responses::ev_completed("director-resp-1"), + ]), + responses::sse(vec![ + responses::ev_response_created("solver-resp-2"), + responses::ev_assistant_message("solver-msg-2", "Acknowledged"), + responses::ev_completed("solver-resp-2"), + ]), + ]; + let response_mock = responses::mount_sse_sequence(&server, bodies).await; + + let hub = Arc::new(CrossSessionHub::new()); + let conversation_manager = ConversationManager::with_auth(CodexAuth::from_api_key("dummy-key")); + let run_id = "run-cross-session".to_string(); + + let solver_config = build_config(&server).await?; + let solver = conversation_manager + .new_conversation_with_cross_session( + solver_config, + CrossSessionSpawnParams { + hub: Arc::clone(&hub), + run_id: Some(run_id.clone()), + role: Some("solver".to_string()), + }, + ) + .await?; + + let director_config = build_config(&server).await?; + let director = conversation_manager + .new_conversation_with_cross_session( + director_config, + CrossSessionSpawnParams { + hub: Arc::clone(&hub), + run_id: Some(run_id.clone()), + role: Some("director".to_string()), + }, + ) + .await?; + + let mut solver_events = hub.stream_events(solver.conversation_id)?; + let mut director_events = hub.stream_events(director.conversation_id)?; + + let solver_handle = hub + .post_user_turn(PostUserTurnRequest { + target: RoleOrId::RunRole { + run_id: run_id.clone(), + role: "solver".to_string(), + }, + text: "kick off plan".to_string(), + final_output_json_schema: None, + }) + .await?; + let solver_first = expect_message(&hub, &solver_handle, "Need direction").await?; + + let director_handle = hub + .post_user_turn(PostUserTurnRequest { + target: RoleOrId::RunRole { + run_id: run_id.clone(), + role: "director".to_string(), + }, + text: solver_first.message.message.clone(), + final_output_json_schema: None, + }) + .await?; + let director_first = expect_message(&hub, &director_handle, "Proceed iteratively").await?; + + let solver_followup = hub + .post_user_turn(PostUserTurnRequest { + target: RoleOrId::Session(solver.conversation_id), + text: director_first.message.message.clone(), + final_output_json_schema: None, + }) + .await?; + let solver_reply = expect_message(&hub, &solver_followup, "Acknowledged").await?; + + let solver_event = expect_agent_event(&mut solver_events).await; + match solver_event { + EventMsg::AgentMessage(msg) => assert_eq!(msg.message, "Need direction"), + _ => panic!("expected solver agent message"), + } + + let director_event = expect_agent_event(&mut director_events).await; + match director_event { + EventMsg::AgentMessage(msg) => assert_eq!(msg.message, "Proceed iteratively"), + _ => panic!("expected director agent message"), + } + + assert_eq!(solver_first.message.message, "Need direction"); + assert_eq!(director_first.message.message, "Proceed iteratively"); + assert_eq!(solver_reply.message.message, "Acknowledged"); + assert_eq!(response_mock.requests().len(), 3); + + 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) +} + +async fn expect_message( + hub: &CrossSessionHub, + handle: &codex_core::cross_session::TurnHandle, + expected: &str, +) -> anyhow::Result { + let message = hub + .await_first_assistant(handle, Duration::from_secs(1)) + .await?; + assert_eq!(message.message.message, expected); + Ok(message) +} + +async fn expect_agent_event(stream: &mut SessionEventStream) -> EventMsg { + loop { + let maybe_event = match tokio::time::timeout(Duration::from_secs(1), stream.next()).await { + Ok(event) => event, + Err(_) => panic!("event timeout"), + }; + + if let Some(event) = maybe_event { + let msg = event.event.msg; + if matches!(msg, EventMsg::AgentMessage(_)) { + return msg; + } + } else { + panic!("stream ended before agent message"); + } + } +} diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index 6008811dea..af9f6844ec 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -6,6 +6,7 @@ mod cli_stream; mod client; mod compact; mod compact_resume_fork; +mod cross_session; mod exec; mod exec_stream_events; mod fork_conversation; diff --git a/codex-rs/infty.md b/codex-rs/infty.md new file mode 100644 index 0000000000..a652fdf84a --- /dev/null +++ b/codex-rs/infty.md @@ -0,0 +1,189 @@ +# Codex Infty: Ultra‑Long Task Orchestration + +Design a clean, extensible way to run arbitrarily long tasks (hours–days) with bounded model context, autonomous continuation, and robust correctness review. Works for code and non‑code. + +Status: Proposed • Scope: New crates using `codex-core` • Compatibility: Non‑breaking + +--- + +## 1) Motivation +- Context windows are limited → we must compact and retrieve. +- Models pause/ask for permission → we must self‑direct. +- No systematic review → we must verify before returning. + +## 2) Approach (High‑Level) +Run three coordinated roles as independent `codex-core` sessions. Reuse existing tools (shell, apply_patch, read_file, list_dir, grep_files) for persistence and retrieval. Add one clean, first‑class cross‑session facility in core for direction/verification — orchestrator‑driven, no model‑visible tool. + +- Solver (Model A): executes plan; writes all results to memory/artifacts; never asks humans to continue. +- Director (Model B): answers Solver’s direction questions and re‑plans when needed. +- Verifier (Model C…Cₙ): evaluates completion claims; returns pass/fail with structured feedback. + +Inter‑role coordination uses a built‑in CrossSessionHub in core. The orchestrator watches assistant messages and bridges them as user turns to the peer role. + +## 3) Architecture +``` +┌────────────────────────────┐ +│ codex-infty │ +│ Orchestrator + CLI/Lib │ +│ - spawns 3 codex sessions │ +│ - supervises long runs │ +│ - configures Run/Role │ +└────────────┬───────────────┘ + │ + ┌─────────▼─────────┐ ┌─────────▼─────────┐ + │ Solver (A) │ │ Director (B) │ + │ codex-core session│ │ codex-core session│ + └─────────┬─────────┘ └─────────┬─────────┘ + │ │ + └──────────┬──────────────┘ + │ + ┌───────▼────────┐ + │ Verifier(s) (C)│ + │ codex-core sess │ + └───────┬─────────┘ + │ + CrossSessionHub (core, orchestrator‑driven) + JSONL rollouts, auto‑compaction +``` + +### Components +- codex-infty (new crate) + - Spawns/owns three `codex-core` sessions (A/B/C) with role‑specific base instructions. + - Supervises progress over very long runs. + - Defines a simple on‑disk Run Store that the models write to using existing tools. + - Configures sessions with Run/Role metadata (for cross‑session routing). +- codex-core (existing, with one addition) + - Reuse streaming, tool routing, JSONL rollouts with resume, auto‑compaction, and existing tools: + - `apply_patch`, `shell`/`exec_command`/`write_stdin` + - `grep_files`, `read_file`, `list_dir` (enable via model family/experimental tools) + - New: built‑in `CrossSessionHub` for intra‑process routing (§5). No new model tool is exposed. + +## 4) Data Model (Durable) and Filesystem Layout +Persist everything in a Run Store directory; models read/write using existing tools. + +- Run Store layout (example under `~/.codex/infty//`): + - `artifacts/` – blobs and text outputs (models can create via `apply_patch` for text; `shell` for binary moves/copies). + - `memory/` – JSON/Markdown notes: facts, hypotheses, plans, decisions, claims, evidence, evaluations. + - `index/` – optional search/index artifacts (built out‑of‑band by orchestrator jobs; models can still use `grep_files`). + +Data is append‑only by convention; items link to each other via ids/paths stored in JSON. + +## 5) New Core API: CrossSessionHub (no model tool) +Add a core facility that lets the orchestrator bridge assistant messages between sessions by posting user turns. + +### 5.1 Hub API +- Registry that maps `{ run_id, role } -> session handle` and `{ session_id } -> session handle`. +- Sessions register on spawn with `run_id` and `role`; unregister on drop. +- Expose async methods for the orchestrator: + - `post_user_turn(to: RoleOrId, text: String) -> TurnHandle` – inject a `UserTurn` as if typed by a user. + - `await_first_assistant(turn: &TurnHandle, timeout: Duration) -> AssistantMessage` – wait until the first assistant message for that turn. + - `stream_events(session_id) -> impl Stream` – optional subscription for higher‑level orchestration. + +### 5.2 Orchestrator Bridge Logic +- Direction: when the Solver emits an assistant message asking for permission/direction, the orchestrator forwards that assistant text verbatim as a user turn to the Director and waits for the Director’s first assistant reply; it then posts that reply as a user turn to Solver. +- Verification: when Solver requests verification, orchestrator forwards request to Verifier(s); structured verdicts (pass/fail/reasons/suggestions) flow back. +- Persistence: Each session persists its own events to rollout; the orchestrator just routes. + +## 6) Run Store Facilities +- Memory notes follow JSON schemas per role (plans, claims, evidence). +- Artifacts include code patches, logs, compiled binaries, docs. Use naming convention `-.`. +- Orchestrator can create `index/` entries (e.g., embeddings) offline; models still access via standard tools. + +## 7) Orchestrator Flow +1. Initialize Run Store + metadata (objective, roles, options). +2. Spawn Solver, Director, Verifier sessions via `CrossSessionHub`. +3. Seed objective as Solver user turn; monitor outputs. +4. Relay direction/verification messages automatically between roles. +5. Trigger periodic checkpoints (copy artifacts/memory to dated snapshots). +6. On completion, ensure Verifier returns pass, then emit final deliverable path. +7. Support resume: reload Run Store, respawn sessions with `InitialHistory::Resumed`. + +## 8) Context Management +- Conversational context: rely on `codex-core` auto‑compaction. +- Long‑term memory: persist facts/results as files; retrieve with `grep_files`/`read_file`/`list_dir`. +- Run Store snapshots allow cold resume even after orchestrator restart. + +## 9) Verification Strategies +- Code: tests, linters, type checks via `shell` under sandbox. +- Text: grader rubrics, citation/contradiction checks. +- Math/research: multi‑verifier consensus, self‑consistency, proof‑sketch validation. + +## 10) Security & Policy +- All execution stays under `codex-core` sandbox/approval. +- Memory/Artifact tools are pure data I/O (no code execution). +- Inter‑role calls run in isolated sessions. + +## 11) MVP (Phased) +1. codex-core + - Add `CrossSessionHub` with registration and post/await APIs. + - Add `run_id` and `role` registration on session spawn (optional fields). + - Tests: two sessions in a run; orchestrator posts user text to Director and bridges reply to Solver. +2. codex-infty + - Orchestrator lib + CLI: create Run Store directories, spawn A/B/C sessions with `run_id`/`role`, run loop; ship role prompts. Enable `grep_files`/`read_file`/`list_dir`. +3. Verification + - Use `shell` to run checks/tests when applicable; use Verifier sessions for rubric‑based judgments. + +## 12) Finalization & Extensibility +- Finalization workflow (after `verdict == pass`): the orchestrator issues a final `UserTurn` to the Solver instructing: + - Create a clean `deliverable/` folder under the Run Store. + - Copy/transform only the necessary end results; remove scratch artifacts. + - Write a `deliverable/README.md` including: overview, contents manifest with paths and sizes, verification steps (how to run tests), and any limitations. + - Summarize the work in the final assistant message and return the path to `deliverable/`. + +- Extensibility: + - Pluggable `IndexStrategy` (keyword/embeddings/hybrid) built by the orchestrator (models still query via `grep_files`). + - Multiple Verifiers with majority/weighted consensus. + - Future: broadcast/multicast cross‑session calls (e.g., ask three verifiers and aggregate). + +## 13) Why This Solves The Three Problems +- Context: conversational compaction + durable memory with retrieval. +- Pauses: assistant questions are bridged to a Director; the orchestrator backstops. +- Review: Solver’s verification request is bridged to Verifier(s) with structured verdicts and remediation. + +This keeps `codex-core` focused and leverages its strengths (streaming, tools, compaction, rollouts) while adding a small, clean cross‑session primitive to enable arbitrarily long, autonomous runs across domains. + +--- + +## 14) End‑to‑End Example (Minimal) + +Assume a run folder at `~/.codex/infty/run_123/`. + +1) User objective → Solver (UserTurn) +- User: "Write a tiny CLI that prints Fibonacci numbers and provide usage docs." + +2) Solver starts +- Tool: `update_plan` → steps: parse request; scaffold CLI; implement logic; write docs; verify; finalize deliverable. +- Tool: `grep_files` → searches `artifacts/` and repo for prior art. + +3) Solver seeks direction +- Solver’s assistant message: “Confirm plan: binary in ./fib, args: N, output first N Fibonacci numbers; docs in memory/docs.md?” +- Orchestrator posts a UserTurn to Director with that question and sets `final_output_json_schema` to the Director schema (strict). +- Director’s first assistant message: +```json +{ "directive": "Proceed. Add tests under memory/tests.md; prefer iterative impl; expose --limit flag.", "rationale": "Keeps stack small; eases verification." } +``` +- Orchestrator posts that reply as a UserTurn to Solver; Solver continues. + +4) Solver implements +- Tool: `apply_patch` → creates `artifacts/fib.rs` and a small Cargo bin, or shell scaffolding. +- Tool: `shell` → `cargo run -- 10` to sanity check (under sandbox). +- Writes `memory/docs.md` and `memory/tests.md`. + +5) Solver claims completion +- Writes `memory/claims/cli.json` (per template) referencing artifacts and docs. +- Solver’s assistant message: “Please verify claim in memory/claims/cli.json with artifacts/fib.rs; run cargo test if present.” +- Orchestrator posts a UserTurn to Verifier with `final_output_json_schema` set to the Verifier schema (strict). Verifier runs checks (via `shell`), returns: +```json +{ "verdict": "fail", "reasons": ["No tests"], "suggestions": ["Add tests covering N=1,2,10"] } +``` +- Orchestrator posts that reply as a UserTurn to Solver; Solver iterates (adds tests, fixes issues). + +6) Pass and finalize +- Verifier returns `{ "verdict": "pass", … }`. +- Orchestrator issues finalization UserTurn to Solver: + - "Create deliverable/: include compiled bin or script, usage docs, and tests; write deliverable/README.md with run instructions; remove scratch files." +- Solver: + - Tool: `shell`/`apply_patch` → builds `deliverable/` with README and artifacts. + - Assistant message: "Deliverable ready at ~/.codex/infty/run_123/deliverable/." + +7) Orchestrator returns the final path to the user.