Add persisted hook enablement state (#19840)

## Why

After `hooks/list` exposes the hook inventory, clients need a way to
persist user hook preferences, make those changes effective in
already-open sessions, and distinguish user-controllable hooks from
managed requirements without adding another bespoke app-server write
API.

## What

- Extends `hooks/list` entries with effective `enabled` state.
- Persists user-level hook state under `hooks.state.<hook-id>` so the
model can grow beyond a single boolean over time.
- Uses the existing `config/batchWrite` path for hook state updates
instead of introducing a dedicated hook write RPC.
- Refreshes live session hook engines after config writes so
already-open threads observe updated enablement without a restart.

## Stack

1. openai/codex#19705
2. openai/codex#19778
3. This PR - openai/codex#19840
4. openai/codex#19882

## Reviewer Notes

The generated schema files account for much of the raw diff. The core
behavior is in:

- `hooks/src/config_rules.rs`, which resolves per-hook user state from
the config layer stack.
- `hooks/src/engine/discovery.rs`, which projects effective enablement
into `hooks/list` from source-derived managedness.
- `config/src/hook_config.rs`, which defines the new `hooks.state`
representation.
- `core/src/session/mod.rs`, which rebuilds live hook state after user
config reloads.

---------

Co-authored-by: Codex <noreply@openai.com>

codex-rs/app-server/tests/suite/v2/hooks_list.rs

@@ -2,7 +2,11 @@ use std::time::Duration;
 
 use anyhow::Result;
 use app_test_support::McpProcess;
+use app_test_support::create_final_assistant_message_sse_response;
+use app_test_support::create_mock_responses_server_sequence_unchecked;
 use app_test_support::to_response;
+use codex_app_server_protocol::ConfigBatchWriteParams;
+use codex_app_server_protocol::ConfigEdit;
 use codex_app_server_protocol::HookEventName;
 use codex_app_server_protocol::HookHandlerType;
 use codex_app_server_protocol::HookMetadata;
@@ -11,10 +15,16 @@ use codex_app_server_protocol::HooksListEntry;
 use codex_app_server_protocol::HooksListParams;
 use codex_app_server_protocol::HooksListResponse;
 use codex_app_server_protocol::JSONRPCResponse;
+use codex_app_server_protocol::MergeStrategy;
 use codex_app_server_protocol::RequestId;
+use codex_app_server_protocol::ThreadStartParams;
+use codex_app_server_protocol::ThreadStartResponse;
+use codex_app_server_protocol::TurnStartParams;
+use codex_app_server_protocol::UserInput as V2UserInput;
 use codex_core::config::set_project_trust_level;
 use codex_protocol::config_types::TrustLevel;
 use codex_utils_absolute_path::AbsolutePathBuf;
+use core_test_support::skip_if_windows;
 use pretty_assertions::assert_eq;
 use tempfile::TempDir;
 use tokio::time::timeout;
@@ -82,23 +92,27 @@ async fn hooks_list_shows_discovered_hook() -> Result<()> {
     )
     .await??;
     let HooksListResponse { data } = to_response(response)?;
+    let config_path = AbsolutePathBuf::from_absolute_path(std::fs::canonicalize(
+        codex_home.path().join("config.toml"),
+    )?)?;
     assert_eq!(
         data,
         vec![HooksListEntry {
             cwd: cwd.path().to_path_buf(),
             hooks: vec![HookMetadata {
+                key: format!("{}:pre_tool_use:0:0", config_path.as_path().display()),
                 event_name: HookEventName::PreToolUse,
                 handler_type: HookHandlerType::Command,
                 matcher: Some("Bash".to_string()),
                 command: Some("python3 /tmp/listed-hook.py".to_string()),
                 timeout_sec: 5,
                 status_message: Some("running listed hook".to_string()),
-                source_path: AbsolutePathBuf::from_absolute_path(std::fs::canonicalize(
-                    codex_home.path().join("config.toml")
-                )?,)?,
+                source_path: config_path,
                 source: HookSource::User,
                 plugin_id: None,
                 display_order: 0,
+                enabled: true,
+                is_managed: false,
             }],
             warnings: Vec::new(),
             errors: Vec::new(),
@@ -146,25 +160,29 @@ async fn hooks_list_shows_discovered_plugin_hook() -> Result<()> {
     )
     .await??;
     let HooksListResponse { data } = to_response(response)?;
+    let plugin_hooks_path = AbsolutePathBuf::from_absolute_path(std::fs::canonicalize(
+        codex_home
+            .path()
+            .join("plugins/cache/test/demo/local/hooks/hooks.json"),
+    )?)?;
     assert_eq!(
         data,
         vec![HooksListEntry {
             cwd: cwd.path().to_path_buf(),
             hooks: vec![HookMetadata {
+                key: "demo@test:hooks/hooks.json:pre_tool_use:0:0".to_string(),
                 event_name: HookEventName::PreToolUse,
                 handler_type: HookHandlerType::Command,
                 matcher: Some("Bash".to_string()),
                 command: Some("echo plugin hook".to_string()),
                 timeout_sec: 7,
                 status_message: Some("running plugin hook".to_string()),
-                source_path: AbsolutePathBuf::from_absolute_path(std::fs::canonicalize(
-                    codex_home
-                        .path()
-                        .join("plugins/cache/test/demo/local/hooks/hooks.json"),
-                )?,)?,
+                source_path: plugin_hooks_path,
                 source: HookSource::Plugin,
                 plugin_id: Some("demo@test".to_string()),
                 display_order: 0,
+                enabled: true,
+                is_managed: false,
             }],
             warnings: Vec::new(),
             errors: Vec::new(),
@@ -252,6 +270,8 @@ timeout = 5
     )
     .await??;
     let HooksListResponse { data } = to_response(response)?;
+    let project_config_path =
+        AbsolutePathBuf::try_from(workspace.path().join(".codex/config.toml"))?;
     assert_eq!(
         data,
         vec![
@@ -264,18 +284,22 @@ timeout = 5
             HooksListEntry {
                 cwd: workspace.path().to_path_buf(),
                 hooks: vec![HookMetadata {
+                    key: format!(
+                        "{}:pre_tool_use:0:0",
+                        project_config_path.as_path().display()
+                    ),
                     event_name: HookEventName::PreToolUse,
                     handler_type: HookHandlerType::Command,
                     matcher: Some("Bash".to_string()),
                     command: Some("echo project hook".to_string()),
                     timeout_sec: 5,
                     status_message: None,
-                    source_path: AbsolutePathBuf::try_from(
-                        workspace.path().join(".codex/config.toml"),
-                    )?,
+                    source_path: project_config_path,
                     source: HookSource::Project,
                     plugin_id: None,
                     display_order: 0,
+                    enabled: true,
+                    is_managed: false,
                 }],
                 warnings: Vec::new(),
                 errors: Vec::new(),
@@ -284,3 +308,270 @@ timeout = 5
     );
     Ok(())
 }
+
+#[tokio::test]
+async fn config_batch_write_toggles_user_hook() -> Result<()> {
+    let codex_home = TempDir::new()?;
+    let cwd = TempDir::new()?;
+    write_user_hook_config(codex_home.path())?;
+
+    let mut mcp = McpProcess::new(codex_home.path()).await?;
+    timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
+
+    let request_id = mcp
+        .send_hooks_list_request(HooksListParams {
+            cwds: vec![cwd.path().to_path_buf()],
+        })
+        .await?;
+    let response: JSONRPCResponse = timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+    let HooksListResponse { data } = to_response(response)?;
+    let hook = &data[0].hooks[0];
+    assert_eq!(hook.enabled, true);
+
+    let write_id = mcp
+        .send_config_batch_write_request(ConfigBatchWriteParams {
+            edits: vec![ConfigEdit {
+                key_path: "hooks.state".to_string(),
+                value: serde_json::json!({
+                    hook.key.clone(): {
+                        "enabled": false
+                    }
+                }),
+                merge_strategy: MergeStrategy::Upsert,
+            }],
+            file_path: None,
+            expected_version: None,
+            reload_user_config: true,
+        })
+        .await?;
+    let response: JSONRPCResponse = timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(write_id)),
+    )
+    .await??;
+    let _: codex_app_server_protocol::ConfigWriteResponse = to_response(response)?;
+
+    let request_id = mcp
+        .send_hooks_list_request(HooksListParams {
+            cwds: vec![cwd.path().to_path_buf()],
+        })
+        .await?;
+    let response: JSONRPCResponse = timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+    let HooksListResponse { data } = to_response(response)?;
+    assert_eq!(data[0].hooks.len(), 1);
+    assert_eq!(data[0].hooks[0].key, hook.key);
+    assert_eq!(data[0].hooks[0].enabled, false);
+
+    let write_id = mcp
+        .send_config_batch_write_request(ConfigBatchWriteParams {
+            edits: vec![ConfigEdit {
+                key_path: "hooks.state".to_string(),
+                value: serde_json::json!({
+                    hook.key.clone(): {
+                        "enabled": true
+                    }
+                }),
+                merge_strategy: MergeStrategy::Upsert,
+            }],
+            file_path: None,
+            expected_version: None,
+            reload_user_config: true,
+        })
+        .await?;
+    let response: JSONRPCResponse = timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(write_id)),
+    )
+    .await??;
+    let _: codex_app_server_protocol::ConfigWriteResponse = to_response(response)?;
+
+    let request_id = mcp
+        .send_hooks_list_request(HooksListParams {
+            cwds: vec![cwd.path().to_path_buf()],
+        })
+        .await?;
+    let response: JSONRPCResponse = timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+    let HooksListResponse { data } = to_response(response)?;
+    assert_eq!(data[0].hooks[0].enabled, true);
+    Ok(())
+}
+
+#[tokio::test]
+async fn config_batch_write_disables_hook_for_loaded_session() -> Result<()> {
+    skip_if_windows!(Ok(()));
+
+    let responses = vec![
+        create_final_assistant_message_sse_response("Warmup")?,
+        create_final_assistant_message_sse_response("First turn")?,
+        create_final_assistant_message_sse_response("Second turn")?,
+    ];
+    let server = create_mock_responses_server_sequence_unchecked(responses).await;
+    let codex_home = TempDir::new()?;
+    let hook_script_path = codex_home.path().join("user_prompt_submit_hook.py");
+    let hook_log_path = codex_home.path().join("user_prompt_submit_hook_log.jsonl");
+    std::fs::write(
+        &hook_script_path,
+        format!(
+            r#"import json
+from pathlib import Path
+import sys
+
+payload = json.load(sys.stdin)
+with Path(r"{hook_log_path}").open("a", encoding="utf-8") as handle:
+    handle.write(json.dumps(payload) + "\n")
+"#,
+            hook_log_path = hook_log_path.display(),
+        ),
+    )?;
+    std::fs::write(
+        codex_home.path().join("config.toml"),
+        format!(
+            r#"
+model = "mock-model"
+approval_policy = "never"
+sandbox_mode = "read-only"
+
+model_provider = "mock_provider"
+
+[model_providers.mock_provider]
+name = "Mock provider for test"
+base_url = "{server_uri}/v1"
+wire_api = "responses"
+request_max_retries = 0
+stream_max_retries = 0
+
+[hooks]
+
+[[hooks.UserPromptSubmit]]
+
+[[hooks.UserPromptSubmit.hooks]]
+type = "command"
+command = "python3 {hook_script_path}"
+"#,
+            server_uri = server.uri(),
+            hook_script_path = hook_script_path.display(),
+        ),
+    )?;
+
+    let mut mcp = McpProcess::new(codex_home.path()).await?;
+    timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
+
+    let hook_list_id = mcp
+        .send_hooks_list_request(HooksListParams {
+            cwds: vec![codex_home.path().to_path_buf()],
+        })
+        .await?;
+    let response: JSONRPCResponse = timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(hook_list_id)),
+    )
+    .await??;
+    let HooksListResponse { data } = to_response(response)?;
+    let hook = &data[0].hooks[0];
+    assert_eq!(hook.enabled, true);
+
+    let thread_start_id = mcp
+        .send_thread_start_request(ThreadStartParams {
+            model: Some("mock-model".to_string()),
+            ..Default::default()
+        })
+        .await?;
+    let response: JSONRPCResponse = timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(thread_start_id)),
+    )
+    .await??;
+    let ThreadStartResponse { thread, .. } = to_response(response)?;
+
+    let first_turn_id = mcp
+        .send_turn_start_request(TurnStartParams {
+            thread_id: thread.id.clone(),
+            input: vec![V2UserInput::Text {
+                text: "first turn".to_string(),
+                text_elements: Vec::new(),
+            }],
+            ..Default::default()
+        })
+        .await?;
+    timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(first_turn_id)),
+    )
+    .await??;
+    timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_notification_message("turn/completed"),
+    )
+    .await??;
+    assert_eq!(
+        std::fs::read_to_string(&hook_log_path)?
+            .lines()
+            .filter(|line| !line.is_empty())
+            .count(),
+        1
+    );
+
+    let write_id = mcp
+        .send_config_batch_write_request(ConfigBatchWriteParams {
+            edits: vec![ConfigEdit {
+                key_path: "hooks.state".to_string(),
+                value: serde_json::json!({
+                    hook.key.clone(): {
+                        "enabled": false
+                    }
+                }),
+                merge_strategy: MergeStrategy::Upsert,
+            }],
+            file_path: None,
+            expected_version: None,
+            reload_user_config: true,
+        })
+        .await?;
+    let response: JSONRPCResponse = timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(write_id)),
+    )
+    .await??;
+    let _: codex_app_server_protocol::ConfigWriteResponse = to_response(response)?;
+
+    let second_turn_id = mcp
+        .send_turn_start_request(TurnStartParams {
+            thread_id: thread.id,
+            input: vec![V2UserInput::Text {
+                text: "second turn".to_string(),
+                text_elements: Vec::new(),
+            }],
+            ..Default::default()
+        })
+        .await?;
+    timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(second_turn_id)),
+    )
+    .await??;
+    timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_notification_message("turn/completed"),
+    )
+    .await??;
+    assert_eq!(
+        std::fs::read_to_string(&hook_log_path)?
+            .lines()
+            .filter(|line| !line.is_empty())
+            .count(),
+        1
+    );
+    Ok(())
+}