codex/codex-rs/core/src/turn_metadata.rs

use std::collections::BTreeMap;
use std::future::Future;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::OnceLock;
use std::sync::RwLock;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;

use serde_json::Value;
use tokio::sync::watch;
use tokio_util::task::AbortOnDropHandle;

use crate::responses_metadata::AGENT_NAME_KEY;
use crate::responses_metadata::CodexResponsesMetadata;
use crate::responses_metadata::CodexResponsesRequestKind;
use crate::responses_metadata::MAX_EXTRA_METADATA_VALUE_BYTES;
use crate::responses_metadata::PARENT_TURN_ID_KEY;
use crate::responses_metadata::ROOT_TURN_ID_KEY;
use crate::responses_metadata::TurnMetadataWorkspace;
use crate::responses_metadata::filter_extra_metadata;
use crate::responses_metadata::subagent_header_value;
use crate::responses_metadata::subagent_metadata_kind;
use crate::sandbox_tags::SandboxTags;
use crate::sandbox_tags::record_policy_metadata;
use crate::session::step_settings::ResolvedStepSettings;
use codex_file_system::WindowsSandboxSelection;
use codex_git_utils::SanitizedGitUrl;
use codex_git_utils::get_git_remote_urls_assume_git_repo;
use codex_git_utils::get_has_changes_in_repo;
use codex_git_utils::get_head_commit_hash;
use codex_protocol::AgentPath;
use codex_protocol::ThreadId;
use codex_protocol::models::PermissionProfile;
use codex_protocol::openai_models::ModelInfo;
use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::ThreadSource;
use codex_protocol::turn_input::ResponsesApiClientMetadataUpdate;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_git_discovery::GitRootDiscovery;

// Memory requests await this optional metadata before sending their model request.
const MEMORY_GIT_METADATA_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(1);

const MODEL_KEY: &str = "model";
const CODEX_VERSION_KEY: &str = "codex_version";
const REASONING_EFFORT_KEY: &str = "reasoning_effort";
const USER_INPUT_REQUESTED_DURING_TURN_KEY: &str = "user_input_requested_during_turn";
const WORKSPACE_KIND_KEY: &str = "workspace_kind";

fn is_targeted_continuation(value: &str) -> bool {
    serde_json::from_str::<Value>(value).is_ok_and(|value| value.get("review_target").is_some())
}

/// Captured execution settings shared by Responses and MCP metadata.
pub(crate) struct ExecutionMetadata<'a> {
    pub(crate) model: &'a str,
    pub(crate) reasoning_effort: Option<ReasoningEffortConfig>,
    pub(crate) node_repl_disabled: bool,
    pub(crate) auto_review_enabled: bool,
    pub(crate) node_repl_auto_review_required: bool,
}

impl<'a> ExecutionMetadata<'a> {
    pub(crate) fn from_settings(settings: &'a ResolvedStepSettings) -> Self {
        Self {
            model: settings.model_info.slug.as_str(),
            reasoning_effort: settings.effective_reasoning_effort(),
            node_repl_disabled: settings.model_info.node_repl_disabled,
            auto_review_enabled: crate::guardian::routes_approval_policy_to_guardian(
                settings.approval_policy(),
                settings.approvals_reviewer(),
            ),
            node_repl_auto_review_required: settings.model_info.computer_use_review_required(),
        }
    }

    pub(crate) fn apply_to(&self, metadata: &mut CodexResponsesMetadata) {
        metadata.auto_review_enabled = Some(self.auto_review_enabled);
        metadata.node_repl_auto_review_required = Some(self.node_repl_auto_review_required);
        metadata.node_repl_disabled = Some(self.node_repl_disabled);
        metadata
            .extra
            .insert(MODEL_KEY.to_string(), self.model.to_string());
        match &self.reasoning_effort {
            Some(reasoning_effort) => {
                metadata.extra.insert(
                    REASONING_EFFORT_KEY.to_string(),
                    reasoning_effort.to_string(),
                );
            }
            None => {
                metadata.extra.remove(REASONING_EFFORT_KEY);
            }
        }
    }
}

#[derive(Clone, Debug, Default)]
struct WorkspaceGitMetadata {
    associated_remote_urls: Option<BTreeMap<String, SanitizedGitUrl>>,
    latest_git_commit_hash: Option<String>,
    has_changes: Option<bool>,
}

