hook trust metadata and enforcement (#20321)

# Why

We want shared hook trust that both the app and the TUI can build on,
but the metadata is only useful if runtime behavior agrees with it. This
PR adds a single backend trust model for hooks so unmanaged hooks cannot
run until the current definition has been reviewed, while managed hooks
remain runnable and non-configurable.

# What

- persist `trusted_hash` alongside hook state in `config.toml`
- expose `currentHash` and derived `trustStatus` through `hooks/list`
- derive trust from normalized hook definitions so equivalent hooks from
`config.toml` and `hooks.json` share the same trust identity
- gate unmanaged hooks on trust before they enter the runnable handler
set

# Reviewer Notes

- key file to review is `codex-rs/hooks/src/engine/discovery.rs`
- the only **core** change is schema related

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

@@ -11,6 +11,7 @@ use codex_app_server_protocol::HookEventName;
 use codex_app_server_protocol::HookHandlerType;
 use codex_app_server_protocol::HookMetadata;
 use codex_app_server_protocol::HookSource;
+use codex_app_server_protocol::HookTrustStatus;
 use codex_app_server_protocol::HooksListEntry;
 use codex_app_server_protocol::HooksListParams;
 use codex_app_server_protocol::HooksListResponse;
@@ -26,11 +27,44 @@ 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 serde::Serialize;
 use tempfile::TempDir;
 use tokio::time::timeout;
 
 const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
 
+#[derive(Serialize)]
+struct NormalizedHookIdentity {
+    event_name: &'static str,
+    #[serde(flatten)]
+    group: codex_config::MatcherGroup,
+}
+
+fn command_hook_hash(
+    event_name: &'static str,
+    matcher: Option<&str>,
+    command: &str,
+    timeout_sec: u64,
+    status_message: Option<&str>,
+) -> String {
+    let identity = NormalizedHookIdentity {
+        event_name,
+        group: codex_config::MatcherGroup {
+            matcher: matcher.map(ToOwned::to_owned),
+            hooks: vec![codex_config::HookHandlerConfig::Command {
+                command: command.to_string(),
+                timeout_sec: Some(timeout_sec),
+                r#async: false,
+                status_message: status_message.map(ToOwned::to_owned),
+            }],
+        },
+    };
+    let Ok(value) = codex_config::TomlValue::try_from(identity) else {
+        unreachable!("normalized hook identity should serialize to TOML");
+    };
+    codex_config::version_for_toml(&value)
+}
+
 fn write_user_hook_config(codex_home: &std::path::Path) -> Result<()> {
     std::fs::write(
         codex_home.join("config.toml"),
@@ -113,6 +147,14 @@ async fn hooks_list_shows_discovered_hook() -> Result<()> {
                 display_order: 0,
                 enabled: true,
                 is_managed: false,
+                current_hash: command_hook_hash(
+                    "pre_tool_use",
+                    Some("Bash"),
+                    "python3 /tmp/listed-hook.py",
+                    /*timeout_sec*/ 5,
+                    Some("running listed hook"),
+                ),
+                trust_status: HookTrustStatus::Untrusted,
             }],
             warnings: Vec::new(),
             errors: Vec::new(),
@@ -183,6 +225,14 @@ async fn hooks_list_shows_discovered_plugin_hook() -> Result<()> {
                 display_order: 0,
                 enabled: true,
                 is_managed: false,
+                current_hash: command_hook_hash(
+                    "pre_tool_use",
+                    Some("Bash"),
+                    "echo plugin hook",
+                    /*timeout_sec*/ 7,
+                    Some("running plugin hook"),
+                ),
+                trust_status: HookTrustStatus::Untrusted,
             }],
             warnings: Vec::new(),
             errors: Vec::new(),
@@ -300,6 +350,14 @@ timeout = 5
                     display_order: 0,
                     enabled: true,
                     is_managed: false,
+                    current_hash: command_hook_hash(
+                        "pre_tool_use",
+                        Some("Bash"),
+                        "echo project hook",
+                        /*timeout_sec*/ 5,
+                        /*status_message*/ None,
+                    ),
+                    trust_status: HookTrustStatus::Untrusted,
                 }],
                 warnings: Vec::new(),
                 errors: Vec::new(),
@@ -408,6 +466,254 @@ async fn config_batch_write_toggles_user_hook() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn config_batch_write_updates_hook_trust_for_loaded_session() -> Result<()> {
+    skip_if_windows!(Ok(()));
+
+    let responses = vec![
+        create_final_assistant_message_sse_response("Warmup")?,
+        create_final_assistant_message_sse_response("Untrusted turn")?,
+        create_final_assistant_message_sse_response("Trusted turn")?,
+        create_final_assistant_message_sse_response("Modified 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].clone();
+    assert_eq!(hook.trust_status, HookTrustStatus::Untrusted);
+
+    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!(!std::fs::exists(&hook_log_path)?);
+
+    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(): {
+                        "trusted_hash": hook.current_hash.clone()
+                    }
+                }),
+                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 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 trusted_hook = &data[0].hooks[0];
+    assert_eq!(trusted_hook.key, hook.key);
+    assert_eq!(trusted_hook.current_hash, hook.current_hash);
+    assert_eq!(trusted_hook.trust_status, HookTrustStatus::Trusted);
+
+    let second_turn_id = mcp
+        .send_turn_start_request(TurnStartParams {
+            thread_id: thread.id.clone(),
+            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
+    );
+
+    let write_id = mcp
+        .send_config_batch_write_request(ConfigBatchWriteParams {
+            edits: vec![ConfigEdit {
+                key_path: "hooks.UserPromptSubmit".to_string(),
+                value: serde_json::json!([{
+                    "hooks": [{
+                        "type": "command",
+                        "command": format!("python3 {}", hook_script_path.display()),
+                        "statusMessage": "modified hook",
+                    }],
+                }]),
+                merge_strategy: MergeStrategy::Replace,
+            }],
+            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 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 modified_hook = &data[0].hooks[0];
+    assert_eq!(modified_hook.key, hook.key);
+    assert_ne!(modified_hook.current_hash, hook.current_hash);
+    assert_eq!(modified_hook.trust_status, HookTrustStatus::Modified);
+
+    let third_turn_id = mcp
+        .send_turn_start_request(TurnStartParams {
+            thread_id: thread.id,
+            input: vec![V2UserInput::Text {
+                text: "third turn".to_string(),
+                text_elements: Vec::new(),
+            }],
+            ..Default::default()
+        })
+        .await?;
+    timeout(
+        DEFAULT_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(third_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(())
+}
+
 #[tokio::test]
 async fn config_batch_write_disables_hook_for_loaded_session() -> Result<()> {
     skip_if_windows!(Ok(()));
@@ -482,6 +788,29 @@ command = "python3 {hook_script_path}"
     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(): {
+                        "trusted_hash": hook.current_hash.clone()
+                    }
+                }),
+                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 thread_start_id = mcp
         .send_thread_start_request(ThreadStartParams {
             model: Some("mock-model".to_string()),