IdeA/crates/infrastructure/src/session/codex.rs

//! [`CodexExecSession`] — adapter structuré Codex (ARCHITECTURE §17.2, spike **S2**).
//!
//! Pilote `codex exec --json` en mode non-interactif et traduit sa sortie structurée
//! vers le contrat universel [`ReplyEvent`]. Le **spike S2 est résolu** : le format
//! réel est vérifié (2026-06-09) et ISOLÉ dans [`parse_event`].
//!
//! # Séparation parsing / machinerie (CRUCIAL — §17.2)
//!
//! Comme pour Claude, la machinerie de process vit dans [`super::process`] et ignore
//! le format. **Seule [`parse_event`] (et la composition de la commande) porte le
//! format Codex** ; la machinerie reste inchangée.

use std::sync::Mutex;

use async_trait::async_trait;
use serde_json::Value;

use domain::ports::{AgentSession, AgentSessionError, ReplyEvent, ReplyStream};
use domain::SessionId;

use super::process::{run_turn, SpawnLine};

/// Résultat du parsing d'une ligne Codex : **zéro ou plusieurs** événements et/ou un
/// id de conversation Codex capté. Miroir de `claude::ParsedLine` (vecteur d'events
/// pour homogénéité du drain ; en pratique Codex rend 0 ou 1 événement par ligne).
#[derive(Debug, Default, PartialEq, Eq)]
pub struct ParsedLine {
    /// Événements universels à émettre (dans l'ordre), vide pour une ligne de contrôle.
    pub events: Vec<ReplyEvent>,
    /// Id de conversation Codex capté (= `thread_id`, pour la reprise).
    pub conversation_id: Option<String>,
}

/// **Parse une ligne de la sortie structurée de `codex exec --json`** vers le contrat
/// universel.
///
/// # Format RÉEL vérifié 2026-06-09 (spike S2 résolu)
///
/// Commande : `codex exec --json --skip-git-repo-check "<prompt>"` ;
/// reprise : `codex exec resume <thread_id> --json --skip-git-repo-check "<prompt>"`.
///
/// Le flux est du **JSONL**. Types réels :
///
/// - `{"type":"thread.started","thread_id":"<id>"}` ⇒ capte le `thread_id`
///   (= id de conversation pour la reprise), **aucun** événement émis.
/// - `{"type":"turn.started"}` ⇒ ignoré.
/// - `{"type":"item.completed","item":{"id":"item_0","type":"agent_message","text":"…"}}`
///   ⇒ si `item.type=="agent_message"` ⇒ [`ReplyEvent::Final`] (`content` = `item.text`,
///   c'est la réponse) ; sinon (`reasoning`/`command`/autre) ⇒
///   [`ReplyEvent::ToolActivity`] (`label` = `item.type`).
/// - `{"type":"turn.completed","usage":{…}}` ⇒ ignoré (le `Final` vient de
///   l'`agent_message`).
///
/// Ligne vide ⇒ ignorée ; type inconnu ⇒ ignoré sans erreur ; JSON illisible ⇒
/// [`AgentSessionError::Decode`] (jamais de JSON brut propagé).
///
/// # Errors
/// [`AgentSessionError::Decode`] si la ligne n'est pas un JSON valide.
pub fn parse_event(line: &str) -> Result<ParsedLine, AgentSessionError> {
    let trimmed = line.trim();
    if trimmed.is_empty() {
        return Ok(ParsedLine::default());
    }
    let value: Value = serde_json::from_str(trimmed)
        .map_err(|e| AgentSessionError::Decode(format!("ligne JSON illisible: {e}")))?;

    let mut conversation_id = None;
    let mut events = Vec::new();

    match value.get("type").and_then(Value::as_str) {
        Some("thread.started") => {
            // Handshake : on ne capte que le thread_id (= id de conversation).
            conversation_id = value
                .get("thread_id")
                .and_then(Value::as_str)
                .map(str::to_owned);
        }
        Some("item.completed") => {
            if let Some(item) = value.get("item") {
                match item.get("type").and_then(Value::as_str) {
                    Some("agent_message") => {
                        let content = item
                            .get("text")
                            .and_then(Value::as_str)
                            .unwrap_or_default()
                            .to_owned();
                        events.push(ReplyEvent::Final { content });
                    }
                    // reasoning / command / tout autre item ⇒ activité (label = type).
                    Some(kind) => events.push(ReplyEvent::ToolActivity {
                        label: kind.to_owned(),
                    }),
                    None => {}
                }
            }
        }
        // turn.started / turn.completed / type inconnu : ignoré (robustesse).
        _ => {}
    }

    Ok(ParsedLine {
        events,
        conversation_id,
    })
}