impl WorkspaceGitMetadata {
    fn is_empty(&self) -> bool {
        self.associated_remote_urls.is_none()
            && self.latest_git_commit_hash.is_none()
            && self.has_changes.is_none()
    }
}

impl From<WorkspaceGitMetadata> for TurnMetadataWorkspace {
    fn from(value: WorkspaceGitMetadata) -> Self {
        Self {
            associated_remote_urls: value.associated_remote_urls,
            latest_git_commit_hash: value.latest_git_commit_hash,
            has_changes: value.has_changes,
        }
    }
}

#[allow(clippy::too_many_arguments)]
pub async fn detached_memory_responses_metadata(
    thread_manager: &crate::ThreadManager,
    installation_id: String,
    session_id: String,
    thread_id: String,
    window_id: String,
    session_source: &SessionSource,
    cwd: &AbsolutePathBuf,
    permission_profile: &PermissionProfile,
    sandbox: Option<&str>,
) -> CodexResponsesMetadata {
    let turn_id = uuid::Uuid::now_v7().to_string();
    let mut metadata = CodexResponsesMetadata {
        turn_id: Some(turn_id.clone()),
        root_turn_id: Some(turn_id),
        request_kind: Some(CodexResponsesRequestKind::Memory),
        thread_source: Some(ThreadSource::MemoryConsolidation),
        turn_trigger: Some("memory_consolidation".to_owned()),
        subagent_header: subagent_header_value(session_source),
        sandbox: sandbox.map(ToString::to_string),
        workspaces: memory_workspaces(
            cwd,
            thread_manager.git_root_discovery().discover(cwd.clone()),
        )
        .await,
        ..CodexResponsesMetadata::new(installation_id, session_id, thread_id, window_id)
    };
    record_policy_metadata(permission_profile, cwd.as_path(), &mut metadata);
    metadata
}

#[derive(Debug)]
pub(crate) struct TurnMetadataState {
    cwd: AbsolutePathBuf,
    session_id: String,
    thread_id: String,
    agent_name: String,
    forked_from_thread_id: Option<ThreadId>,
    parent_thread_id: Option<ThreadId>,
    parent_turn_id: OnceLock<String>,
    initiating_agent_path: OnceLock<AgentPath>,
    root_turn_id: OnceLock<String>,
    subagent_header: Option<String>,
    subagent_kind: Option<String>,
    thread_source: Option<ThreadSource>,
    turn_trigger: OnceLock<String>,
    turn_id: String,
    // TODO(anp): Derive this cached tag from TurnEnvironment::sandbox_context
    // so metadata reflects the selected environment's backend.
    pub(crate) sandbox_tags: SandboxTags,
    auto_review_enabled: bool,
    node_repl_auto_review_required: bool,
    node_repl_disabled: bool,
    enriched_workspaces: RwLock<Option<BTreeMap<String, TurnMetadataWorkspace>>>,
    turn_started_at_unix_ms: RwLock<Option<i64>>,
    responses_api_metadata: RwLock<BTreeMap<String, String>>,
    responsesapi_client_metadata: RwLock<BTreeMap<String, String>>,
    user_input_requested_during_turn: AtomicBool,
    enrichment_task: Mutex<Option<AbortOnDropHandle<()>>>,
    git_enrichment_complete: watch::Sender<bool>,
}

impl codex_analytics::TurnAnalyticsMetadata for TurnMetadataState {
    fn root_turn_id(&self) -> Option<String> {
        TurnMetadataState::root_turn_id(self)
    }

    fn turn_trigger(&self) -> Option<String> {
        self.turn_trigger
            .get()
            .filter(|trigger| trigger.len() <= MAX_EXTRA_METADATA_VALUE_BYTES)
            .cloned()
    }

    fn codex_turn_source(&self) -> Option<String> {
        // Match Responses metadata precedence without copying oversized client values to analytics.
        let configured_metadata = self
            .responses_api_metadata
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let client_metadata = self
            .responsesapi_client_metadata
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        configured_metadata
            .get("source")
            .or_else(|| client_metadata.get("source"))
            .filter(|source| source.len() <= MAX_EXTRA_METADATA_VALUE_BYTES)
            .cloned()
    }
}

