V1

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]]

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",

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 }

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<PathBuf>,
+
+    #[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<String>,
+
+    /// Optional objective to send to the solver immediately after creation.
+    #[arg(long)]
+    objective: Option<String>,
+
+    /// 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<String>,
+}
+
+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<PathBuf>,
+    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<PathBuf>, 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<PathBuf>, 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<PathBuf>,
+    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<Config> {
+    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<CodexAuth> {
+    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<PathBuf>) -> Result<PathBuf> {
+    if let Some(path) = override_path {
+        return Ok(path);
+    }
+    Ok(codex_infty::default_runs_root()?)
+}
+
+fn collect_run_summaries(root: &Path) -> Result<Vec<RunSummary>> {
+    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());
+    }
+}

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<PathBuf>) -> 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(

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 }

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<PathBuf>,
+}
+
+impl RoleConfig {
+    pub fn new(role: impl Into<String>, config: Config) -> Self {
+        Self {
+            role: role.into(),
+            config,
+            config_path: None,
+        }
+    }
+
+    pub fn with_path(role: impl Into<String>, 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<PathBuf>,
+    pub solver: RoleConfig,
+    pub director: RoleConfig,
+    pub verifiers: Vec<RoleConfig>,
+}
+
+pub struct ResumeParams {
+    pub run_path: PathBuf,
+    pub solver: RoleConfig,
+    pub director: RoleConfig,
+    pub verifiers: Vec<RoleConfig>,
+}
+
+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\":\"<path>\",\"summary\":\"<summary>\"}.";
+
+#[derive(Clone)]
+pub struct RunExecutionOptions {
+    pub objective: Option<String>,
+    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<String>,
+    pub raw_message: String,
+}
+
+pub struct RoleSession {
+    pub role: String,
+    pub conversation_id: ConversationId,
+    pub conversation: Arc<CodexConversation>,
+    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<String>,
+    },
+    FinalDelivery {
+        deliverable_path: String,
+        #[serde(default)]
+        summary: Option<String>,
+    },
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+struct DirectiveResponse {
+    directive: String,
+    #[serde(default)]
+    rationale: Option<String>,
+}
+
+#[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<String>,
+    #[serde(default)]
+    suggestions: Vec<String>,
+}
+
+#[derive(Debug, Serialize)]
+struct VerifierReport {
+    role: String,
+    verdict: VerifierDecision,
+    #[serde(default)]
+    reasons: Vec<String>,
+    #[serde(default)]
+    suggestions: Vec<String>,
+}
+
+#[derive(Debug, Serialize)]
+struct AggregatedVerifierVerdict {
+    #[serde(rename = "type")]
+    kind: &'static str,
+    overall: VerifierDecision,
+    verdicts: Vec<VerifierReport>,
+}
+
+struct SessionCleanup {
+    conversation_id: ConversationId,
+    conversation: Arc<CodexConversation>,
+}
+
+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<RoleSession>,
+    pub store: RunStore,
+}
+
+pub struct InftyOrchestrator {
+    hub: Arc<CrossSessionHub>,
+    conversation_manager: ConversationManager,
+    runs_root: PathBuf,
+}
+
+impl InftyOrchestrator {
+    pub fn new(auth: CodexAuth) -> Result<Self> {
+        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<PathBuf>) -> 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<CrossSessionHub> {
+        Arc::clone(&self.hub)
+    }
+
+    pub async fn execute_new_run(
+        &self,
+        params: RunParams,
+        options: RunExecutionOptions,
+    ) -> Result<RunOutcome> {
+        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<RunOutcome> {
+        let sessions = self.resume_run(params).await?;
+        self.drive_run(sessions, options).await
+    }
+
+    pub async fn spawn_run(&self, params: RunParams) -> Result<RunSessions> {
+        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<RunSessions> {
+        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<RunOutcome> {
+        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<RunOutcome> {
+        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<bool> {
+        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<SessionCleanup>, 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<SessionCleanup>) {
+        self.shutdown_sessions(sessions).await;
+    }
+
+    async fn shutdown_sessions(&self, sessions: Vec<SessionCleanup>) {
+        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<String>,
+        final_output_json_schema: Option<Value>,
+    ) -> Result<codex_core::cross_session::TurnHandle> {
+        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<AssistantMessage> {
+        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<String>,
+        timeout: Duration,
+        final_output_json_schema: Option<Value>,
+    ) -> Result<AssistantMessage> {
+        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<Value>,
+    ) -> Result<AssistantMessage> {
+        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<SessionEventStream, codex_core::cross_session::CrossSessionError> {
+        self.hub.stream_events(conversation_id)
+    }
+
+    async fn spawn_role_session(
+        &self,
+        run_id: &str,
+        run_path: &Path,
+        role_config: RoleConfig,
+    ) -> Result<RoleSession> {
+        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<RoleSession> {
+        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<RoleMetadata> {
+    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<RoleMetadata> {
+    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<SolverSignal> {
+    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<T>(message: &str) -> Result<T>
+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::<T>()))
+}
+
+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<PathBuf> {
+    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<SessionCleanup> {
+    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<PathBuf> {
+    let home = home_dir().ok_or_else(|| anyhow!("failed to determine home directory"))?;
+    Ok(home.join(".codex").join("infty"))
+}

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":"<concise question or decision>"}
+  ```
+- Verification request (send to Verifier):
+  ```json
+  {"type":"verification_request","claim_path":"memory/claims/<file>.json","notes":"<optional context>"}
+  ```
+- Final delivery (after receiving the finalization instruction):
+  ```json
+  {"type":"final_delivery","deliverable_path":"deliverable","summary":"<one paragraph>"}
+  ```
+
+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":"<go/no-go decision or next step>","rationale":"<why this is the right move>"}
+```
+
+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"));
+    }
+}

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<PathBuf>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub config_path: Option<PathBuf>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RunMetadata {
+    pub run_id: String,
+    pub created_at: DateTime<Utc>,
+    pub updated_at: DateTime<Utc>,
+    pub roles: Vec<RoleMetadata>,
+}
+
+#[derive(Debug, Clone)]
+pub struct RunStore {
+    path: PathBuf,
+    metadata: RunMetadata,
+}
+
+impl RunStore {
+    pub fn initialize(
+        run_path: impl AsRef<Path>,
+        run_id: &str,
+        roles: &[RoleMetadata],
+    ) -> Result<Self> {
+        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<Path>) -> Result<Self> {
+        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);
+    }
+}

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<Config> {
+    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)
+}

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"] }

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<CrossSessionHub>,
+    pub run_id: Option<String>,
+    pub role: Option<String>,
+}
+
 /// [`ConversationManager`] is responsible for creating conversations and
 /// maintaining them in memory.
 pub struct ConversationManager {
     conversations: Arc<RwLock<HashMap<ConversationId, Arc<CodexConversation>>>>,
+    cross_session_registrations: Arc<RwLock<HashMap<ConversationId, RegisteredSession>>>,
     auth_manager: Arc<AuthManager>,
     session_source: SessionSource,
 }
@@ -43,6 +57,7 @@ impl ConversationManager {
     pub fn new(auth_manager: Arc<AuthManager>, 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<NewConversation> {
-        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<NewConversation> {
+        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<AuthManager>,
+        initial_history: InitialHistory,
+        cross_session: Option<CrossSessionSpawnParams>,
     ) -> CodexResult<NewConversation> {
+        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<CodexConversation>,
+    ) -> 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<CodexConversation>,
+    ) {
+        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<AuthManager>,
     ) -> CodexResult<NewConversation> {
         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<AuthManager>,
+        params: CrossSessionSpawnParams,
+    ) -> CodexResult<NewConversation> {
+        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<NewConversation> {
+        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<Arc<CodexConversation>> {
+        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<NewConversation> {
+        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
     }
 }
 

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<ReasoningEffortConfig>,
+    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<Value>,
+}
+
+/// 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<Option<oneshot::Receiver<AssistantMessage>>>,
+}
+
+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<SessionEvent>,
+}
+
+impl SessionEventStream {
+    fn new(receiver: broadcast::Receiver<SessionEvent>) -> 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<Option<Self::Item>> {
+        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<str>,
+    role: Arc<str>,
+}
+
+impl RoleKey {
+    fn new(run_id: String, role: String) -> Self {
+        Self {
+            run_id: Arc::<str>::from(run_id),
+            role: Arc::<str>::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<H: std::hash::Hasher>(&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<CodexConversation>,
+    defaults: SessionDefaults,
+    role_key: Option<RoleKey>,
+    event_tx: broadcast::Sender<SessionEvent>,
+    turn_watchers: TokioMutex<HashMap<String, oneshot::Sender<AssistantMessage>>>,
+    pending_messages: TokioMutex<HashMap<String, AssistantMessage>>,
+    shutdown_tx: StdMutex<Option<oneshot::Sender<()>>>,
+}
+
+impl SessionEntry {
+    fn new(
+        conversation_id: ConversationId,
+        conversation: Arc<CodexConversation>,
+        defaults: SessionDefaults,
+        role_key: Option<RoleKey>,
+        event_tx: broadcast::Sender<SessionEvent>,
+        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<AssistantMessage>,
+    ) {
+        {
+            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<SessionEvent> {
+        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<RoleKey> {
+        self.role_key.clone()
+    }
+}
+
+/// Input for registering a session with the hub.
+pub struct SessionRegistration {
+    pub conversation_id: ConversationId,
+    pub conversation: Arc<CodexConversation>,
+    pub defaults: SessionDefaults,
+    pub run_id: Option<String>,
+    pub role: Option<String>,
+}
+
+/// Guard that unregisters the session on drop.
+pub struct RegisteredSession {
+    inner: Arc<Inner>,
+    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<HashMap<ConversationId, Arc<SessionEntry>>>,
+    roles: RwLock<HashMap<RoleKey, ConversationId>>,
+}
+
+impl Inner {
+    fn sessions_read(
+        &self,
+    ) -> Result<RwLockReadGuard<'_, HashMap<ConversationId, Arc<SessionEntry>>>, CrossSessionError>
+    {
+        self.sessions
+            .read()
+            .map_err(|_| CrossSessionError::LockPoisoned("sessions"))
+    }
+
+    fn sessions_write(
+        &self,
+    ) -> Result<RwLockWriteGuard<'_, HashMap<ConversationId, Arc<SessionEntry>>>, CrossSessionError>
+    {
+        self.sessions
+            .write()
+            .map_err(|_| CrossSessionError::LockPoisoned("sessions"))
+    }
+
+    fn roles_read(
+        &self,
+    ) -> Result<RwLockReadGuard<'_, HashMap<RoleKey, ConversationId>>, CrossSessionError> {
+        self.roles
+            .read()
+            .map_err(|_| CrossSessionError::LockPoisoned("roles"))
+    }
+
+    fn roles_write(
+        &self,
+    ) -> Result<RwLockWriteGuard<'_, HashMap<RoleKey, ConversationId>>, CrossSessionError> {
+        self.roles
+            .write()
+            .map_err(|_| CrossSessionError::LockPoisoned("roles"))
+    }
+
+    fn insert(&self, entry: Arc<SessionEntry>) -> 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<Arc<SessionEntry>> {
+        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<Arc<SessionEntry>, CrossSessionError> {
+        self.sessions_read()?
+            .get(&conversation_id)
+            .cloned()
+            .ok_or(CrossSessionError::SessionNotFound(conversation_id))
+    }
+
+    fn resolve_target(&self, target: &RoleOrId) -> Result<Arc<SessionEntry>, 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<Inner>,
+}
+
+impl CrossSessionHub {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    pub fn register_session(
+        &self,
+        registration: SessionRegistration,
+    ) -> Result<RegisteredSession, CrossSessionError> {
+        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<TurnHandle, CrossSessionError> {
+        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<AssistantMessage, CrossSessionError> {
+        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<SessionEventStream, CrossSessionError> {
+        let entry = self.inner.resolve_session(conversation_id)?;
+        Ok(SessionEventStream::new(entry.subscribe()))
+    }
+
+    fn spawn_event_forwarder(
+        &self,
+        entry: Arc<SessionEntry>,
+        conversation: Arc<CodexConversation>,
+        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),
+}

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;

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<Config> {
+    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<AssistantMessage> {
+    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");
+        }
+    }
+}

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;

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/<run-id>/`):
+  - `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<Item = Event>` – 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 `<timestamp>-<summary>.<ext>`.
+- 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.