diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index cb9f9399ec1..73392794c5b 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -36,6 +36,7 @@ use codex_tui::Cli as TuiCli; use codex_tui::ExitReason; use codex_tui::UpdateAction; use codex_utils_absolute_path::AbsolutePathBuf; +use codex_utils_absolute_path::canonicalize_existing_preserving_symlinks; use codex_utils_cli::CliConfigOverrides; use codex_utils_cli::ProfileV2Name; use codex_utils_cli::SharedCliOptions; @@ -83,6 +84,7 @@ use codex_login::read_codex_access_token_from_env; use codex_memories_write::clear_memory_roots_contents; use codex_models_manager::bundled_models_response; use codex_models_manager::manager::RefreshStrategy; +use codex_protocol::models::BUILT_IN_PERMISSION_PROFILE_WORKSPACE; use codex_protocol::protocol::AskForApproval; use codex_protocol::user_input::UserInput; use codex_terminal_detection::TerminalName; @@ -1854,6 +1856,9 @@ async fn run_debug_prompt_input_command( ) -> anyhow::Result<()> { let loader_overrides = loader_overrides_for_profile(interactive.config_profile_v2.as_ref())?; let shared = interactive.shared.into_inner(); + shared + .validate_workspace_root_mode() + .map_err(anyhow::Error::msg)?; let mut cli_kv_overrides = root_config_overrides .parse_overrides() .map_err(anyhow::Error::msg)?; @@ -1874,10 +1879,34 @@ async fn run_debug_prompt_input_command( } else { shared.sandbox_mode.map(Into::into) }; + let workspace_base = match shared.cwd.as_deref() { + Some(path) => { + AbsolutePathBuf::from_absolute_path(canonicalize_existing_preserving_symlinks(path)?)? + } + None => AbsolutePathBuf::current_dir()?, + }; + let workspace_roots = (!shared.workspace_root.is_empty()).then(|| { + shared + .workspace_root + .iter() + .cloned() + .map(|path| AbsolutePathBuf::resolve_path_against_base(path, workspace_base.as_path())) + .collect() + }); + let exact_workspace_profile = workspace_roots.as_ref().is_some_and(|_| { + sandbox_mode == Some(codex_protocol::config_types::SandboxMode::WorkspaceWrite) + }); + let sandbox_mode_override = if exact_workspace_profile { + None + } else { + sandbox_mode + }; let overrides = ConfigOverrides { model: shared.model, approval_policy, - sandbox_mode, + sandbox_mode: sandbox_mode_override, + default_permissions: exact_workspace_profile + .then(|| BUILT_IN_PERMISSION_PROFILE_WORKSPACE.to_string()), cwd: shared.cwd, codex_self_exe: arg0_paths.codex_self_exe, codex_linux_sandbox_exe: arg0_paths.codex_linux_sandbox_exe, @@ -1886,6 +1915,7 @@ async fn run_debug_prompt_input_command( ephemeral: Some(true), bypass_hook_trust: shared.bypass_hook_trust.then_some(true), additional_writable_roots: shared.add_dir, + workspace_roots, ..Default::default() }; let config = ConfigBuilder::default() diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index 6e14ccfba98..97cdc3de8f6 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -4682,6 +4682,47 @@ async fn add_dir_override_extends_workspace_writable_roots() -> std::io::Result< Ok(()) } +#[tokio::test] +async fn explicit_workspace_roots_replace_cwd_for_workspace_write() -> std::io::Result<()> { + let temp_dir = tempfile::tempdir_in(std::env::current_dir()?)?; + let workspace = temp_dir.path().join("workspace"); + let tenant = temp_dir.path().join("tenant"); + let devkit = temp_dir.path().join("devkit"); + std::fs::create_dir_all(&workspace)?; + std::fs::create_dir_all(&tenant)?; + std::fs::create_dir_all(&devkit)?; + + let tenant_abs = tenant.abs(); + let devkit_abs = devkit.abs(); + let overrides = ConfigOverrides { + cwd: Some(workspace.clone()), + default_permissions: Some(BUILT_IN_PERMISSION_PROFILE_WORKSPACE.to_string()), + workspace_roots: Some(vec![tenant_abs.clone(), devkit_abs.clone()]), + ..Default::default() + }; + + let config = Config::load_from_base_config_with_overrides( + ConfigToml::default(), + overrides, + temp_dir.path().abs(), + ) + .await?; + + assert_eq!( + config.workspace_roots, + vec![tenant_abs.clone(), devkit_abs.clone()] + ); + let policy = config.permissions.file_system_sandbox_policy(); + assert!(policy.can_write_path_with_cwd(tenant_abs.as_path(), &workspace)); + assert!(policy.can_write_path_with_cwd(devkit_abs.as_path(), &workspace)); + assert!( + !policy.can_write_path_with_cwd(&workspace, &workspace), + "workspace cwd should remain read-only: {policy:#?}" + ); + + Ok(()) +} + #[tokio::test] async fn default_zsh_path_sets_runtime_zsh_path() -> std::io::Result<()> { let codex_home = TempDir::new()?; diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index b5660c6686f..f7b3b4b0ded 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -2831,7 +2831,7 @@ async fn session_configured_reports_permission_profile_for_external_sandbox() -> } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn session_permission_profile_rebinds_runtime_workspace_roots() -> anyhow::Result<()> { +async fn session_permission_profile_rebinds_exact_runtime_workspace_roots() -> anyhow::Result<()> { let codex_home = tempfile::TempDir::new()?; let cwd = tempfile::TempDir::new()?; let old_root = test_path_buf("/workspace/old").abs(); @@ -2841,11 +2841,12 @@ async fn session_permission_profile_rebinds_runtime_workspace_roots() -> anyhow: .harness_overrides(crate::config::ConfigOverrides { cwd: Some(cwd.path().to_path_buf()), default_permissions: Some(BUILT_IN_PERMISSION_PROFILE_WORKSPACE.to_string()), - additional_writable_roots: vec![old_root.to_path_buf()], + workspace_roots: Some(vec![old_root.clone()]), ..Default::default() }) .build() .await?; + assert_eq!(config.workspace_roots, vec![old_root.clone()]); let session_permission_profile_state = session_permission_profile_state_from_config(&config)?; let stored_file_system_policy = session_permission_profile_state diff --git a/codex-rs/exec/src/lib.rs b/codex-rs/exec/src/lib.rs index 63868bd242f..f3eb1c26f86 100644 --- a/codex-rs/exec/src/lib.rs +++ b/codex-rs/exec/src/lib.rs @@ -85,6 +85,7 @@ use codex_protocol::ThreadId; use codex_protocol::config_types::ApprovalsReviewer; use codex_protocol::config_types::SandboxMode; use codex_protocol::models::ActivePermissionProfile; +use codex_protocol::models::BUILT_IN_PERMISSION_PROFILE_WORKSPACE; use codex_protocol::models::PermissionProfile; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::ReviewRequest; @@ -268,6 +269,9 @@ pub async fn run_main(cli: Cli, arg0_paths: Arg0DispatchPaths) -> anyhow::Result config_overrides, } = cli; let shared = shared.into_inner(); + shared + .validate_workspace_root_mode() + .map_err(anyhow::Error::msg)?; let SharedCliOptions { images, model: model_cli_arg, @@ -280,6 +284,7 @@ pub async fn run_main(cli: Cli, arg0_paths: Arg0DispatchPaths) -> anyhow::Result bypass_hook_trust, cwd, add_dir, + workspace_root, } = shared; let (_stdout_with_ansi, stderr_with_ansi) = match color { @@ -320,6 +325,20 @@ pub async fn run_main(cli: Cli, arg0_paths: Arg0DispatchPaths) -> anyhow::Result } None => AbsolutePathBuf::current_dir()?, }; + let workspace_roots = (!workspace_root.is_empty()).then(|| { + workspace_root + .into_iter() + .map(|path| AbsolutePathBuf::resolve_path_against_base(path, config_cwd.as_path())) + .collect() + }); + let exact_workspace_profile = workspace_roots + .as_ref() + .is_some_and(|_| sandbox_mode == Some(SandboxMode::WorkspaceWrite)); + let sandbox_mode_override = if exact_workspace_profile { + None + } else { + sandbox_mode + }; // we load config.toml here to determine project state. #[allow(clippy::print_stderr)] @@ -426,11 +445,12 @@ pub async fn run_main(cli: Cli, arg0_paths: Arg0DispatchPaths) -> anyhow::Result // the fully resolved reviewer is AutoReview. approval_policy: Some(AskForApproval::Never), approvals_reviewer: None, - sandbox_mode, + sandbox_mode: sandbox_mode_override, permission_profile: None, - default_permissions: None, + default_permissions: exact_workspace_profile + .then(|| BUILT_IN_PERMISSION_PROFILE_WORKSPACE.to_string()), cwd: resolved_cwd, - workspace_roots: None, + workspace_roots, model_provider: model_provider.clone(), service_tier: None, codex_self_exe: arg0_paths.codex_self_exe.clone(), diff --git a/codex-rs/prompts/src/permissions_instructions_tests.rs b/codex-rs/prompts/src/permissions_instructions_tests.rs index 580e9781ce7..ec8f59dacf1 100644 --- a/codex-rs/prompts/src/permissions_instructions_tests.rs +++ b/codex-rs/prompts/src/permissions_instructions_tests.rs @@ -14,7 +14,7 @@ use std::path::PathBuf; fn renders_sandbox_mode_text() { assert_eq!( sandbox_text(SandboxMode::WorkspaceWrite, NetworkAccess::Restricted), - "Filesystem sandboxing defines which files can be read or written. `sandbox_mode` is `workspace-write`: The sandbox permits reading files, and editing files in `cwd` and `writable_roots`. Editing files in other directories requires approval. Network access is restricted." + "Filesystem sandboxing defines which files can be read or written. `sandbox_mode` is `workspace-write`: The sandbox permits reading files and editing files only within the writable roots listed below. Editing files in other directories requires approval. Network access is restricted." ); assert_eq!( diff --git a/codex-rs/prompts/templates/permissions/sandbox_mode/workspace_write.md b/codex-rs/prompts/templates/permissions/sandbox_mode/workspace_write.md index 0732a998e89..69bfeba86b4 100644 --- a/codex-rs/prompts/templates/permissions/sandbox_mode/workspace_write.md +++ b/codex-rs/prompts/templates/permissions/sandbox_mode/workspace_write.md @@ -1 +1 @@ -Filesystem sandboxing defines which files can be read or written. `sandbox_mode` is `workspace-write`: The sandbox permits reading files, and editing files in `cwd` and `writable_roots`. Editing files in other directories requires approval. Network access is {{network_access}}. +Filesystem sandboxing defines which files can be read or written. `sandbox_mode` is `workspace-write`: The sandbox permits reading files and editing files only within the writable roots listed below. Editing files in other directories requires approval. Network access is {{network_access}}. diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index 4c33b82246b..4627588d21d 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -55,6 +55,7 @@ use codex_protocol::config_types::AltScreenMode; use codex_protocol::config_types::SandboxMode; #[cfg(target_os = "windows")] use codex_protocol::config_types::WindowsSandboxLevel; +use codex_protocol::models::BUILT_IN_PERMISSION_PROFILE_WORKSPACE; use codex_rollout::StateDbHandle; use codex_rollout::state_db; use codex_state::log_db; @@ -915,6 +916,8 @@ pub async fn run_main( loader_overrides: LoaderOverrides, explicit_remote_endpoint: Option, ) -> std::io::Result { + cli.validate_workspace_root_mode() + .map_err(|message| std::io::Error::new(std::io::ErrorKind::InvalidInput, message))?; let strict_config = cli.strict_config; let (sandbox_mode, approval_policy) = if cli.dangerously_bypass_approvals_and_sandbox { ( @@ -1107,11 +1110,37 @@ pub async fn run_main( }; let additional_dirs = cli.add_dir.clone(); + let workspace_roots = if cli.workspace_root.is_empty() { + None + } else { + let workspace_base = config_cwd.as_ref().ok_or_else(|| { + std::io::Error::other("--workspace-root is unavailable for remote workspaces") + })?; + Some( + cli.workspace_root + .iter() + .cloned() + .map(|path| { + AbsolutePathBuf::resolve_path_against_base(path, workspace_base.as_path()) + }) + .collect(), + ) + }; + let exact_workspace_profile = workspace_roots + .as_ref() + .is_some_and(|_| sandbox_mode == Some(SandboxMode::WorkspaceWrite)); + let sandbox_mode_override = if exact_workspace_profile { + None + } else { + sandbox_mode + }; let overrides = ConfigOverrides { model, approval_policy, - sandbox_mode, + sandbox_mode: sandbox_mode_override, + default_permissions: exact_workspace_profile + .then(|| BUILT_IN_PERMISSION_PROFILE_WORKSPACE.to_string()), cwd: cwd_override, model_provider: model_provider_override.clone(), codex_self_exe: arg0_paths.codex_self_exe.clone(), @@ -1120,6 +1149,7 @@ pub async fn run_main( show_raw_agent_reasoning: cli.oss.then_some(true), bypass_hook_trust: cli.bypass_hook_trust.then_some(true), additional_writable_roots: additional_dirs, + workspace_roots, ..Default::default() }; diff --git a/codex-rs/utils/cli/src/shared_options.rs b/codex-rs/utils/cli/src/shared_options.rs index 97ae12da9e3..37f189b20cf 100644 --- a/codex-rs/utils/cli/src/shared_options.rs +++ b/codex-rs/utils/cli/src/shared_options.rs @@ -64,9 +64,36 @@ pub struct SharedCliOptions { /// Additional directories that should be writable alongside the primary workspace. #[arg(long = "add-dir", value_name = "DIR", value_hint = clap::ValueHint::DirPath)] pub add_dir: Vec, + + /// Exact runtime workspace roots. Unlike --add-dir, this replaces the + /// implicit cwd root and is intended for split-root workspaces. + #[arg( + long = "workspace-root", + value_name = "DIR", + value_hint = clap::ValueHint::DirPath, + conflicts_with = "add_dir" + )] + pub workspace_root: Vec, } impl SharedCliOptions { + pub fn validate_workspace_root_mode(&self) -> Result<(), &'static str> { + if !self.workspace_root.is_empty() && !self.add_dir.is_empty() { + return Err("--workspace-root cannot be combined with --add-dir"); + } + if !self.workspace_root.is_empty() && self.dangerously_bypass_approvals_and_sandbox { + return Err( + "--workspace-root cannot be combined with --dangerously-bypass-approvals-and-sandbox", + ); + } + if !self.workspace_root.is_empty() + && !matches!(self.sandbox_mode, Some(SandboxModeCliArg::WorkspaceWrite)) + { + return Err("--workspace-root requires --sandbox workspace-write"); + } + Ok(()) + } + pub fn inherit_exec_root_options(&mut self, root: &Self) { let self_selected_sandbox_mode = self.sandbox_mode.is_some() || self.dangerously_bypass_approvals_and_sandbox; @@ -82,6 +109,7 @@ impl SharedCliOptions { bypass_hook_trust, cwd, add_dir, + workspace_root, } = self; let Self { images: root_images, @@ -95,6 +123,7 @@ impl SharedCliOptions { bypass_hook_trust: root_bypass_hook_trust, cwd: root_cwd, add_dir: root_add_dir, + workspace_root: root_workspace_root, } = root; if model.is_none() { @@ -135,6 +164,11 @@ impl SharedCliOptions { merged_add_dir.append(add_dir); *add_dir = merged_add_dir; } + if !root_workspace_root.is_empty() { + let mut merged_workspace_root = root_workspace_root.clone(); + merged_workspace_root.append(workspace_root); + *workspace_root = merged_workspace_root; + } } pub fn apply_subcommand_overrides(&mut self, subcommand: Self) { @@ -152,6 +186,7 @@ impl SharedCliOptions { bypass_hook_trust, cwd, add_dir, + workspace_root, } = subcommand; if let Some(model) = model { @@ -186,5 +221,99 @@ impl SharedCliOptions { if !add_dir.is_empty() { self.add_dir.extend(add_dir); } + if !workspace_root.is_empty() { + self.workspace_root = workspace_root; + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use clap::Parser; + + #[derive(Debug, Parser)] + struct TestCli { + #[command(flatten)] + shared: SharedCliOptions, + } + + #[test] + fn workspace_root_is_repeatable() { + let cli = TestCli::try_parse_from([ + "test", + "--sandbox", + "workspace-write", + "--workspace-root", + "tenant", + "--workspace-root", + "devkit", + ]) + .expect("workspace roots should parse"); + + assert_eq!( + cli.shared.workspace_root, + vec![PathBuf::from("tenant"), PathBuf::from("devkit")] + ); + assert!(cli.shared.validate_workspace_root_mode().is_ok()); + } + + #[test] + fn workspace_root_conflicts_with_add_dir() { + let error = + TestCli::try_parse_from(["test", "--workspace-root", "tenant", "--add-dir", "extra"]) + .expect_err("workspace-root and add-dir should conflict"); + + assert_eq!(error.kind(), clap::error::ErrorKind::ArgumentConflict); + } + + #[test] + fn inherited_root_modes_cannot_be_mixed_across_exec_scopes() { + let mut subcommand = SharedCliOptions { + add_dir: vec![PathBuf::from("extra")], + ..Default::default() + }; + let root = SharedCliOptions { + sandbox_mode: Some(SandboxModeCliArg::WorkspaceWrite), + workspace_root: vec![PathBuf::from("tenant")], + ..Default::default() + }; + + subcommand.inherit_exec_root_options(&root); + + assert_eq!( + subcommand.validate_workspace_root_mode(), + Err("--workspace-root cannot be combined with --add-dir") + ); + } + + #[test] + fn workspace_root_requires_explicit_workspace_write() { + let shared = SharedCliOptions { + workspace_root: vec![PathBuf::from("tenant")], + ..Default::default() + }; + + assert_eq!( + shared.validate_workspace_root_mode(), + Err("--workspace-root requires --sandbox workspace-write") + ); + } + + #[test] + fn workspace_root_rejects_dangerous_sandbox_bypass() { + let shared = SharedCliOptions { + sandbox_mode: Some(SandboxModeCliArg::WorkspaceWrite), + dangerously_bypass_approvals_and_sandbox: true, + workspace_root: vec![PathBuf::from("tenant")], + ..Default::default() + }; + + assert_eq!( + shared.validate_workspace_root_mode(), + Err( + "--workspace-root cannot be combined with --dangerously-bypass-approvals-and-sandbox" + ) + ); } } diff --git a/justfile b/justfile index e8ee1ac7cf5..689036789bb 100644 --- a/justfile +++ b/justfile @@ -51,8 +51,16 @@ app-server-test-client *args: [no-cd] exec-harness-test: eval "$({{ justfile_directory() }}/scripts/local/exec-harness-env.sh)" && \ - cargo build --manifest-path {{ justfile_directory() }}/codex-rs/Cargo.toml -p codex-cli --bin codex && \ - codex_bin="${CARGO_TARGET_DIR:-{{ justfile_directory() }}/codex-rs/target}/debug/codex" && \ + cargo_manifest="{{ justfile_directory() }}/codex-rs/Cargo.toml" && \ + target_dir="${CARGO_TARGET_DIR:-{{ justfile_directory() }}/codex-rs/target}" && \ + cargo build --manifest-path "$cargo_manifest" -p codex-cli --bin codex && \ + codex_bin="$target_dir/debug/codex" && \ + if [ "$(uname -s)" = Linux ]; then \ + cargo build --manifest-path "$cargo_manifest" -p codex-bwrap --bin bwrap && \ + mkdir -p "$target_dir/debug/codex-resources" && \ + cp "$target_dir/debug/bwrap" "$target_dir/debug/codex-resources/bwrap" && \ + chmod 0755 "$target_dir/debug/codex-resources/bwrap"; \ + fi && \ {{ python }} {{ justfile_directory() }}/tools/codex-exec-harness/run_all.py --codex-bin "$codex_bin" --output-root "$CODEX_EXEC_HARNESS_OUTPUT_ROOT" --report-json "$CODEX_EXEC_HARNESS_REPORT_JSON" [no-cd] diff --git a/tools/codex-exec-harness/README.md b/tools/codex-exec-harness/README.md index 9085a6e7402..ba74164187a 100644 --- a/tools/codex-exec-harness/README.md +++ b/tools/codex-exec-harness/README.md @@ -17,6 +17,10 @@ Run the full harness suite against a freshly built local Codex binary: just exec-harness-test ``` +On Linux, this recipe also builds and stages the bundled +`codex-resources/bwrap` helper required to enforce writable roots outside the +launch working directory. + The full runner verifies the binary once before any scenario starts. Its aggregate report records the exact executable path, source commit, dirty state, build profile/channel, and binary digest under `provenance`. A stale or @@ -61,6 +65,13 @@ Scenarios are JSON files. Supported fields: - `turns`: ordered turn objects; turn 1 runs `codex exec`, later turns resume the captured thread id with `codex exec resume` - `files`: workspace files created before the run +- `external_files`: files created under the run-local `external/` directory +- `symlinks`: workspace-relative links to run-local fixture paths; targets may + use `{workspace}`, `{external}`, or `{run_dir}` +- `add_dirs`: run-local directories passed to Codex with `--add-dir` +- `workspace_roots`: exact run-local workspace roots passed to Codex with + repeatable `--workspace-root`; unlike `add_dirs`, these replace the implicit + writable launch CWD and require `sandbox: workspace-write` - `config_toml`: isolated `CODEX_LAB_HOME/config.toml` contents - `config_overrides`: `-c key=value` arguments passed to `codex exec` - `responses_api`: start a local fake Responses API and point Codex at it @@ -70,9 +81,16 @@ Scenarios are JSON files. Supported fields: provider scenarios while keeping Codex Lab, shell, XDG config, and cache paths isolated; this lets provider CLIs reuse their existing authentication - `skip_run_all`: omit a scenario from `run_all.py` and CI's all-scenario sweep +- `workspace_outside_git`: materialize the workspace under + `~/.codex-exec-harness-workspaces` or + `CODEX_EXEC_HARNESS_EXTERNAL_WORKSPACE_ROOT`, and fail unless Git reports no + containing worktree and no ancestor `AGENTS.md` or `AGENTS.override.md` - `expect`: assertions over return code, turn count, captured thread id, and fake Responses request bodies, captured agent messages and commands, and - optional durable Background Review target/currentness metadata + optional durable Background Review target/currentness metadata; + `launch_command` checks candidate argv and `workspace_paths` verifies + post-run file, directory, symlink, or absence evidence; `workspace_git` + checks recorded Git metadata such as `git_root`, `branch`, or `head_sha` - `timeout_seconds`: per-run timeout, defaulting to 90 seconds The fake Responses API is for request-shape proof only. Use direct scenario runs @@ -85,6 +103,48 @@ locally authenticated external provider. It intentionally gives spawned commands access to the caller's home directory, so it must remain excluded from `run_all.py` and CI. +## Generated Odoo Workspaces + +`odoo_workspace.py` is the bounded consumer for a devkit-generated non-Git +workspace. It does not parse `workspace.lock.toml` or Odoo markers. Instead it +runs the provider-owned command below and fails closed unless the returned +schema, guidance, source materialization, reserved override, and edit roots are +all current: + +```sh +uv --directory /path/to/odoo-devkit run platform workspace status \ + --manifest /path/to/tenant/workspace.toml --check +``` + +Launch an exact Codex Lab candidate in editable mode: + +```sh +python3 tools/codex-exec-harness/odoo_workspace.py \ + --devkit /path/to/odoo-devkit \ + --manifest /path/to/tenant/workspace.toml \ + --codex-bin /absolute/path/to/codex-lab \ + --source-repo /path/to/codex-lab \ + --mode exec \ + --access editable \ + --prompt 'Inspect the generated workspace.' \ + --evidence-file /path/to/evidence.json +``` + +Editable mode selects the built-in workspace-write sandbox and passes each +status-declared editable linked path through repeatable `--workspace-root`. +Those exact roots replace the implicit launch CWD, leaving the generated root +and managed checkouts read-only. Passing the same paths through legacy +`--add-dir` would also make the launch CWD writable, so the bounded adapter +intentionally does not combine the two mechanisms. `--access read-only` uses +the built-in read-only sandbox. Both modes require current candidate provenance +from `scripts/local/codex_lab_provenance.py`; prompt text is hashed and redacted +from evidence. + +`generated-odoo-workspace-consumption.json` is the deterministic exact-candidate +proof. It keeps the root non-Git, loads the canonical root guide, permits writes +through tenant/devkit links, and proves managed, generated-root, and outside +writes remain denied. + ## Auto-Validation Characterization Issue #284's first auto-validation contract is documented in diff --git a/tools/codex-exec-harness/harness.py b/tools/codex-exec-harness/harness.py index cfec230f61d..18f774244d2 100644 --- a/tools/codex-exec-harness/harness.py +++ b/tools/codex-exec-harness/harness.py @@ -23,6 +23,7 @@ AUTO_REVIEW_SUMMARY_MAX_FIELD_BYTES = 240 AUTO_REVIEW_SUMMARY_MAX_BYTES = 4096 AUTO_REVIEW_SUMMARY_OMITTED_TEMPLATE = "... {count} more finding(s) omitted" +ANCESTOR_GUIDANCE_FILENAMES = ("AGENTS.override.md", "AGENTS.md") class HarnessError(Exception): @@ -364,6 +365,7 @@ def git_text(*args: str) -> str | None: status = git_text("status", "--porcelain") return { "branch": git_text("branch", "--show-current"), + "git_root": git_text("rev-parse", "--show-toplevel"), "head_sha": git_text("rev-parse", "HEAD"), "worktree_path": str(workspace.resolve()), "clean": status == "" if status is not None else None, @@ -460,25 +462,71 @@ def base_url(self) -> str: return f"http://{host}:{port}/v1" -def make_paths(output_root: Path, scenario_name: str) -> RunPaths: +def make_paths( + output_root: Path, scenario_name: str, workspace_outside_git: bool = False +) -> RunPaths: stamp = time.strftime("%Y%m%d-%H%M%S") safe_name = safe_path_component(scenario_name) - run_dir = output_root / f"{stamp}-{safe_name}" + run_name = f"{stamp}-{safe_name}" + run_dir = output_root / run_name + external_workspace_root = Path( + os.environ.get( + "CODEX_EXEC_HARNESS_EXTERNAL_WORKSPACE_ROOT", + Path.home() / ".codex-exec-harness-workspaces", + ) + ).expanduser().resolve() + workspace = ( + external_workspace_root / run_name + if workspace_outside_git + else run_dir / "workspace" + ) suffix = 1 - while run_dir.exists(): + while run_dir.exists() or workspace.exists(): suffix += 1 - run_dir = output_root / f"{stamp}-{safe_name}-{suffix}" + run_name = f"{stamp}-{safe_name}-{suffix}" + run_dir = output_root / run_name + workspace = ( + external_workspace_root / run_name + if workspace_outside_git + else run_dir / "workspace" + ) return RunPaths( run_dir=run_dir, - workspace=run_dir / "workspace", + workspace=workspace, codex_home=run_dir / "codex-home", home=run_dir / "home", artifacts=run_dir / "artifacts", ) +def workspace_lexical_path(root: Path, rel_path: str, label: str) -> Path: + relative = Path(rel_path) + if relative.is_absolute() or ".." in relative.parts: + raise HarnessError(f"{label} escapes workspace: {rel_path}") + return root / relative + + +def ancestor_guidance_paths(workspace: Path) -> list[Path]: + return [ + candidate + for ancestor in workspace.parents + for filename in ANCESTOR_GUIDANCE_FILENAMES + if (candidate := ancestor / filename).exists() or candidate.is_symlink() + ] + + def materialize_workspace(scenario: dict[str, Any], paths: RunPaths) -> None: paths.workspace.mkdir(parents=True, exist_ok=True) + external_root = paths.run_dir / "external" + external_files = scenario.get("external_files", {}) + if not isinstance(external_files, dict): + raise HarnessError("external_files must be an object") + for rel_path, content in external_files.items(): + if not isinstance(rel_path, str): + raise HarnessError("external file paths must be strings") + file_text = str(content).replace("{workspace}", str(paths.workspace)) + save_text(resolve_under(external_root, rel_path, "external file path"), file_text) + files = scenario.get("files", {}) if not isinstance(files, dict): raise HarnessError("files must be an object") @@ -488,6 +536,28 @@ def materialize_workspace(scenario: dict[str, Any], paths: RunPaths) -> None: file_text = str(content).replace("{workspace}", str(paths.workspace)) save_text(resolve_under(paths.workspace, rel_path, "file path"), file_text) + symlinks = scenario.get("symlinks", {}) + if not isinstance(symlinks, dict): + raise HarnessError("symlinks must be an object") + for rel_path, target_template in symlinks.items(): + if not isinstance(rel_path, str) or not isinstance(target_template, str): + raise HarnessError("symlink paths and targets must be strings") + link_path = workspace_lexical_path(paths.workspace, rel_path, "symlink path") + target_text = ( + target_template.replace("{workspace}", str(paths.workspace)) + .replace("{external}", str(external_root)) + .replace("{run_dir}", str(paths.run_dir)) + ) + target_path = Path(target_text) + if not target_path.is_absolute(): + target_path = paths.run_dir / target_path + target_path = target_path.resolve(strict=True) + run_root = paths.run_dir.resolve(strict=True) + if not target_path.is_relative_to(run_root): + raise HarnessError(f"symlink target escapes run directory: {target_path}") + link_path.parent.mkdir(parents=True, exist_ok=True) + link_path.symlink_to(target_path, target_is_directory=target_path.is_dir()) + executable_home_files = scenario.get("executable_home_files", {}) if not isinstance(executable_home_files, dict): raise HarnessError("executable_home_files must be an object") @@ -525,6 +595,9 @@ def save_config(scenario: dict[str, Any], paths: RunPaths, base_url: str | None) config = config.replace("{workspace}", str(paths.workspace)).replace( "{home}", str(paths.home) ) + config = config.replace("{external}", str(paths.run_dir / "external")).replace( + "{run_dir}", str(paths.run_dir) + ) uses_responses_base_url = "{responses_base_url}" in config if uses_responses_base_url: if base_url is None: @@ -587,28 +660,48 @@ def build_command( raise HarnessError("prompt must be a string") resume = session_id is not None - sandbox = str(scenario.get("sandbox", "danger-full-access")) + sandbox_value = scenario.get("sandbox", "danger-full-access") + sandbox = str(sandbox_value) if sandbox_value is not None else None + command = [ + codex_bin, + "exec", + "--json", + "--skip-git-repo-check", + "-C", + str(paths.workspace), + ] + if sandbox is not None: + command.extend(["--sandbox", sandbox]) + add_dirs = scenario.get("add_dirs", []) + if not isinstance(add_dirs, list): + raise HarnessError("add_dirs must be a list") + for add_dir in add_dirs: + if not isinstance(add_dir, str): + raise HarnessError("add_dirs entries must be strings") + add_dir_path = Path( + add_dir.replace("{workspace}", str(paths.workspace)) + .replace("{external}", str(paths.run_dir / "external")) + .replace("{run_dir}", str(paths.run_dir)) + ).resolve(strict=True) + command.extend(["--add-dir", str(add_dir_path)]) + workspace_roots = scenario.get("workspace_roots", []) + if not isinstance(workspace_roots, list): + raise HarnessError("workspace_roots must be a list") + if add_dirs and workspace_roots: + raise HarnessError("workspace_roots cannot be combined with add_dirs") + if workspace_roots and sandbox != "workspace-write": + raise HarnessError("workspace_roots require sandbox=workspace-write") + for workspace_root in workspace_roots: + if not isinstance(workspace_root, str): + raise HarnessError("workspace_roots entries must be strings") + workspace_root_path = Path( + workspace_root.replace("{workspace}", str(paths.workspace)) + .replace("{external}", str(paths.run_dir / "external")) + .replace("{run_dir}", str(paths.run_dir)) + ).resolve(strict=True) + command.extend(["--workspace-root", str(workspace_root_path)]) if resume: - command = [ - codex_bin, - "exec", - "--json", - "--skip-git-repo-check", - "--sandbox", - sandbox, - "resume", - ] - else: - command = [ - codex_bin, - "exec", - "--json", - "--skip-git-repo-check", - "-C", - str(paths.workspace), - "--sandbox", - sandbox, - ] + command.append("resume") model = scenario.get("model") if isinstance(model, str) and model: command.extend(["-m", model]) @@ -616,10 +709,16 @@ def build_command( if not isinstance(config_overrides, list): raise HarnessError("config_overrides must be a list") for override in config_overrides: - command.extend(["-c", str(override)]) + rendered_override = ( + str(override) + .replace("{workspace}", str(paths.workspace)) + .replace("{external}", str(paths.run_dir / "external")) + .replace("{run_dir}", str(paths.run_dir)) + ) + command.extend(["-c", rendered_override]) if resume and session_id is not None: command.append(session_id) - command.append(prompt) + command.extend(["--", prompt]) return command @@ -833,6 +932,7 @@ def run_turns( "token_usage": token_usage_delta, "token_usage_snapshot": token_usage_snapshot, "artifact_dir": str(artifact_dir), + "command": command, } ) if result["returncode"] != 0: @@ -847,6 +947,8 @@ def run_turns( "turns": turn_results, "thread_id": thread_id, "token_usage": previous_token_usage, + "workspace": str(paths.workspace), + "run_dir": str(paths.run_dir), } @@ -1172,6 +1274,55 @@ def add_input_prefix_assertion_failures( ) +def add_workspace_path_assertion_failures( + failures: list[str], run: dict[str, Any], assertions: Any +) -> None: + if assertions is None: + return + if not isinstance(assertions, list): + raise HarnessError("expect.workspace_paths must be a list") + workspace_value = run.get("workspace") + if not isinstance(workspace_value, str): + raise HarnessError("run workspace is unavailable for path assertions") + workspace = Path(workspace_value) + for index, assertion in enumerate(assertions): + label = f"workspace_paths[{index}]" + if not isinstance(assertion, dict): + raise HarnessError(f"expect.{label} must be an object") + rel_path = assertion.get("path") + if not isinstance(rel_path, str): + raise HarnessError(f"expect.{label}.path must be a string") + path = workspace_lexical_path(workspace, rel_path, label) + exists = path.exists() or path.is_symlink() + expected_exists = assertion.get("exists", True) + if type(expected_exists) is not bool: + raise HarnessError(f"expect.{label}.exists must be boolean") + if exists != expected_exists: + failures.append(f"{label}: expected exists={expected_exists}, found {exists}") + continue + if not exists: + continue + expected_type = assertion.get("type") + if expected_type == "file" and not path.is_file(): + failures.append(f"{label}: expected file") + elif expected_type == "directory" and not path.is_dir(): + failures.append(f"{label}: expected directory") + elif expected_type == "symlink" and not path.is_symlink(): + failures.append(f"{label}: expected symlink") + elif expected_type not in {None, "file", "directory", "symlink"}: + raise HarnessError(f"expect.{label}.type is unsupported") + if "contains" in assertion: + if not path.is_file(): + failures.append(f"{label}: cannot inspect content of non-file") + continue + add_text_assertion_failures( + failures, + path.read_text(encoding="utf-8"), + assertion, + label, + ) + + def evaluate_expectations( scenario: dict[str, Any], run: dict[str, Any], requests: list[dict[str, Any]] ) -> list[str]: @@ -1203,6 +1354,21 @@ def evaluate_expectations( if expect.get("thread_id") == "required" and not run.get("thread_id"): failures.append("expected a captured thread_id") + workspace_git_assertion = expect.get("workspace_git") + if workspace_git_assertion is not None: + if not isinstance(workspace_git_assertion, dict): + raise HarnessError("expect.workspace_git must be an object") + workspace_git = run.get("workspace_git") + if not isinstance(workspace_git, dict): + failures.append("workspace_git: missing workspace Git evidence") + else: + for field, expected_value in workspace_git_assertion.items(): + if workspace_git.get(field) != expected_value: + failures.append( + f"workspace_git.{field}: expected {expected_value!r}, " + f"found {workspace_git.get(field)!r}" + ) + add_list_text_assertion_failures( failures, run.get("agent_messages"), @@ -1215,6 +1381,18 @@ def evaluate_expectations( expect.get("commands"), "commands", ) + launch_command_assertion = expect.get("launch_command") + if launch_command_assertion is not None: + if not isinstance(launch_command_assertion, dict): + raise HarnessError("expect.launch_command must be an object") + first_command = run.get("turns", [{}])[0].get("command", []) + add_text_assertion_failures( + failures, + first_command, + launch_command_assertion, + "launch_command", + ) + add_workspace_path_assertion_failures(failures, run, expect.get("workspace_paths")) add_background_review_assertion_failures( failures, run, expect.get("background_review") ) @@ -1426,9 +1604,29 @@ def run_scenario(args: argparse.Namespace) -> int: if not codex_bin: raise HarnessError("codex binary not found; pass --codex-bin") - paths = make_paths(Path(args.output_root).resolve(), name) + workspace_outside_git = scenario.get("workspace_outside_git", False) + if type(workspace_outside_git) is not bool: + raise HarnessError("workspace_outside_git must be boolean") + paths = make_paths( + Path(args.output_root).resolve(), + name, + workspace_outside_git=workspace_outside_git, + ) paths.artifacts.mkdir(parents=True, exist_ok=True) materialize_workspace(scenario, paths) + if workspace_outside_git: + initial_git_state = collect_workspace_git_state(paths.workspace) + if initial_git_state.get("git_root") is not None: + raise HarnessError( + "workspace_outside_git resolved inside a Git work tree: " + f"{initial_git_state['git_root']}" + ) + ancestor_guidance = ancestor_guidance_paths(paths.workspace) + if ancestor_guidance: + raise HarnessError( + "workspace_outside_git resolved below ancestor guidance: " + + ", ".join(str(path) for path in ancestor_guidance) + ) responses_api = scenario.get("responses_api") if responses_api is not None and not isinstance(responses_api, dict): diff --git a/tools/codex-exec-harness/odoo_workspace.py b/tools/codex-exec-harness/odoo_workspace.py new file mode 100644 index 00000000000..c501ec27ef0 --- /dev/null +++ b/tools/codex-exec-harness/odoo_workspace.py @@ -0,0 +1,770 @@ +#!/usr/bin/env python3 +"""Launch Codex from a current generated Odoo workspace.""" + +import argparse +import hashlib +import json +import os +import shutil +import subprocess +import sys +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + + +SCHEMA_VERSION = 1 +SUPPORTED_STATUS_SCHEMA_VERSION = 1 +DEFAULT_STATUS_TIMEOUT_SECONDS = 120 +PROVENANCE_TIMEOUT_SECONDS = 30 +MAX_STATUS_DETAIL_ITEMS = 12 +ACCESS_MODES = {"editable", "read-only"} +ANCESTOR_GUIDANCE_FILENAMES = ("AGENTS.override.md", "AGENTS.md") + + +class OdooWorkspaceError(Exception): + """Raised when generated workspace evidence is unsafe or stale.""" + + +@dataclass(frozen=True) +class WorkspaceSource: + role: str + workspace_relative_path: str + workspace_entry_path: Path + resolved_path: Path + materialization: str + editable: bool + + def evidence(self) -> dict[str, object]: + return { + "role": self.role, + "workspace_relative_path": self.workspace_relative_path, + "workspace_entry_path": str(self.workspace_entry_path), + "resolved_path": str(self.resolved_path), + "materialization": self.materialization, + "editable": self.editable, + } + + +@dataclass(frozen=True) +class OdooWorkspaceLaunch: + workspace_path: Path + manifest_path: Path + agents_path: Path + local_notes_path: Path + git_root: Path | None + status_command: tuple[str, ...] + status_payload_sha256: str + editable_sources: tuple[WorkspaceSource, ...] + managed_sources: tuple[WorkspaceSource, ...] + + @property + def writable_roots(self) -> tuple[Path, ...]: + roots: list[Path] = [] + seen: set[Path] = set() + for source in self.editable_sources: + if source.resolved_path in seen: + continue + seen.add(source.resolved_path) + roots.append(source.resolved_path) + return tuple(roots) + + def evidence(self) -> dict[str, object]: + return { + "workspace_path": str(self.workspace_path), + "manifest_path": str(self.manifest_path), + "agents_path": str(self.agents_path), + "local_notes_path": str(self.local_notes_path), + "git_root": str(self.git_root) if self.git_root is not None else None, + "non_git_workspace": self.git_root is None, + "status_command": list(self.status_command), + "status_payload_sha256": self.status_payload_sha256, + "editable_sources": [source.evidence() for source in self.editable_sources], + "managed_sources": [source.evidence() for source in self.managed_sources], + "writable_roots": [str(path) for path in self.writable_roots], + } + + +def _required_text(value: object, label: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise OdooWorkspaceError(f"{label} must be a non-empty string") + return value.strip() + + +def _required_bool(payload: dict[str, Any], key: str) -> None: + if payload.get(key) is not True: + raise OdooWorkspaceError(f"workspace status requires {key}=true") + + +def _resolve_existing_directory(value: object, label: str) -> Path: + path_text = _required_text(value, label) + path = Path(path_text).expanduser() + if not path.is_absolute(): + raise OdooWorkspaceError(f"{label} must be absolute: {path_text}") + try: + resolved = path.resolve(strict=True) + except OSError as error: + raise OdooWorkspaceError(f"{label} is unavailable: {path_text}: {error}") from error + if not resolved.is_dir(): + raise OdooWorkspaceError(f"{label} is not a directory: {resolved}") + if resolved == Path(resolved.anchor): + raise OdooWorkspaceError(f"{label} cannot be the filesystem root") + return resolved + + +def _absolute_existing_directory(value: object, label: str) -> Path: + path_text = _required_text(value, label) + raw_path = Path(os.path.abspath(os.path.expanduser(path_text))) + if not raw_path.is_dir(): + raise OdooWorkspaceError(f"{label} is not a directory: {raw_path}") + try: + path = raw_path.parent.resolve(strict=True) / raw_path.name + except OSError as error: + raise OdooWorkspaceError(f"{label} is unavailable: {raw_path}: {error}") from error + if path == Path(path.anchor): + raise OdooWorkspaceError(f"{label} cannot be the filesystem root") + return path + + +def _resolve_existing_file(value: object, label: str) -> Path: + path_text = _required_text(value, label) + path = Path(path_text).expanduser() + if not path.is_absolute(): + raise OdooWorkspaceError(f"{label} must be absolute: {path_text}") + try: + resolved = path.resolve(strict=True) + except OSError as error: + raise OdooWorkspaceError(f"{label} is unavailable: {path_text}: {error}") from error + if not resolved.is_file(): + raise OdooWorkspaceError(f"{label} is not a file: {resolved}") + return resolved + + +def _workspace_relative_path(value: object, label: str) -> str: + path_text = _required_text(value, label) + path = Path(path_text) + if path.is_absolute() or ".." in path.parts: + raise OdooWorkspaceError(f"{label} must remain inside the generated workspace") + return path.as_posix() + + +def _canonical_json_sha256(value: object) -> str: + payload = json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8") + return hashlib.sha256(payload).hexdigest() + + +def _paths_overlap(first: Path, second: Path) -> bool: + return first == second or first.is_relative_to(second) or second.is_relative_to(first) + + +def _ancestor_guidance_paths(workspace_path: Path) -> tuple[Path, ...]: + return tuple( + candidate + for ancestor in workspace_path.parents + for filename in ANCESTOR_GUIDANCE_FILENAMES + if (candidate := ancestor / filename).exists() or candidate.is_symlink() + ) + + +def _git_root(path: Path) -> Path | None: + try: + completed = subprocess.run( + ["git", "-C", str(path), "rev-parse", "--show-toplevel"], + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + except OSError as error: + raise OdooWorkspaceError(f"could not verify non-Git workspace root: {error}") from error + if completed.returncode == 0: + root_text = completed.stdout.strip() + if not root_text: + raise OdooWorkspaceError("Git reported an empty workspace root") + return Path(root_text).expanduser().resolve(strict=True) + detail = (completed.stderr or completed.stdout).strip().lower() + if "not a git repository" in detail: + return None + raise OdooWorkspaceError( + "could not verify non-Git workspace root: " + + ((completed.stderr or completed.stdout).strip() or f"git exited {completed.returncode}") + ) + + +def _parse_status_json(stdout: str) -> dict[str, Any]: + try: + payload = json.loads(stdout) + except json.JSONDecodeError as error: + raise OdooWorkspaceError(f"workspace status emitted invalid JSON: {error}") from error + if not isinstance(payload, dict): + raise OdooWorkspaceError("workspace status must emit a JSON object") + return payload + + +def _bounded_status_detail(payload: dict[str, Any], fallback: str) -> str: + stale_reasons = payload.get("stale_reasons") + if isinstance(stale_reasons, list): + reasons = [str(item).strip() for item in stale_reasons if str(item).strip()] + if reasons: + bounded = reasons[:MAX_STATUS_DETAIL_ITEMS] + suffix = "" if len(reasons) <= len(bounded) else ", ..." + return f"stale reasons: {', '.join(bounded)}{suffix}" + return fallback.strip() or "no diagnostic output" + + +def run_workspace_status( + *, + uv_bin: str, + devkit_path: Path, + manifest_path: Path, + timeout_seconds: int = DEFAULT_STATUS_TIMEOUT_SECONDS, +) -> tuple[dict[str, Any], tuple[str, ...]]: + if timeout_seconds <= 0: + raise OdooWorkspaceError("workspace status timeout must be positive") + resolved_uv = resolve_executable(uv_bin) + resolved_devkit = devkit_path.expanduser().resolve(strict=True) + resolved_manifest = manifest_path.expanduser().resolve(strict=True) + if not resolved_devkit.is_dir(): + raise OdooWorkspaceError(f"devkit path is not a directory: {resolved_devkit}") + if not resolved_manifest.is_file(): + raise OdooWorkspaceError(f"workspace manifest is not a file: {resolved_manifest}") + command = ( + str(resolved_uv), + "--directory", + str(resolved_devkit), + "run", + "platform", + "workspace", + "status", + "--manifest", + str(resolved_manifest), + "--check", + ) + try: + completed = subprocess.run( + command, + cwd=resolved_devkit, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=timeout_seconds, + ) + except subprocess.TimeoutExpired as error: + raise OdooWorkspaceError( + f"workspace status --check timed out after {timeout_seconds}s" + ) from error + except OSError as error: + raise OdooWorkspaceError(f"could not run workspace status --check: {error}") from error + + payload = _parse_status_json(completed.stdout) + if completed.returncode != 0: + detail = _bounded_status_detail(payload, completed.stderr) + raise OdooWorkspaceError( + f"workspace status --check failed with exit {completed.returncode}: {detail}" + ) + return payload, command + + +def _source_records(payload: dict[str, Any], workspace_path: Path) -> tuple[WorkspaceSource, ...]: + sources_value = payload.get("sources") + if not isinstance(sources_value, list) or not sources_value: + raise OdooWorkspaceError("workspace status requires a non-empty sources list") + + sources: list[WorkspaceSource] = [] + roles: set[str] = set() + for index, raw_source in enumerate(sources_value): + if not isinstance(raw_source, dict): + raise OdooWorkspaceError(f"workspace source {index} must be an object") + role = _required_text(raw_source.get("role"), f"workspace source {index} role") + if role in roles: + raise OdooWorkspaceError(f"workspace source role is duplicated: {role}") + roles.add(role) + relative_path = _workspace_relative_path( + raw_source.get("workspace_relative_path"), + f"workspace source {role} relative path", + ) + resolved_path = _resolve_existing_directory( + raw_source.get("resolved_path"), f"workspace source {role} resolved path" + ) + workspace_entry = _absolute_existing_directory( + raw_source.get("workspace_entry_path"), + f"workspace source {role} entry path", + ) + expected_entry = workspace_path / relative_path + if workspace_entry != expected_entry: + raise OdooWorkspaceError( + f"workspace source {role} entry is redirected outside its declared workspace path" + ) + try: + expected_entry_resolved = expected_entry.resolve(strict=True) + except OSError as error: + raise OdooWorkspaceError( + f"workspace source {role} entry is unavailable: {expected_entry}: {error}" + ) from error + if expected_entry_resolved != resolved_path: + raise OdooWorkspaceError( + f"workspace source {role} entry does not resolve to its declared source root" + ) + materialization = _required_text( + raw_source.get("materialization"), f"workspace source {role} materialization" + ) + if materialization not in {"linked_path", "managed_checkout"}: + raise OdooWorkspaceError( + f"workspace source {role} has unsupported materialization: {materialization}" + ) + editable = raw_source.get("editable") + if type(editable) is not bool: + raise OdooWorkspaceError(f"workspace source {role} editable must be boolean") + if editable and materialization != "linked_path": + raise OdooWorkspaceError( + f"workspace source {role} cannot be editable when materialized as {materialization}" + ) + if raw_source.get("materialization_current") is not True: + raise OdooWorkspaceError(f"workspace source {role} materialization is not current") + if raw_source.get("materialization_state") != "current": + raise OdooWorkspaceError(f"workspace source {role} materialization state is not current") + if materialization == "linked_path" and not workspace_entry.is_symlink(): + raise OdooWorkspaceError(f"workspace source {role} must be a linked path") + if materialization == "managed_checkout" and workspace_entry.is_symlink(): + raise OdooWorkspaceError(f"workspace source {role} managed checkout cannot be a symlink") + sources.append( + WorkspaceSource( + role=role, + workspace_relative_path=relative_path, + workspace_entry_path=workspace_entry, + resolved_path=resolved_path, + materialization=materialization, + editable=editable, + ) + ) + return tuple(sources) + + +def validate_workspace_status( + payload: dict[str, Any], + *, + manifest_path: Path, + status_command: tuple[str, ...], +) -> OdooWorkspaceLaunch: + if type(payload.get("schema_version")) is not int or payload["schema_version"] != SUPPORTED_STATUS_SCHEMA_VERSION: + raise OdooWorkspaceError("workspace status uses an unsupported schema version") + for key in ( + "current", + "workspace_exists", + "lock_file_exists", + "lock_file_current", + "surface_current", + "materialization_current", + "managed_source_baseline_current", + ): + _required_bool(payload, key) + + workspace_path = _absolute_existing_directory( + payload.get("workspace_path"), "workspace path" + ) + if workspace_path.is_symlink(): + raise OdooWorkspaceError("workspace path must not be redirected through a symlink") + git_root = _git_root(workspace_path) + if git_root is not None: + raise OdooWorkspaceError( + f"generated workspace must be outside a Git work tree; found {git_root}" + ) + ancestor_guidance = _ancestor_guidance_paths(workspace_path) + if ancestor_guidance: + raise OdooWorkspaceError( + "generated workspace is shadowed by ancestor guidance: " + + ", ".join(str(path) for path in ancestor_guidance) + ) + agents_path = _resolve_existing_file( + payload.get("workspace_agents_path"), "workspace AGENTS.md path" + ) + expected_agents = workspace_path / "AGENTS.md" + if agents_path != expected_agents.resolve(strict=True) or expected_agents.is_symlink(): + raise OdooWorkspaceError("canonical workspace AGENTS.md must be a regular root file") + + resolved_manifest = manifest_path.expanduser().resolve(strict=True) + manifest = payload.get("manifest") + if not isinstance(manifest, dict) or manifest.get("current") is not True: + raise OdooWorkspaceError("workspace status requires a current manifest") + reported_manifest = _resolve_existing_file(manifest.get("path"), "reported manifest path") + if reported_manifest != resolved_manifest: + raise OdooWorkspaceError("workspace status manifest does not match the requested manifest") + manifest_sha256 = _required_text(manifest.get("sha256"), "reported manifest sha256") + if _sha256_file(resolved_manifest) != manifest_sha256: + raise OdooWorkspaceError("workspace manifest changed after status verification") + + reserved_override = payload.get("reserved_override") + if not isinstance(reserved_override, dict): + raise OdooWorkspaceError("workspace status requires reserved override evidence") + override_path_text = _required_text( + reserved_override.get("path"), "workspace reserved override path" + ) + override_path = Path(override_path_text).expanduser() + if not override_path.is_absolute(): + raise OdooWorkspaceError("workspace reserved override path must be absolute") + override_path = override_path.parent.resolve(strict=True) / override_path.name + if override_path != workspace_path / "AGENTS.override.md": + raise OdooWorkspaceError("workspace reserved override path does not match the root contract") + if type(reserved_override.get("exists")) is not bool: + raise OdooWorkspaceError("workspace reserved override existence must be boolean") + if ( + reserved_override["exists"] is not False + or override_path.exists() + or override_path.is_symlink() + ): + raise OdooWorkspaceError( + "generated workspace is shadowed by reserved AGENTS.override.md; resync or remove it" + ) + if reserved_override.get("semantics") != "full_replacement": + raise OdooWorkspaceError("workspace status reported unknown override semantics") + if reserved_override.get("allowed_in_normal_flow") is not False: + raise OdooWorkspaceError("workspace status reported unsafe override semantics") + + local_notes = payload.get("local_notes") + if not isinstance(local_notes, dict) or local_notes.get("valid") is not True: + raise OdooWorkspaceError("workspace local notes are invalid") + if type(local_notes.get("exists")) is not bool: + raise OdooWorkspaceError("workspace local notes existence must be boolean") + if local_notes.get("semantics") != "supplemental_non_secret_notes": + raise OdooWorkspaceError("workspace status reported unknown local note semantics") + local_notes_path_text = _required_text( + local_notes.get("path"), "workspace local notes path" + ) + local_notes_path = Path(local_notes_path_text).expanduser() + if not local_notes_path.is_absolute(): + raise OdooWorkspaceError("workspace local notes path must be absolute") + local_notes_path = local_notes_path.parent.resolve(strict=True) / local_notes_path.name + expected_local_notes = workspace_path / "workspace.local.md" + if local_notes_path != expected_local_notes: + raise OdooWorkspaceError("workspace local notes path does not match the root contract") + local_notes_exists = local_notes_path.exists() or local_notes_path.is_symlink() + if local_notes_exists != local_notes["exists"]: + raise OdooWorkspaceError("workspace local notes changed after status verification") + if local_notes_exists and (local_notes_path.is_symlink() or not local_notes_path.is_file()): + raise OdooWorkspaceError("workspace local notes must be a regular file when present") + + sources = _source_records(payload, workspace_path) + source_by_role = {source.role: source for source in sources} + edit_roots = payload.get("edit_roots") + if not isinstance(edit_roots, list): + raise OdooWorkspaceError("workspace status requires edit_roots") + edit_roles: list[str] = [] + for index, raw_root in enumerate(edit_roots): + if not isinstance(raw_root, dict): + raise OdooWorkspaceError(f"workspace edit root {index} must be an object") + role = _required_text(raw_root.get("role"), f"workspace edit root {index} role") + source = source_by_role.get(role) + if source is None or not source.editable: + raise OdooWorkspaceError(f"workspace edit root {role} is not an editable source") + resolved_path = _resolve_existing_directory( + raw_root.get("resolved_path"), f"workspace edit root {role} resolved path" + ) + relative_path = _workspace_relative_path( + raw_root.get("workspace_relative_path"), + f"workspace edit root {role} relative path", + ) + if resolved_path != source.resolved_path or relative_path != source.workspace_relative_path: + raise OdooWorkspaceError(f"workspace edit root {role} disagrees with source evidence") + edit_roles.append(role) + expected_edit_roles = [source.role for source in sources if source.editable] + if edit_roles != expected_edit_roles: + raise OdooWorkspaceError( + "workspace edit roots do not exactly match status-declared editable sources" + ) + + editable_sources = tuple(source for source in sources if source.editable) + read_only_sources = tuple(source for source in sources if not source.editable) + for editable_source in editable_sources: + if _paths_overlap(editable_source.resolved_path, workspace_path): + raise OdooWorkspaceError( + f"workspace editable source {editable_source.role} overlaps the generated workspace" + ) + for read_only_source in read_only_sources: + if _paths_overlap(editable_source.resolved_path, read_only_source.resolved_path): + raise OdooWorkspaceError( + "workspace editable source " + f"{editable_source.role} overlaps read-only source {read_only_source.role}" + ) + + return OdooWorkspaceLaunch( + workspace_path=workspace_path, + manifest_path=resolved_manifest, + agents_path=agents_path, + local_notes_path=local_notes_path, + git_root=git_root, + status_command=status_command, + status_payload_sha256=_canonical_json_sha256(payload), + editable_sources=editable_sources, + managed_sources=read_only_sources, + ) + + +def inspect_odoo_workspace( + *, + uv_bin: str, + devkit_path: Path, + manifest_path: Path, + timeout_seconds: int = DEFAULT_STATUS_TIMEOUT_SECONDS, +) -> tuple[OdooWorkspaceLaunch, dict[str, Any]]: + payload, command = run_workspace_status( + uv_bin=uv_bin, + devkit_path=devkit_path, + manifest_path=manifest_path, + timeout_seconds=timeout_seconds, + ) + return ( + validate_workspace_status( + payload, + manifest_path=manifest_path, + status_command=command, + ), + payload, + ) + + +def resolve_executable(value: str) -> Path: + requested = Path(value).expanduser() + if requested.parent != Path(".") or requested.is_absolute(): + path = requested.resolve(strict=True) + else: + discovered = shutil.which(value) + if discovered is None: + raise OdooWorkspaceError(f"could not find executable: {value}") + path = Path(discovered).resolve(strict=True) + if not path.is_file(): + raise OdooWorkspaceError(f"executable path is not a file: {path}") + if not os.access(path, os.X_OK): + raise OdooWorkspaceError(f"executable path is not executable: {path}") + return path + + +def build_codex_command( + *, + launch: OdooWorkspaceLaunch, + codex_bin: Path, + mode: str, + access: str, + prompt: str | None, + model: str | None = None, + config_profile: str | None = None, + auth_profile: str | None = None, +) -> tuple[str, ...]: + if mode not in {"exec", "interactive"}: + raise OdooWorkspaceError(f"unsupported launch mode: {mode}") + if access not in ACCESS_MODES: + raise OdooWorkspaceError(f"unsupported access mode: {access}") + if mode == "exec" and (prompt is None or not prompt.strip()): + raise OdooWorkspaceError("exec mode requires a prompt") + + command = [str(codex_bin)] + if mode == "exec": + command.extend(("exec", "--json", "--skip-git-repo-check")) + command.extend(("-C", str(launch.workspace_path))) + if model: + command.extend(("--model", model)) + if config_profile: + command.extend(("--profile", config_profile)) + if auth_profile: + command.extend(("--auth-profile", auth_profile)) + if access == "read-only": + command.extend(("--sandbox", "read-only")) + else: + if not launch.writable_roots: + raise OdooWorkspaceError("editable mode requires at least one declared edit root") + command.extend(("--sandbox", "workspace-write")) + for writable_root in launch.writable_roots: + command.extend(("--workspace-root", str(writable_root))) + if prompt is not None and prompt.strip(): + command.extend(("--", prompt)) + return tuple(command) + + +def _sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def verify_codex_provenance(*, source_repo: Path, codex_bin: Path) -> dict[str, Any]: + resolved_source = source_repo.expanduser().resolve(strict=True) + script = resolved_source / "scripts" / "local" / "codex_lab_provenance.py" + if not script.is_file() or script.is_symlink(): + raise OdooWorkspaceError(f"Codex provenance verifier is unavailable: {script}") + command = [ + sys.executable, + str(script), + "--repo-root", + str(resolved_source), + "--binary", + str(codex_bin), + "--verify-only", + "--json", + ] + try: + completed = subprocess.run( + command, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=PROVENANCE_TIMEOUT_SECONDS, + ) + except subprocess.TimeoutExpired as error: + raise OdooWorkspaceError("Codex provenance verification timed out") from error + except OSError as error: + raise OdooWorkspaceError(f"could not run Codex provenance verification: {error}") from error + if completed.returncode != 0: + detail = (completed.stderr or completed.stdout).strip() or "no diagnostic output" + raise OdooWorkspaceError(f"Codex provenance verification failed: {detail}") + try: + report = json.loads(completed.stdout) + except json.JSONDecodeError as error: + raise OdooWorkspaceError(f"Codex provenance verifier emitted invalid JSON: {error}") from error + if not isinstance(report, dict) or report.get("status") != "current": + raise OdooWorkspaceError( + "Codex binary provenance is not current: " + + json.dumps(report, sort_keys=True, separators=(",", ":")) + ) + return report + + +def _redacted_command(command: tuple[str, ...], prompt: str | None) -> list[str]: + redacted = list(command) + if prompt is not None and prompt.strip() and redacted and redacted[-1] == prompt: + redacted[-1] = "" + return redacted + + +def build_evidence( + *, + launch: OdooWorkspaceLaunch, + codex_bin: Path, + provenance: dict[str, Any], + command: tuple[str, ...], + mode: str, + access: str, + prompt: str | None, + returncode: int | None, +) -> dict[str, object]: + return { + "schema_version": SCHEMA_VERSION, + "generated_at": datetime.now(UTC).isoformat(), + "status": "planned" if returncode is None else ("pass" if returncode == 0 else "fail"), + "mode": mode, + "access": access, + "command": _redacted_command(command, prompt), + "prompt_sha256": ( + hashlib.sha256(prompt.encode("utf-8")).hexdigest() + if prompt is not None and prompt.strip() + else None + ), + "returncode": returncode, + "codex_binary": { + "path": str(codex_bin), + "sha256": _sha256_file(codex_bin), + "provenance": provenance, + }, + "workspace": launch.evidence(), + "permissions": { + "profile": ":workspace" if access == "editable" else ":read-only", + "workspace_root_writable": False, + "writable_roots": ( + [str(path) for path in launch.writable_roots] + if access == "editable" + else [] + ), + }, + "guidance": { + "path": str(launch.agents_path), + "sha256": _sha256_file(launch.agents_path), + "local_notes_exists": launch.local_notes_path.is_file(), + "reserved_override_exists": (launch.workspace_path / "AGENTS.override.md").exists(), + }, + } + + +def write_json(path: Path, value: object) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") + + +def parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--devkit", required=True, help="Path to the odoo-devkit checkout") + parser.add_argument("--manifest", required=True, help="Path to the tenant workspace.toml") + parser.add_argument("--uv-bin", default="uv", help="uv executable used for workspace status") + parser.add_argument("--codex-bin", required=True, help="Exact Codex Lab candidate binary") + parser.add_argument("--source-repo", required=True, help="Codex Lab source repo for provenance") + parser.add_argument("--mode", choices=("exec", "interactive"), default="exec") + parser.add_argument("--access", choices=tuple(sorted(ACCESS_MODES)), default="editable") + parser.add_argument("--model", default=None) + parser.add_argument("--config-profile", default=None) + parser.add_argument("--auth-profile", default=None) + parser.add_argument("--prompt", default=None) + parser.add_argument("--status-timeout-seconds", type=int, default=DEFAULT_STATUS_TIMEOUT_SECONDS) + parser.add_argument("--evidence-file", default=None) + parser.add_argument("--dry-run", action="store_true") + return parser.parse_args(argv) + + +def main(argv: list[str]) -> int: + args = parse_args(argv) + try: + launch, _payload = inspect_odoo_workspace( + uv_bin=args.uv_bin, + devkit_path=Path(args.devkit), + manifest_path=Path(args.manifest), + timeout_seconds=args.status_timeout_seconds, + ) + codex_bin = resolve_executable(args.codex_bin) + provenance = verify_codex_provenance( + source_repo=Path(args.source_repo), codex_bin=codex_bin + ) + command = build_codex_command( + launch=launch, + codex_bin=codex_bin, + mode=args.mode, + access=args.access, + prompt=args.prompt, + model=args.model, + config_profile=args.config_profile, + auth_profile=args.auth_profile, + ) + planned_evidence = build_evidence( + launch=launch, + codex_bin=codex_bin, + provenance=provenance, + command=command, + mode=args.mode, + access=args.access, + prompt=args.prompt, + returncode=None, + ) + if args.dry_run: + if args.evidence_file: + write_json(Path(args.evidence_file), planned_evidence) + print(json.dumps(planned_evidence, indent=2, sort_keys=True)) + return 0 + + completed = subprocess.run(command, cwd=launch.workspace_path) + final_evidence = build_evidence( + launch=launch, + codex_bin=codex_bin, + provenance=provenance, + command=command, + mode=args.mode, + access=args.access, + prompt=args.prompt, + returncode=completed.returncode, + ) + if args.evidence_file: + write_json(Path(args.evidence_file), final_evidence) + return completed.returncode + except (OdooWorkspaceError, OSError, ValueError) as error: + print(f"odoo-workspace: {error}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) diff --git a/tools/codex-exec-harness/scenarios/generated-odoo-workspace-consumption.json b/tools/codex-exec-harness/scenarios/generated-odoo-workspace-consumption.json new file mode 100644 index 00000000000..3eef121d524 --- /dev/null +++ b/tools/codex-exec-harness/scenarios/generated-odoo-workspace-consumption.json @@ -0,0 +1,138 @@ +{ + "name": "generated-odoo-workspace-consumption", + "characterization": { + "issue": 360, + "status": "runtime-covered" + }, + "model": "gpt-5.5", + "prompt": "Follow the canonical generated-workspace guide and prove the declared write boundary.", + "git_init": false, + "workspace_outside_git": true, + "sandbox": "workspace-write", + "workspace_roots": [ + "{external}/tenant", + "{external}/devkit" + ], + "files": { + "AGENTS.md": "# Canonical generated Odoo workspace guide\n\nMarker: CANONICAL_ODOO_WORKSPACE_GUIDE\n\n- `sources/tenant` and `sources/devkit` are declared editable roots.\n- `sources/runtime`, generated surfaces, and paths outside the declared roots are read-only.\n- `workspace.local.md` is supplemental and must never replace this guide.\n- Preserve Launchplane ownership boundaries; do not perform shared or production mutations from this local workspace.\n", + "workspace.local.md": "Fixture-only supplemental note.\n", + "workspace.lock.toml": "schema_version = 1\nworkspace_path = \"{workspace}\"\n\n[agent_workspace]\ncontract_version = 1\nlocal_notes_path = \"workspace.local.md\"\nreserved_override_path = \"AGENTS.override.md\"\nreserved_override_semantics = \"full_replacement\"\n\n[repos.tenant]\nworkspace_relative_path = \"sources/tenant\"\nresolved_path = \"{external}/tenant\"\nmaterialization = \"linked_path\"\neditable = true\n\n[repos.devkit]\nworkspace_relative_path = \"sources/devkit\"\nresolved_path = \"{external}/devkit\"\nmaterialization = \"linked_path\"\neditable = true\n\n[repos.runtime]\nworkspace_relative_path = \"sources/runtime\"\nresolved_path = \"{workspace}/sources/runtime\"\nmaterialization = \"managed_checkout\"\neditable = false\n", + "sources/runtime/README.md": "managed runtime checkout\n" + }, + "external_files": { + "tenant/README.md": "editable tenant source\n", + "devkit/README.md": "editable devkit source\n" + }, + "symlinks": { + "sources/tenant": "{external}/tenant", + "sources/devkit": "{external}/devkit" + }, + "config_toml": "approval_policy = \"never\"\nmodel_provider = \"harness\"\n\n[model_providers.harness]\nname = \"Harness\"\nbase_url = \"{responses_base_url}\"\nwire_api = \"responses\"\nrequires_openai_auth = false\nsupports_websockets = false\n", + "responses_api": { + "responses": [ + { + "response_id": "resp-odoo-workspace-write-proof", + "events": [ + { + "item": { + "type": "function_call", + "name": "shell_command", + "arguments": "{\"command\":\"printf 'tenant-write\\n' > sources/tenant/agent-write.txt && printf 'devkit-write\\n' > sources/devkit/agent-write.txt && if printf 'blocked\\n' > sources/runtime/blocked.txt 2>/dev/null; then managed=unexpected; else managed=denied; fi && if printf 'blocked\\n' > workspace-root-blocked.txt 2>/dev/null; then root=unexpected; else root=denied; fi && if printf 'blocked\\n' > ../outside-blocked.txt 2>/dev/null; then outside=unexpected; else outside=denied; fi && printf 'tenant=written devkit=written managed=%s root=%s outside=%s\\n' $managed $root $outside\"}", + "call_id": "call-odoo-workspace-write-proof" + } + } + ] + }, + { + "response_id": "resp-odoo-workspace-complete", + "events": [ + { + "item": { + "type": "message", + "id": "msg-odoo-workspace-complete", + "role": "assistant", + "status": "completed", + "content": [ + { + "type": "output_text", + "text": "Generated non-Git workspace boundary proof complete." + } + ] + } + } + ] + } + ] + }, + "expect": { + "returncode": 0, + "responses_request_count": 2, + "turn_count": 1, + "thread_id": "required", + "workspace_git": { + "git_root": null, + "branch": null, + "head_sha": null + }, + "launch_command": { + "contains_all": [ + "--skip-git-repo-check", + "--workspace-root" + ], + "not_contains": "--add-dir", + "count": { + "--workspace-root": 2 + } + }, + "responses": [ + { + "request": 0, + "scope": "input", + "contains_all": [ + "CANONICAL_ODOO_WORKSPACE_GUIDE", + "workspace.local.md", + "Preserve Launchplane ownership boundaries" + ], + "not_contains": "# Codex Lab" + } + ], + "tool_outputs": [ + { + "request": 1, + "call_id": "call-odoo-workspace-write-proof", + "contains_all": [ + "tenant=written", + "devkit=written", + "managed=denied", + "root=denied", + "outside=denied" + ] + } + ], + "workspace_paths": [ + { + "path": ".git", + "exists": false + }, + { + "path": "sources/tenant/agent-write.txt", + "type": "file", + "contains": "tenant-write" + }, + { + "path": "sources/devkit/agent-write.txt", + "type": "file", + "contains": "devkit-write" + }, + { + "path": "sources/runtime/blocked.txt", + "exists": false + }, + { + "path": "workspace-root-blocked.txt", + "exists": false + } + ] + }, + "timeout_seconds": 60 +} diff --git a/tools/codex-exec-harness/test_harness.py b/tools/codex-exec-harness/test_harness.py index 8e35220a5f8..f7196236007 100644 --- a/tools/codex-exec-harness/test_harness.py +++ b/tools/codex-exec-harness/test_harness.py @@ -4,6 +4,7 @@ import importlib.util import json import os +import subprocess import sys import tempfile import unittest @@ -341,7 +342,7 @@ def test_build_command_carries_sandbox_on_resume(self) -> None: sandbox_index = command.index("--sandbox") self.assertLess(sandbox_index, resume_index) self.assertEqual(command[sandbox_index + 1], "read-only") - self.assertEqual(command[-2:], ["session-1", "next"]) + self.assertEqual(command[-3:], ["session-1", "--", "next"]) def test_materialize_workspace_rejects_escaping_paths(self) -> None: with tempfile.TemporaryDirectory() as tmp: @@ -519,6 +520,175 @@ def test_token_usage_expectations_accept_bounds_and_cache_ratio(self) -> None: self.assertEqual([], failures) + +class GeneratedWorkspaceFixtureTest(unittest.TestCase): + def test_make_paths_can_materialize_workspace_outside_git(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + root = Path(temporary_directory) + repository = root / "repo" + output_root = repository / ".tmp" + external_root = root / "external" + repository.mkdir() + subprocess.run(["git", "init", "-q"], cwd=repository, check=True) + + with unittest.mock.patch.dict( + os.environ, + {"CODEX_EXEC_HARNESS_EXTERNAL_WORKSPACE_ROOT": str(external_root)}, + ): + paths = HARNESS.make_paths( + output_root, + "generated-workspace", + workspace_outside_git=True, + ) + paths.workspace.mkdir(parents=True) + + self.assertFalse(paths.workspace.is_relative_to(repository)) + self.assertIsNone(HARNESS.collect_workspace_git_state(paths.workspace)["git_root"]) + + (external_root / "AGENTS.md").write_text( + "# Ancestor guidance\n", encoding="utf-8" + ) + self.assertEqual( + HARNESS.ancestor_guidance_paths(paths.workspace), + [external_root.resolve() / "AGENTS.md"], + ) + + def test_materializes_external_sources_and_workspace_symlinks(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + paths = HARNESS.make_paths(Path(temporary_directory), "generated-workspace") + + HARNESS.materialize_workspace( + { + "git_init": False, + "files": {"AGENTS.md": "canonical guide\n"}, + "external_files": {"tenant/README.md": "tenant source\n"}, + "symlinks": {"sources/tenant": "{external}/tenant"}, + }, + paths, + ) + + source_link = paths.workspace / "sources" / "tenant" + self.assertTrue(source_link.is_symlink()) + self.assertEqual( + source_link.resolve(strict=True), + (paths.run_dir / "external" / "tenant").resolve(strict=True), + ) + self.assertFalse((paths.workspace / ".git").exists()) + + def test_build_command_supports_exact_workspace_roots(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + paths = HARNESS.make_paths(Path(temporary_directory), "generated-workspace") + tenant_root = paths.run_dir / "external" / "tenant" + tenant_root.mkdir(parents=True) + + command = HARNESS.build_command( + { + "sandbox": "workspace-write", + "workspace_roots": ["{external}/tenant"], + "config_overrides": ["fixture_root={external}/tenant"], + }, + {"prompt": "inspect"}, + "/tmp/codex-lab", + paths, + None, + ) + + self.assertIn("--sandbox", command) + self.assertIn("workspace-write", command) + self.assertIn("--workspace-root", command) + self.assertIn(str(tenant_root.resolve(strict=True)), command) + self.assertIn(f"fixture_root={tenant_root}", command) + + def test_build_command_rejects_mixed_or_unrestricted_workspace_roots(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + paths = HARNESS.make_paths(Path(temporary_directory), "generated-workspace") + tenant_root = paths.run_dir / "external" / "tenant" + tenant_root.mkdir(parents=True) + + with self.assertRaisesRegex(HARNESS.HarnessError, "sandbox=workspace-write"): + HARNESS.build_command( + {"sandbox": None, "workspace_roots": ["{external}/tenant"]}, + {"prompt": "inspect"}, + "/tmp/codex-lab", + paths, + None, + ) + + with self.assertRaisesRegex(HARNESS.HarnessError, "cannot be combined"): + HARNESS.build_command( + { + "sandbox": "workspace-write", + "add_dirs": ["{external}/tenant"], + "workspace_roots": ["{external}/tenant"], + }, + {"prompt": "inspect"}, + "/tmp/codex-lab", + paths, + None, + ) + + def test_resume_command_preserves_exact_workspace_roots(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + paths = HARNESS.make_paths(Path(temporary_directory), "generated-workspace") + tenant_root = paths.run_dir / "external" / "tenant" + tenant_root.mkdir(parents=True) + + command = HARNESS.build_command( + { + "sandbox": "workspace-write", + "workspace_roots": ["{external}/tenant"], + }, + {"prompt": "continue"}, + "/tmp/codex-lab", + paths, + "thread-123", + ) + + self.assertIn("-C", command) + self.assertIn(str(paths.workspace), command) + self.assertIn("--workspace-root", command) + self.assertIn(str(tenant_root.resolve(strict=True)), command) + self.assertLess(command.index("--workspace-root"), command.index("resume")) + self.assertEqual(command[-3:], ["thread-123", "--", "continue"]) + + def test_build_command_separates_option_like_prompt(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + paths = HARNESS.make_paths(Path(temporary_directory), "option-like-prompt") + + command = HARNESS.build_command( + {"sandbox": "read-only"}, + {"prompt": "--dangerously-bypass-approvals-and-sandbox"}, + "/tmp/codex-lab", + paths, + None, + ) + + self.assertEqual( + command[-2:], + ["--", "--dangerously-bypass-approvals-and-sandbox"], + ) + + def test_workspace_path_assertions_follow_declared_links_without_escaping(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + workspace = Path(temporary_directory) / "workspace" + external = Path(temporary_directory) / "external" + workspace.mkdir() + external.mkdir() + (external / "proof.txt").write_text("written\n", encoding="utf-8") + (workspace / "tenant").symlink_to(external, target_is_directory=True) + failures: list[str] = [] + + HARNESS.add_workspace_path_assertion_failures( + failures, + {"workspace": str(workspace)}, + [ + {"path": "tenant/proof.txt", "type": "file", "contains": "written"}, + {"path": "blocked.txt", "exists": False}, + ], + ) + + self.assertEqual([], failures) + def test_token_usage_expectations_report_prompt_bloat_and_cache_miss(self) -> None: failures = HARNESS.evaluate_expectations( { diff --git a/tools/codex-exec-harness/test_odoo_workspace.py b/tools/codex-exec-harness/test_odoo_workspace.py new file mode 100644 index 00000000000..a435ef85139 --- /dev/null +++ b/tools/codex-exec-harness/test_odoo_workspace.py @@ -0,0 +1,462 @@ +#!/usr/bin/env python3 +"""Regression tests for generated Odoo workspace launching.""" + +import importlib.util +import hashlib +import json +import subprocess +import sys +import tempfile +import unittest +import unittest.mock +from pathlib import Path + + +MODULE_PATH = Path(__file__).with_name("odoo_workspace.py") +SPEC = importlib.util.spec_from_file_location("odoo_workspace", MODULE_PATH) +if SPEC is None or SPEC.loader is None: + raise RuntimeError(f"failed to load {MODULE_PATH}") +ODOO_WORKSPACE = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = ODOO_WORKSPACE +SPEC.loader.exec_module(ODOO_WORKSPACE) + + +class WorkspaceFixture: + def __init__(self, root: Path) -> None: + self.root = root + self.workspace = root / "workspace" + self.workspace.mkdir() + self.sources = self.workspace / "sources" + self.sources.mkdir() + self.tenant = root / "tenant" + self.devkit = root / "devkit" + self.tenant.mkdir() + self.devkit.mkdir() + (self.sources / "tenant").symlink_to(self.tenant, target_is_directory=True) + (self.sources / "devkit").symlink_to(self.devkit, target_is_directory=True) + self.runtime = self.sources / "runtime" + self.runtime.mkdir() + self.manifest = self.tenant / "workspace.toml" + self.manifest.write_text("schema_version = 1\n", encoding="utf-8") + self.agents = self.workspace / "AGENTS.md" + self.agents.write_text("# Canonical Odoo workspace guide\n", encoding="utf-8") + self.local_notes = self.workspace / "workspace.local.md" + self.local_notes.write_text("local supplemental notes\n", encoding="utf-8") + + def payload(self) -> dict[str, object]: + sources = [ + self._source("tenant", self.tenant, "linked_path", True), + self._source("devkit", self.devkit, "linked_path", True), + self._source("runtime", self.runtime, "managed_checkout", False), + ] + return { + "schema_version": 1, + "workspace_path": str(self.workspace), + "workspace_exists": True, + "current": True, + "stale_reasons": [], + "lock_file_exists": True, + "lock_file_current": True, + "manifest": { + "path": str(self.manifest), + "sha256": hashlib.sha256(self.manifest.read_bytes()).hexdigest(), + "current": True, + }, + "surface_current": True, + "materialization_current": True, + "source_baseline_current": True, + "managed_source_baseline_current": True, + "sources": sources, + "edit_roots": [ + { + "role": source["role"], + "workspace_relative_path": source["workspace_relative_path"], + "resolved_path": source["resolved_path"], + } + for source in sources + if source["editable"] + ], + "local_notes": { + "path": str(self.local_notes), + "exists": True, + "valid": True, + "semantics": "supplemental_non_secret_notes", + }, + "reserved_override": { + "path": str(self.workspace / "AGENTS.override.md"), + "exists": False, + "semantics": "full_replacement", + "allowed_in_normal_flow": False, + }, + "workspace_agents_path": str(self.agents), + "workspace_agents_exists": True, + } + + def _source( + self, + role: str, + resolved_path: Path, + materialization: str, + editable: bool, + ) -> dict[str, object]: + relative_path = f"sources/{role}" + return { + "role": role, + "workspace_relative_path": relative_path, + "workspace_entry_path": str(self.workspace / relative_path), + "resolved_path": str(resolved_path.resolve()), + "actual_resolved_path": str(resolved_path.resolve()), + "materialization": materialization, + "materialization_state": "current", + "materialization_current": True, + "editable": editable, + } + + +class OdooWorkspaceValidationTest(unittest.TestCase): + def test_current_non_git_workspace_exposes_only_declared_edit_roots(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + fixture = WorkspaceFixture(Path(temporary_directory)) + + launch = ODOO_WORKSPACE.validate_workspace_status( + fixture.payload(), + manifest_path=fixture.manifest, + status_command=("uv", "workspace", "status", "--check"), + ) + + self.assertEqual(launch.workspace_path, fixture.workspace.resolve()) + self.assertIsNone(launch.git_root) + self.assertEqual( + launch.writable_roots, + (fixture.tenant.resolve(), fixture.devkit.resolve()), + ) + self.assertEqual([source.role for source in launch.managed_sources], ["runtime"]) + + def test_reserved_override_fails_closed(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + fixture = WorkspaceFixture(Path(temporary_directory)) + payload = fixture.payload() + payload["reserved_override"] = { + "path": str(fixture.workspace / "AGENTS.override.md"), + "exists": True, + "semantics": "full_replacement", + "allowed_in_normal_flow": False, + } + + with self.assertRaisesRegex(ODOO_WORKSPACE.OdooWorkspaceError, "shadowed"): + ODOO_WORKSPACE.validate_workspace_status( + payload, + manifest_path=fixture.manifest, + status_command=("uv",), + ) + + def test_editable_source_baseline_drift_does_not_block_current_workspace(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + fixture = WorkspaceFixture(Path(temporary_directory)) + payload = fixture.payload() + payload["source_baseline_current"] = False + + launch = ODOO_WORKSPACE.validate_workspace_status( + payload, + manifest_path=fixture.manifest, + status_command=("uv",), + ) + + self.assertEqual( + launch.writable_roots, + (fixture.tenant.resolve(), fixture.devkit.resolve()), + ) + + def test_redirected_workspace_entry_fails_closed(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + fixture = WorkspaceFixture(Path(temporary_directory)) + payload = fixture.payload() + payload["sources"][0]["workspace_entry_path"] = str(fixture.tenant) + + with self.assertRaisesRegex(ODOO_WORKSPACE.OdooWorkspaceError, "redirected"): + ODOO_WORKSPACE.validate_workspace_status( + payload, + manifest_path=fixture.manifest, + status_command=("uv",), + ) + + def test_redirected_local_notes_fail_closed(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + fixture = WorkspaceFixture(Path(temporary_directory)) + redirected_notes = fixture.root / "redirected-notes.md" + redirected_notes.write_text("outside guidance\n", encoding="utf-8") + fixture.local_notes.unlink() + fixture.local_notes.symlink_to(redirected_notes) + + with self.assertRaisesRegex(ODOO_WORKSPACE.OdooWorkspaceError, "regular file"): + ODOO_WORKSPACE.validate_workspace_status( + fixture.payload(), + manifest_path=fixture.manifest, + status_command=("uv",), + ) + + def test_managed_checkout_cannot_be_declared_editable(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + fixture = WorkspaceFixture(Path(temporary_directory)) + payload = fixture.payload() + payload["sources"][2]["editable"] = True + payload["edit_roots"].append( + { + "role": "runtime", + "workspace_relative_path": "sources/runtime", + "resolved_path": str(fixture.runtime), + } + ) + + with self.assertRaisesRegex(ODOO_WORKSPACE.OdooWorkspaceError, "cannot be editable"): + ODOO_WORKSPACE.validate_workspace_status( + payload, + manifest_path=fixture.manifest, + status_command=("uv",), + ) + + def test_editable_source_cannot_overlap_generated_workspace(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + fixture = WorkspaceFixture(Path(temporary_directory)) + payload = fixture.payload() + tenant_link = fixture.sources / "tenant" + tenant_link.unlink() + tenant_link.symlink_to(fixture.root, target_is_directory=True) + redirected_root = str(fixture.root.resolve()) + payload["sources"][0]["resolved_path"] = redirected_root + payload["sources"][0]["actual_resolved_path"] = redirected_root + payload["edit_roots"][0]["resolved_path"] = redirected_root + + with self.assertRaisesRegex(ODOO_WORKSPACE.OdooWorkspaceError, "generated workspace"): + ODOO_WORKSPACE.validate_workspace_status( + payload, + manifest_path=fixture.manifest, + status_command=("uv",), + ) + + def test_editable_source_cannot_overlap_read_only_source(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + fixture = WorkspaceFixture(Path(temporary_directory)) + shared = fixture.tenant / "shared" + shared.mkdir() + (fixture.sources / "shared").symlink_to(shared, target_is_directory=True) + payload = fixture.payload() + payload["sources"].append( + fixture._source("shared", shared, "linked_path", False) + ) + + with self.assertRaisesRegex(ODOO_WORKSPACE.OdooWorkspaceError, "read-only source"): + ODOO_WORKSPACE.validate_workspace_status( + payload, + manifest_path=fixture.manifest, + status_command=("uv",), + ) + + def test_workspace_inside_git_repository_is_rejected(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + fixture = WorkspaceFixture(Path(temporary_directory)) + subprocess.run( + ["git", "init", "-q"], + cwd=fixture.workspace, + check=True, + ) + + with self.assertRaisesRegex(ODOO_WORKSPACE.OdooWorkspaceError, "outside a Git work tree"): + ODOO_WORKSPACE.validate_workspace_status( + fixture.payload(), + manifest_path=fixture.manifest, + status_command=("uv",), + ) + + def test_workspace_below_ancestor_guidance_is_rejected(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + fixture = WorkspaceFixture(Path(temporary_directory)) + (fixture.root / "AGENTS.md").write_text( + "# Untrusted ancestor guidance\n", encoding="utf-8" + ) + + with self.assertRaisesRegex(ODOO_WORKSPACE.OdooWorkspaceError, "ancestor guidance"): + ODOO_WORKSPACE.validate_workspace_status( + fixture.payload(), + manifest_path=fixture.manifest, + status_command=("uv",), + ) + + +class OdooWorkspaceCommandTest(unittest.TestCase): + def _launch(self, fixture: WorkspaceFixture): + return ODOO_WORKSPACE.validate_workspace_status( + fixture.payload(), + manifest_path=fixture.manifest, + status_command=("uv", "workspace", "status", "--check"), + ) + + def test_exec_uses_exact_workspace_roots_without_writable_workspace_cwd(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + fixture = WorkspaceFixture(Path(temporary_directory)) + codex_bin = fixture.root / "codex-lab" + codex_bin.write_text("binary", encoding="utf-8") + + command = ODOO_WORKSPACE.build_codex_command( + launch=self._launch(fixture), + codex_bin=codex_bin, + mode="exec", + access="editable", + prompt="inspect the workspace", + ) + + self.assertEqual(command[:4], (str(codex_bin), "exec", "--json", "--skip-git-repo-check")) + self.assertIn("--sandbox", command) + self.assertIn("workspace-write", command) + self.assertNotIn("--add-dir", command) + self.assertEqual(command.count("--workspace-root"), 2) + self.assertIn(str(fixture.tenant.resolve()), command) + self.assertIn(str(fixture.devkit.resolve()), command) + workspace_roots = [ + command[index + 1] + for index, value in enumerate(command) + if value == "--workspace-root" + ] + self.assertNotIn(str(fixture.workspace.resolve()), workspace_roots) + self.assertEqual(command[-2], "--") + self.assertEqual(command[-1], "inspect the workspace") + + def test_option_like_prompt_is_separated_from_codex_flags(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + fixture = WorkspaceFixture(Path(temporary_directory)) + codex_bin = fixture.root / "codex-lab" + codex_bin.write_text("binary", encoding="utf-8") + + command = ODOO_WORKSPACE.build_codex_command( + launch=self._launch(fixture), + codex_bin=codex_bin, + mode="exec", + access="editable", + prompt="--dangerously-bypass-approvals-and-sandbox", + ) + + self.assertEqual( + command[-2:], + ("--", "--dangerously-bypass-approvals-and-sandbox"), + ) + + def test_interactive_read_only_path_is_explicit(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + fixture = WorkspaceFixture(Path(temporary_directory)) + codex_bin = fixture.root / "codex-lab" + codex_bin.write_text("binary", encoding="utf-8") + + command = ODOO_WORKSPACE.build_codex_command( + launch=self._launch(fixture), + codex_bin=codex_bin, + mode="interactive", + access="read-only", + prompt=None, + ) + + self.assertEqual(command[0], str(codex_bin)) + self.assertNotIn("exec", command) + self.assertIn("--sandbox", command) + self.assertIn("read-only", command) + self.assertNotIn("--add-dir", command) + self.assertFalse(any("default_permissions" in value for value in command)) + + def test_evidence_redacts_prompt_and_records_exact_permissions(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + fixture = WorkspaceFixture(Path(temporary_directory)) + codex_bin = fixture.root / "codex-lab" + codex_bin.write_text("binary", encoding="utf-8") + launch = self._launch(fixture) + prompt = "private task text" + command = ODOO_WORKSPACE.build_codex_command( + launch=launch, + codex_bin=codex_bin, + mode="exec", + access="editable", + prompt=prompt, + ) + provenance = {"status": "current", "binary_sha256": "abc"} + + evidence = ODOO_WORKSPACE.build_evidence( + launch=launch, + codex_bin=codex_bin, + provenance=provenance, + command=command, + mode="exec", + access="editable", + prompt=prompt, + returncode=0, + ) + + serialized = json.dumps(evidence) + self.assertNotIn(prompt, serialized) + self.assertEqual(evidence["command"][-1], "") + self.assertFalse(evidence["permissions"]["workspace_root_writable"]) + self.assertEqual( + evidence["permissions"]["writable_roots"], + [str(fixture.tenant.resolve()), str(fixture.devkit.resolve())], + ) + self.assertEqual(evidence["codex_binary"]["provenance"], provenance) + + +class OdooWorkspaceProcessTest(unittest.TestCase): + def test_nonzero_status_uses_bounded_stale_reasons(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + root = Path(temporary_directory) + devkit = root / "devkit" + devkit.mkdir() + manifest = root / "workspace.toml" + manifest.write_text("schema_version = 1\n", encoding="utf-8") + completed = subprocess.CompletedProcess( + args=[], + returncode=2, + stdout=json.dumps( + { + "stale_reasons": [f"reason-{index}" for index in range(20)] + } + ), + stderr="ignored stderr", + ) + + with unittest.mock.patch.object( + ODOO_WORKSPACE, "resolve_executable", return_value=Path("/usr/bin/uv") + ), unittest.mock.patch.object( + ODOO_WORKSPACE.subprocess, "run", return_value=completed + ), self.assertRaisesRegex( + ODOO_WORKSPACE.OdooWorkspaceError, + r"reason-0.*reason-11, \.\.\.", + ): + ODOO_WORKSPACE.run_workspace_status( + uv_bin="uv", + devkit_path=devkit, + manifest_path=manifest, + ) + + def test_stale_binary_provenance_fails_closed(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + root = Path(temporary_directory) + script = root / "scripts" / "local" / "codex_lab_provenance.py" + script.parent.mkdir(parents=True) + script.write_text("print('unused')\n", encoding="utf-8") + binary = root / "codex-lab" + binary.write_text("binary", encoding="utf-8") + completed = subprocess.CompletedProcess( + args=[], + returncode=0, + stdout=json.dumps({"status": "stale", "failures": ["commit mismatch"]}), + stderr="", + ) + + with unittest.mock.patch.object( + ODOO_WORKSPACE.subprocess, "run", return_value=completed + ), self.assertRaisesRegex( + ODOO_WORKSPACE.OdooWorkspaceError, "provenance is not current" + ): + ODOO_WORKSPACE.verify_codex_provenance( + source_repo=root, + codex_bin=binary, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/codex-exec-harness/test_run_all.py b/tools/codex-exec-harness/test_run_all.py index c137011015a..8f4bbba7ad2 100644 --- a/tools/codex-exec-harness/test_run_all.py +++ b/tools/codex-exec-harness/test_run_all.py @@ -137,7 +137,7 @@ def fake_run(command, **_kwargs): self.assertTrue(report["passed"]) self.assertFalse(report["partial"]) self.assertEqual("/tmp/codex", report["codex_bin"]) - self.assertEqual(17, report["scenario_total"]) + self.assertEqual(18, report["scenario_total"]) self.assertEqual("abc123", report["git_revision"]) self.assertEqual("current", report["provenance"]["status"]) self.assertEqual("f" * 64, report["provenance"]["binary_sha256"])