impl TurnMetadataState {
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn new(
        session_id: String,
        thread_id: String,
        forked_from_thread_id: Option<ThreadId>,
        parent_thread_id: Option<ThreadId>,
        session_source: &SessionSource,
        thread_source: Option<ThreadSource>,
        turn_id: String,
        cwd: AbsolutePathBuf,
        permission_profile: &PermissionProfile,
        windows_sandbox_selection: impl Into<WindowsSandboxSelection>,
        enforce_managed_network: bool,
        auto_review_enabled: bool,
        model_info: &ModelInfo,
    ) -> Self {
        let sandbox_tags = SandboxTags::new(
            permission_profile,
            cwd.as_path(),
            windows_sandbox_selection.into(),
            enforce_managed_network,
        );
        let agent_name = session_source
            .get_agent_path()
            .unwrap_or_else(AgentPath::root)
            .to_string();
        Self {
            cwd,
            session_id,
            thread_id,
            agent_name,
            forked_from_thread_id,
            parent_thread_id,
            parent_turn_id: OnceLock::new(),
            initiating_agent_path: OnceLock::new(),
            root_turn_id: OnceLock::new(),
            subagent_header: subagent_header_value(session_source),
            subagent_kind: subagent_metadata_kind(session_source),
            thread_source,
            turn_trigger: OnceLock::new(),
            turn_id,
            sandbox_tags,
            auto_review_enabled,
            node_repl_auto_review_required: model_info.computer_use_review_required(),
            node_repl_disabled: model_info.node_repl_disabled,
            enriched_workspaces: RwLock::new(None),
            turn_started_at_unix_ms: RwLock::new(None),
            responses_api_metadata: RwLock::new(BTreeMap::new()),
            responsesapi_client_metadata: RwLock::new(BTreeMap::new()),
            user_input_requested_during_turn: AtomicBool::new(false),
            enrichment_task: Mutex::new(None),
            git_enrichment_complete: watch::channel(/*init*/ true).0,
        }
    }

    pub(crate) fn current_meta_value_for_mcp_request(
        &self,
        execution: ExecutionMetadata<'_>,
    ) -> Option<serde_json::Value> {
        self.current_meta_value_for_tool_request(self.mcp_metadata_template(), execution)
    }

    pub(crate) fn current_meta_value_for_extension_request(
        &self,
        execution: ExecutionMetadata<'_>,
    ) -> Option<serde_json::Value> {
        // Native extensions need configured Responses metadata for request attribution,
        // which the MCP template omits for external servers.
        self.current_meta_value_for_tool_request(self.responses_metadata_template(), execution)
    }

    fn current_meta_value_for_tool_request(
        &self,
        mut responses_metadata: CodexResponsesMetadata,
        execution: ExecutionMetadata<'_>,
    ) -> Option<serde_json::Value> {
        if responses_metadata
            .extra
            .get("misalignment_override")
            .is_some_and(|value| is_targeted_continuation(value))
        {
            // A review target is for Responses admission, not external tools or hooks.
            responses_metadata.extra.remove("misalignment_override");
        }
        execution.apply_to(&mut responses_metadata);
        // Never serialize harness-owned tool inventory for tool requests.
        responses_metadata.tool_namespaces_info = None;
        let Value::Object(mut metadata) = responses_metadata.turn_metadata_value()? else {
            return None;
        };
        metadata.remove(AGENT_NAME_KEY);
        metadata.remove(PARENT_TURN_ID_KEY);
        metadata.remove(ROOT_TURN_ID_KEY);
        metadata.insert(
            CODEX_VERSION_KEY.to_string(),
            Value::String(env!("CARGO_PKG_VERSION").to_string()),
        );
        if self
            .user_input_requested_during_turn
            .load(Ordering::Relaxed)
        {
            metadata.insert(
                USER_INPUT_REQUESTED_DURING_TURN_KEY.to_string(),
                Value::Bool(true),
            );
        } else {
            metadata.remove(USER_INPUT_REQUESTED_DURING_TURN_KEY);
        }
        Some(Value::Object(metadata))
    }

    pub(crate) fn to_responses_metadata(
        &self,
        installation_id: String,
        window_id: String,
        request_kind: CodexResponsesRequestKind,
    ) -> CodexResponsesMetadata {
        let mut metadata = self.responses_metadata_template();
        if matches!(request_kind, CodexResponsesRequestKind::Compaction(_))
            && metadata
                .extra
                .get("misalignment_override")
                .is_some_and(|value| is_targeted_continuation(value))
        {
            // Target-based intent belongs to the pending turn, not compaction. Preserve the
            // legacy timestamp path until servers can admit pre-turn compaction separately.
            metadata.extra.remove("misalignment_override");
        }
        CodexResponsesMetadata {
            installation_id,
            window_id,
            request_kind: Some(request_kind),
            ..metadata
        }
    }