/// Adapter de session structurée Codex.
///
/// Incarnation « un `codex exec` par tour » (§17.2 (b)) : chaque `send` relance
/// `codex exec --json` avec le prompt et, dès qu'un `thread_id` a été capté, la
/// sous-commande de reprise. L'incarnation « process persistant » resterait derrière
/// le **même** port sans toucher au parsing.
pub struct CodexExecSession {
    /// Id de session IdeA.
    id: SessionId,
    /// Binaire à lancer (`codex` en prod, fake CLI en test).
    command: String,
    /// Répertoire de travail (run dir isolé §14.1).
    cwd: String,
    /// Id de conversation **du moteur** Codex, capté au premier tour, `None` avant.
    conversation_id: Mutex<Option<String>>,
}

impl CodexExecSession {
    /// Construit l'adapter. `command` est injecté (⇒ testable avec un fake CLI) ;
    /// `seed_conversation_id` amorce la reprise ou reste `None` (conversation neuve).
    #[must_use]
    pub fn new(
        id: SessionId,
        command: impl Into<String>,
        cwd: impl Into<String>,
        seed_conversation_id: Option<String>,
    ) -> Self {
        Self {
            id,
            command: command.into(),
            cwd: cwd.into(),
            conversation_id: Mutex::new(seed_conversation_id),
        }
    }

    /// Compose la ligne de commande d'un tour.
    ///
    /// Format RÉEL vérifié 2026-06-09 :
    /// - Conversation neuve : `codex exec --json --skip-git-repo-check <prompt>`.
    /// - Reprise (id connu) : `codex exec resume <thread_id> --json
    ///   --skip-git-repo-check <prompt>`.
    ///
    /// NOTE intégration (D3, hors parsing) : pour qu'un agent puisse **écrire**, le
    /// lancement devra ajouter `--sandbox workspace-write` + une politique
    /// d'approbation non bloquante. Ce n'est pas le rôle de cette composition.
    fn build_spawn_line(&self, prompt: &str) -> SpawnLine {
        let mut args = vec!["exec".to_owned()];
        if let Some(id) = self.conversation_id.lock().expect("mutex sain").as_ref() {
            args.push("resume".to_owned());
            args.push(id.clone());
        }
        args.push("--json".to_owned());
        args.push("--skip-git-repo-check".to_owned());
        args.push(prompt.to_owned());
        SpawnLine {
            command: self.command.clone(),
            args,
            cwd: self.cwd.clone(),
            env: Vec::new(),
            stdin: None,
        }
    }
}

#[async_trait]
impl AgentSession for CodexExecSession {
    fn id(&self) -> SessionId {
        self.id
    }

    fn conversation_id(&self) -> Option<String> {
        self.conversation_id.lock().expect("mutex sain").clone()
    }

    async fn send(&self, prompt: &str) -> Result<ReplyStream, AgentSessionError> {
        let spec = self.build_spawn_line(prompt);
        let raw_lines = run_turn(&spec, None).await?;

        let mut events = Vec::new();
        let mut captured_id = None;
        for line in &raw_lines {
            let parsed = parse_event(line)?;
            if let Some(id) = parsed.conversation_id {
                captured_id = Some(id);
            }
            events.extend(parsed.events);
        }
        if let Some(id) = captured_id {
            *self.conversation_id.lock().expect("mutex sain") = Some(id);
        }

        Ok(Box::new(events.into_iter()))
    }

    async fn shutdown(&self) -> Result<(), AgentSessionError> {
        // « un run par tour » ⇒ pas de process long survivant : idempotent, no-op.
        Ok(())
    }
}