diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 1951db642f..4eeb0d10b3 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3022,8 +3022,10 @@ name = "codex-hooks" version = "0.0.0" dependencies = [ "anyhow", + "async-trait", "chrono", "codex-config", + "codex-exec-server", "codex-plugin", "codex-protocol", "codex-utils-absolute-path", diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index b78690a4f6..83bd998df6 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -7958,6 +7958,12 @@ "null" ] }, + "environmentId": { + "type": [ + "string", + "null" + ] + }, "statusMessage": { "type": [ "string", @@ -9841,6 +9847,12 @@ "enabled": { "type": "boolean" }, + "environmentId": { + "type": [ + "string", + "null" + ] + }, "eventName": { "$ref": "#/definitions/v2/HookEventName" }, diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index b74ed83f8b..ead98e9a08 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -4327,6 +4327,12 @@ "null" ] }, + "environmentId": { + "type": [ + "string", + "null" + ] + }, "statusMessage": { "type": [ "string", @@ -6321,6 +6327,12 @@ "enabled": { "type": "boolean" }, + "environmentId": { + "type": [ + "string", + "null" + ] + }, "eventName": { "$ref": "#/definitions/HookEventName" }, diff --git a/codex-rs/app-server-protocol/schema/json/v2/ConfigRequirementsReadResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ConfigRequirementsReadResponse.json index a82421fb01..247b12f56b 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ConfigRequirementsReadResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ConfigRequirementsReadResponse.json @@ -169,6 +169,12 @@ "null" ] }, + "environmentId": { + "type": [ + "string", + "null" + ] + }, "statusMessage": { "type": [ "string", diff --git a/codex-rs/app-server-protocol/schema/json/v2/HooksListResponse.json b/codex-rs/app-server-protocol/schema/json/v2/HooksListResponse.json index c58a5c767d..2e97a37d71 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/HooksListResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/HooksListResponse.json @@ -61,6 +61,12 @@ "enabled": { "type": "boolean" }, + "environmentId": { + "type": [ + "string", + "null" + ] + }, "eventName": { "$ref": "#/definitions/HookEventName" }, diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ConfiguredHookHandler.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ConfiguredHookHandler.ts index 42b05cf8cc..04110ff0e5 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ConfiguredHookHandler.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ConfiguredHookHandler.ts @@ -2,4 +2,4 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. -export type ConfiguredHookHandler = { "type": "command", command: string, commandWindows: string | null, timeoutSec: bigint | null, async: boolean, statusMessage: string | null, } | { "type": "prompt", } | { "type": "agent", }; +export type ConfiguredHookHandler = { "type": "command", command: string, commandWindows: string | null, environmentId: string | null, timeoutSec: bigint | null, async: boolean, statusMessage: string | null, } | { "type": "prompt", } | { "type": "agent", }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/HookMetadata.ts b/codex-rs/app-server-protocol/schema/typescript/v2/HookMetadata.ts index 94e3c30c92..5eb67b4b07 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/HookMetadata.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/HookMetadata.ts @@ -7,4 +7,4 @@ import type { HookHandlerType } from "./HookHandlerType"; import type { HookSource } from "./HookSource"; import type { HookTrustStatus } from "./HookTrustStatus"; -export type HookMetadata = { key: string, eventName: HookEventName, handlerType: HookHandlerType, matcher: string | null, command: string | null, timeoutSec: bigint, statusMessage: string | null, sourcePath: AbsolutePathBuf, source: HookSource, pluginId: string | null, displayOrder: bigint, enabled: boolean, isManaged: boolean, currentHash: string, trustStatus: HookTrustStatus, }; +export type HookMetadata = { key: string, eventName: HookEventName, handlerType: HookHandlerType, matcher: string | null, command: string | null, environmentId: string | null, timeoutSec: bigint, statusMessage: string | null, sourcePath: AbsolutePathBuf, source: HookSource, pluginId: string | null, displayOrder: bigint, enabled: boolean, isManaged: boolean, currentHash: string, trustStatus: HookTrustStatus, }; diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index b7bdff813b..63451da2ea 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -1008,6 +1008,10 @@ "default": null, "type": "string" }, + "environmentId": { + "default": null, + "type": "string" + }, "statusMessage": { "default": null, "type": "string" diff --git a/codex-rs/core/tests/suite/remote_env.rs b/codex-rs/core/tests/suite/remote_env.rs index ac838f9564..91bbe44ea3 100644 --- a/codex-rs/core/tests/suite/remote_env.rs +++ b/codex-rs/core/tests/suite/remote_env.rs @@ -385,7 +385,7 @@ async fn structured_hook_routes_to_explicit_remote_environment() -> Result<()> { SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis() ); let hook_input_path_for_config = hook_input_path.clone(); - let builder = test_codex() + let mut builder = test_codex() .with_pre_build_hook(move |home| { let hooks = json!({ "hooks": { diff --git a/codex-rs/hooks/Cargo.toml b/codex-rs/hooks/Cargo.toml index 1bd7f455f2..874df2d2d9 100644 --- a/codex-rs/hooks/Cargo.toml +++ b/codex-rs/hooks/Cargo.toml @@ -16,6 +16,7 @@ workspace = true anyhow = { workspace = true } chrono = { workspace = true, features = ["serde"] } codex-config = { workspace = true } +codex-exec-server = { workspace = true } codex-plugin = { workspace = true } codex-protocol = { workspace = true } codex-utils-absolute-path = { workspace = true } @@ -30,6 +31,7 @@ tracing = { workspace = true } uuid = { workspace = true, features = ["v4"] } [dev-dependencies] +async-trait = { workspace = true } pretty_assertions = { workspace = true } tempfile = { workspace = true } tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time"] } diff --git a/codex-rs/hooks/src/engine/command_runner.rs b/codex-rs/hooks/src/engine/command_runner.rs index 8b0b11eb2a..e58a9466cc 100644 --- a/codex-rs/hooks/src/engine/command_runner.rs +++ b/codex-rs/hooks/src/engine/command_runner.rs @@ -43,7 +43,7 @@ pub(crate) async fn run_command( ); }; if environment.is_remote() { - return run_remote_command(handler, environment, input_json, cwd).await; + return run_remote_hook_command(handler, environment, input_json, cwd).await; } } @@ -97,7 +97,7 @@ pub(crate) async fn run_command( } } -async fn run_remote_command( +async fn run_remote_hook_command( handler: &ConfiguredHandler, environment: std::sync::Arc, input_json: &str, @@ -122,6 +122,16 @@ async fn run_remote_command( Ok(started) => started.process, Err(err) => return command_error(started_at, started, err.to_string()), }; + run_started_remote_hook_command(process, handler, input_json, started_at, started).await +} + +async fn run_started_remote_hook_command( + process: std::sync::Arc, + handler: &ConfiguredHandler, + input_json: &str, + started_at: i64, + started: Instant, +) -> CommandRunResult { if let Err(error) = write_remote_stdin(&process, input_json).await { let _ = process.terminate().await; return command_error(started_at, started, error); @@ -279,16 +289,125 @@ fn default_shell_command() -> Command { #[cfg(test)] mod tests { + use std::collections::VecDeque; + use std::future::pending; + use std::sync::Arc; + use std::sync::atomic::AtomicUsize; + use std::sync::atomic::Ordering; + use std::time::Instant; + + use async_trait::async_trait; use codex_exec_server::EnvironmentManager; + use codex_exec_server::ExecProcessEventReceiver; + use codex_exec_server::ExecServerError; + use codex_exec_server::ProcessOutputChunk; + use codex_exec_server::ReadResponse; + use codex_exec_server::WriteResponse; + use codex_exec_server::WriteStatus; use codex_protocol::protocol::HookEventName; use codex_protocol::protocol::HookSource; use codex_utils_absolute_path::test_support::PathBufExt; use codex_utils_absolute_path::test_support::test_path_buf; use pretty_assertions::assert_eq; + use tokio::sync::Mutex; + use tokio::sync::watch; use super::CommandShell; use super::ConfiguredHandler; + use super::collect_output; use super::run_command; + use super::run_started_remote_hook_command; + use super::write_remote_stdin; + + struct MockExecProcess { + process_id: codex_exec_server::ProcessId, + write_response: WriteResponse, + read_responses: Mutex>>, + block_reads: bool, + writes: Mutex>, bool)>>, + terminate_calls: AtomicUsize, + wake_tx: watch::Sender, + } + + impl MockExecProcess { + fn new( + write_status: WriteStatus, + read_responses: Vec>, + ) -> Arc { + let (wake_tx, _wake_rx) = watch::channel(0); + Arc::new(Self { + process_id: "hook-process".to_string().into(), + write_response: WriteResponse { + status: write_status, + }, + read_responses: Mutex::new(VecDeque::from(read_responses)), + block_reads: false, + writes: Mutex::new(Vec::new()), + terminate_calls: AtomicUsize::new(0), + wake_tx, + }) + } + + fn blocking() -> Arc { + let (wake_tx, _wake_rx) = watch::channel(0); + Arc::new(Self { + process_id: "hook-process".to_string().into(), + write_response: WriteResponse { + status: WriteStatus::Accepted, + }, + read_responses: Mutex::new(VecDeque::new()), + block_reads: true, + writes: Mutex::new(Vec::new()), + terminate_calls: AtomicUsize::new(0), + wake_tx, + }) + } + } + + #[async_trait] + impl codex_exec_server::ExecProcess for MockExecProcess { + fn process_id(&self) -> &codex_exec_server::ProcessId { + &self.process_id + } + + fn subscribe_wake(&self) -> watch::Receiver { + self.wake_tx.subscribe() + } + + fn subscribe_events(&self) -> ExecProcessEventReceiver { + ExecProcessEventReceiver::empty() + } + + async fn read( + &self, + _after_seq: Option, + _max_bytes: Option, + _wait_ms: Option, + ) -> Result { + if self.block_reads { + return pending().await; + } + self.read_responses + .lock() + .await + .pop_front() + .unwrap_or_else(|| Ok(closed_read_response(Vec::new(), Some(0)))) + } + + async fn write( + &self, + chunk: Option>, + close_stdin: bool, + ) -> Result { + self.writes.lock().await.push((chunk, close_stdin)); + Ok(self.write_response.clone()) + } + + async fn terminate(&self) -> Result<(), ExecServerError> { + self.terminate_calls.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + } fn shell(environment_manager: std::sync::Arc) -> CommandShell { CommandShell { @@ -313,6 +432,42 @@ mod tests { } } + fn read_response( + chunks: Vec, + next_seq: u64, + exited: bool, + exit_code: Option, + closed: bool, + ) -> ReadResponse { + ReadResponse { + chunks, + next_seq, + exited, + exit_code, + closed, + failure: None, + } + } + + fn closed_read_response( + chunks: Vec, + exit_code: Option, + ) -> ReadResponse { + read_response(chunks, 4, true, exit_code, true) + } + + fn output_chunk( + seq: u64, + stream: codex_exec_server::ExecOutputStream, + chunk: &[u8], + ) -> ProcessOutputChunk { + ProcessOutputChunk { + seq, + stream, + chunk: chunk.to_vec().into(), + } + } + #[tokio::test] async fn omitted_environment_id_runs_locally() { let result = run_command( @@ -328,6 +483,55 @@ mod tests { assert_eq!(result.error, None); } + #[tokio::test] + async fn explicit_local_environment_id_runs_locally() { + let result = run_command( + &shell(std::sync::Arc::new(EnvironmentManager::default_for_tests())), + &handler("printf explicit-local-hook", Some("local")), + "{}", + test_path_buf("/tmp").as_path(), + ) + .await; + + assert_eq!(result.exit_code, Some(0)); + assert_eq!(result.stdout, "explicit-local-hook"); + assert_eq!(result.error, None); + } + + #[tokio::test] + async fn local_hook_receives_stdin_and_captures_stderr_and_exit_code() { + let result = run_command( + &shell(std::sync::Arc::new(EnvironmentManager::default_for_tests())), + &handler("cat; printf stderr >&2; exit 7", None), + "{\"hook\":true}", + test_path_buf("/tmp").as_path(), + ) + .await; + + assert_eq!(result.exit_code, Some(7)); + assert_eq!(result.stdout, "{\"hook\":true}"); + assert_eq!(result.stderr, "stderr"); + assert_eq!(result.error, None); + } + + #[tokio::test] + async fn local_hook_timeout_returns_error() { + let mut handler = handler("sleep 5", None); + handler.timeout_sec = 1; + let result = run_command( + &shell(std::sync::Arc::new(EnvironmentManager::default_for_tests())), + &handler, + "{}", + test_path_buf("/tmp").as_path(), + ) + .await; + + assert_eq!(result.exit_code, None); + assert_eq!(result.stdout, ""); + assert_eq!(result.stderr, ""); + assert_eq!(result.error.as_deref(), Some("hook timed out after 1s")); + } + #[tokio::test] async fn unknown_environment_id_does_not_fall_back_to_local() { let result = run_command( @@ -369,4 +573,130 @@ mod tests { .is_some_and(|error| error.contains("failed")) ); } + + #[tokio::test] + async fn remote_hook_stdin_writes_payload_and_closes_stdin() { + let process = MockExecProcess::new(WriteStatus::Accepted, Vec::new()); + + write_remote_stdin( + &(process.clone() as Arc), + "{\"hook\":true}", + ) + .await + .expect("remote hook stdin should be accepted"); + + assert_eq!( + process.writes.lock().await.as_slice(), + &[(Some(b"{\"hook\":true}".to_vec()), true)] + ); + } + + #[tokio::test] + async fn remote_hook_stdin_rejects_non_accepted_write_status() { + let process = MockExecProcess::new(WriteStatus::StdinClosed, Vec::new()); + + let err = write_remote_stdin( + &(process.clone() as Arc), + "{}", + ) + .await + .expect_err("closed stdin should fail"); + + assert_eq!(err, "failed to write hook stdin: StdinClosed"); + } + + #[tokio::test] + async fn remote_hook_collects_stdout_stderr_and_pty_output() { + let process = MockExecProcess::new( + WriteStatus::Accepted, + vec![Ok(closed_read_response( + vec![ + output_chunk(1, codex_exec_server::ExecOutputStream::Stdout, b"stdout"), + output_chunk(2, codex_exec_server::ExecOutputStream::Stderr, b"stderr"), + output_chunk(3, codex_exec_server::ExecOutputStream::Pty, b"pty"), + ], + Some(0), + ))], + ); + + let actual = collect_output(process as Arc) + .await + .expect("remote output should collect"); + + assert_eq!( + actual, + ("stdoutpty".to_string(), "stderr".to_string(), Some(0)) + ); + } + + #[tokio::test] + async fn remote_hook_collect_output_surfaces_process_failure() { + let mut response = closed_read_response(Vec::new(), None); + response.failure = Some("transport disconnected".to_string()); + let process = MockExecProcess::new(WriteStatus::Accepted, vec![Ok(response)]); + + let err = collect_output(process as Arc) + .await + .expect_err("remote output failure should surface"); + + assert_eq!(err, "transport disconnected"); + } + + #[tokio::test] + async fn remote_hook_write_failure_terminates_process() { + let process = MockExecProcess::new(WriteStatus::StdinClosed, Vec::new()); + let result = run_started_remote_hook_command( + process.clone() as Arc, + &handler("printf ignored", Some("remote-hook")), + "{}", + chrono::Utc::now().timestamp(), + Instant::now(), + ) + .await; + + assert_eq!(result.exit_code, None); + assert_eq!( + result.error.as_deref(), + Some("failed to write hook stdin: StdinClosed") + ); + assert_eq!(process.terminate_calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn remote_hook_read_failure_terminates_process() { + let mut response = closed_read_response(Vec::new(), None); + response.failure = Some("transport disconnected".to_string()); + let process = MockExecProcess::new(WriteStatus::Accepted, vec![Ok(response)]); + let result = run_started_remote_hook_command( + process.clone() as Arc, + &handler("printf ignored", Some("remote-hook")), + "{}", + chrono::Utc::now().timestamp(), + Instant::now(), + ) + .await; + + assert_eq!(result.exit_code, None); + assert_eq!(result.error.as_deref(), Some("transport disconnected")); + assert_eq!(process.terminate_calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn remote_hook_timeout_terminates_process() { + let process = MockExecProcess::blocking(); + let mut handler = handler("printf ignored", Some("remote-hook")); + handler.timeout_sec = 0; + let result = run_started_remote_hook_command( + process.clone() as Arc, + &handler, + "{}", + chrono::Utc::now().timestamp(), + Instant::now(), + ) + .await; + + assert_eq!(result.exit_code, None); + assert_eq!(result.error.as_deref(), Some("hook timed out after 0s")); + assert_eq!(process.terminate_calls.load(Ordering::SeqCst), 1); + } } diff --git a/codex-rs/hooks/src/engine/mod_tests.rs b/codex-rs/hooks/src/engine/mod_tests.rs index 738ff2ddde..c9953c1c3b 100644 --- a/codex-rs/hooks/src/engine/mod_tests.rs +++ b/codex-rs/hooks/src/engine/mod_tests.rs @@ -226,6 +226,7 @@ with Path(r"{log_path}").open("a", encoding="utf-8") as handle: plugin_hook_load_warnings: Vec::new(), shell_program: None, shell_args: Vec::new(), + environment_manager: std::sync::Arc::new(EnvironmentManager::default_for_tests()), }); assert!(listed.hooks[0].is_managed); let cwd = cwd(); @@ -1190,6 +1191,7 @@ print(json.dumps({ plugin_hook_load_warnings: Vec::new(), shell_program: None, shell_args: Vec::new(), + environment_manager: std::sync::Arc::new(EnvironmentManager::default_for_tests()), }); assert_eq!( listed.hooks[0].plugin_id.as_deref(),