    pub(crate) fn mark_user_input_requested_during_turn(&self) {
        self.user_input_requested_during_turn
            .store(true, Ordering::Relaxed);
    }

    pub(crate) fn set_parent_turn_id(&self, parent_turn_id: String) {
        if parent_turn_id.trim().is_empty() {
            return;
        }
        let _ = self.parent_turn_id.set(parent_turn_id);
    }

    pub(crate) fn parent_turn_id(&self) -> Option<String> {
        self.parent_turn_id.get().cloned()
    }

    pub(crate) fn set_initiating_agent_path(&self, initiating_agent_path: AgentPath) {
        let _ = self.initiating_agent_path.set(initiating_agent_path);
    }

    pub(crate) fn initiating_agent_path(&self) -> Option<&AgentPath> {
        self.initiating_agent_path.get()
    }

    pub(crate) fn set_root_turn_id(&self, root_turn_id: String) {
        if root_turn_id.trim().is_empty() {
            return;
        }
        let _ = self.root_turn_id.set(root_turn_id);
    }

    pub(crate) fn set_turn_trigger(&self, turn_trigger: String) {
        if turn_trigger.trim().is_empty() {
            return;
        }
        let _ = self.turn_trigger.set(turn_trigger);
    }

    pub(crate) fn current_turn_trigger(&self) -> Option<String> {
        self.turn_trigger.get().cloned()
    }

    pub(crate) fn root_turn_id(&self) -> Option<String> {
        self.root_turn_id.get().cloned()
    }

    pub(crate) fn update_responsesapi_client_metadata(
        &self,
        update: ResponsesApiClientMetadataUpdate,
    ) {
        let replacement = update.replacement.map(filter_extra_metadata);
        let patch = update.patch.map(filter_extra_metadata).unwrap_or_default();
        if replacement.is_none() && patch.is_empty() {
            return;
        }
        let mut current = self
            .responsesapi_client_metadata
            .write()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let continuation = current
            .get("misalignment_override")
            .filter(|value| is_targeted_continuation(value))
            .cloned();
        if let Some(replacement) = replacement {
            *current = replacement;
        }
        current.extend(patch);
        if let Some(continuation) = continuation {
            // Steering may update other metadata, but cannot replace the reviewed block.
            current.insert("misalignment_override".to_string(), continuation);
        }
    }

    pub(crate) fn set_responses_api_metadata(
        &self,
        responses_api_metadata: BTreeMap<String, String>,
    ) {
        let mut metadata = filter_extra_metadata(responses_api_metadata);
        // Only explicit per-turn metadata can acknowledge a reviewed block.
        metadata.remove("misalignment_override");
        *self
            .responses_api_metadata
            .write()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = metadata;
    }

    pub(crate) fn workspace_kind(&self) -> Option<String> {
        self.responsesapi_client_metadata
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .get(WORKSPACE_KIND_KEY)
            .cloned()
    }

    fn responses_metadata_template(&self) -> CodexResponsesMetadata {
        let mut metadata = self.mcp_metadata_template();
        if metadata.parent_thread_id.is_some() {
            metadata.forked_from_thread_id = None;
        }
        metadata.extra.extend(
            self.responses_api_metadata
                .read()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .clone(),
        );
        metadata
    }

    fn mcp_metadata_template(&self) -> CodexResponsesMetadata {
        let mut extra = self
            .responsesapi_client_metadata
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone();
        // Extract Guardian's internal parent before filtering configured metadata keys.
        // Ordinary app-server client metadata stays in `extra`.
        let parent_response_id =
            if self.subagent_header.as_deref() == Some(crate::guardian::GUARDIAN_REVIEWER_NAME) {
                extra.remove("parent_response_id")
            } else {
                None
            };
        for key in self
            .responses_api_metadata
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .keys()
        {
            extra.remove(key);
        }
        let mut metadata = CodexResponsesMetadata {
            parent_response_id,
            turn_id: Some(self.turn_id.clone()),
            agent_name: Some(self.agent_name.clone()),
            forked_from_thread_id: self.forked_from_thread_id,
            parent_thread_id: self.parent_thread_id,
            parent_turn_id: self.parent_turn_id.get().cloned(),
            root_turn_id: self.root_turn_id(),
            subagent_header: self.subagent_header.clone(),
            subagent_kind: self.subagent_kind.clone(),
            thread_source: self.thread_source.clone(),
            turn_trigger: self.turn_trigger.get().cloned(),
            auto_review_enabled: Some(self.auto_review_enabled),
            node_repl_auto_review_required: Some(self.node_repl_auto_review_required),
            node_repl_disabled: Some(self.node_repl_disabled),
            workspaces: self.current_workspaces(),
            tool_namespaces_info: None,
            turn_started_at_unix_ms: self.current_turn_started_at_unix_ms(),
            extra,
            ..CodexResponsesMetadata::new(
                String::new(),
                self.session_id.clone(),
                self.thread_id.clone(),
                String::new(),
            )
        };
        self.sandbox_tags.record_metadata(&mut metadata);
        metadata
    }

    fn current_workspaces(&self) -> BTreeMap<String, TurnMetadataWorkspace> {
        self.enriched_workspaces
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone()
            .unwrap_or_default()
    }

    fn current_turn_started_at_unix_ms(&self) -> Option<i64> {
        *self
            .turn_started_at_unix_ms
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    pub(crate) fn set_turn_started_at_unix_ms(&self, turn_started_at_unix_ms: i64) {
        *self
            .turn_started_at_unix_ms
            .write()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(turn_started_at_unix_ms);
    }

    pub(crate) fn spawn_git_enrichment_task(
        self: &Arc<Self>,
        git_root_discovery: Arc<GitRootDiscovery>,
    ) {
        let mut task_guard = self
            .enrichment_task
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if task_guard.is_some() {
            return;
        }

        self.git_enrichment_complete.send_replace(/*value*/ false);
        // History-only contexts have no active task to cancel enrichment. Keep the
        // state weak across awaits so dropping its last owner aborts this waiter.
        let state = Arc::downgrade(self);
        let cwd = self.cwd.clone();
        *task_guard = Some(AbortOnDropHandle::new(tokio::spawn(async move {
            let workspaces = git_workspaces(&cwd, git_root_discovery.discover(cwd.clone())).await;
            let Some(state) = state.upgrade() else {
                return;
            };
            if !workspaces.is_empty() {
                *state
                    .enriched_workspaces
                    .write()
                    .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(workspaces);
            }

            state.git_enrichment_complete.send_replace(/*value*/ true);
        })));
    }

    pub(crate) async fn wait_for_git_enrichment(&self) {
        let mut completion = self.git_enrichment_complete.subscribe();
        let _ = completion.wait_for(|complete| *complete).await;
    }

    pub(crate) fn cancel_git_enrichment_task(&self) {
        let mut task_guard = self
            .enrichment_task
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if let Some(task) = task_guard.take() {
            task.abort();
            self.git_enrichment_complete.send_replace(/*value*/ true);
        }
    }
}

async fn memory_workspaces(
    cwd: &AbsolutePathBuf,
    repo_root: impl Future<Output = Option<PathBuf>>,
) -> BTreeMap<String, TurnMetadataWorkspace> {
    tokio::time::timeout(MEMORY_GIT_METADATA_TIMEOUT, git_workspaces(cwd, repo_root))
        .await
        .unwrap_or_default()
}

async fn git_workspaces(
    cwd: &AbsolutePathBuf,
    repo_root: impl Future<Output = Option<PathBuf>>,
) -> BTreeMap<String, TurnMetadataWorkspace> {
    let Some(repo_root) = repo_root.await else {
        return BTreeMap::new();
    };
    let (head_commit_hash, associated_remote_urls, has_changes) = tokio::join!(
        get_head_commit_hash(cwd),
        get_git_remote_urls_assume_git_repo(cwd),
        get_has_changes_in_repo(cwd, &repo_root),
    );
    let workspace_git_metadata = WorkspaceGitMetadata {
        associated_remote_urls,
        latest_git_commit_hash: head_commit_hash.map(|sha| sha.0),
        has_changes,
    };
    let mut workspaces = BTreeMap::new();
    if !workspace_git_metadata.is_empty() {
        workspaces.insert(
            repo_root.to_string_lossy().into_owned(),
            workspace_git_metadata.into(),
        );
    }
    workspaces
}

#[cfg(test)]
#[path = "turn_metadata_tests.rs"]
mod tests;