//! [`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::{Arc, Mutex}; use async_trait::async_trait; use serde_json::Value; use domain::ports::{AgentSession, AgentSessionError, ReplyEvent, ReplyStream}; use domain::sandbox::{SandboxEnforcer, SandboxPlan}; 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, /// Id de conversation Codex capté (= `thread_id`, pour la reprise). pub conversation_id: Option, } /// **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 ""` ; /// reprise : `codex exec resume --json --skip-git-repo-check ""`. /// /// Le flux est du **JSONL**. Types réels : /// /// - `{"type":"thread.started","thread_id":""}` ⇒ capte le `thread_id` /// (= id de conversation pour la reprise), **aucun** événement émis. /// - `{"type":"turn.started"}` ⇒ [`ReplyEvent::Heartbeat`] (vivacité non terminale). /// - `{"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":{…}}` ⇒ [`ReplyEvent::Heartbeat`] (le `Final` /// vient de l'`agent_message`, pas de `turn.completed`). /// /// 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 { 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); } // Début/fin de tour côté moteur : pas de contenu, mais preuve de vivacité ⇒ // battement de cœur non terminal (readiness/heartbeat lot 1). Le `Final` vient // toujours de l'`agent_message`, jamais de `turn.completed`. Some("turn.started") | Some("turn.completed") => events.push(ReplyEvent::Heartbeat), 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 => {} } } } // 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>, /// Plan de sandbox OS **par lancement** (lot LP4-4), porté dans chaque /// [`SpawnLine`]. `None` ⇒ aucun sandboxing (drain async natif). sandbox: Option, /// Enforcer OS **par instance** (lot LP4-4), passé à [`run_turn`]. `None` ⇒ pas /// de sandboxing même si un plan est présent (cohérent avec le chemin PTY). sandbox_enforcer: Option>, } impl CodexExecSession { /// Construit l'adapter. `command` est injecté (⇒ testable avec un fake CLI) ; /// `seed_conversation_id` amorce la reprise ou reste `None` (conversation neuve). /// `sandbox` / `sandbox_enforcer` (lot LP4-4) pilotent le sandboxing OS ; /// `None`/`None` ⇒ chemin natif inchangé. #[must_use] pub fn new( id: SessionId, command: impl Into, cwd: impl Into, seed_conversation_id: Option, sandbox: Option, sandbox_enforcer: Option>, ) -> Self { Self { id, command: command.into(), cwd: cwd.into(), conversation_id: Mutex::new(seed_conversation_id), sandbox, sandbox_enforcer, } } /// Compose la ligne de commande d'un tour. /// /// Format RÉEL vérifié 2026-06-10 (codex 0.137.0) : /// - Conversation neuve : `codex exec --json --skip-git-repo-check /// --sandbox workspace-write `. /// - Reprise (id connu) : `codex exec resume --json /// --skip-git-repo-check --sandbox workspace-write `. /// /// **Autonomie d'écriture (D3)** : `--sandbox workspace-write` autorise l'agent à /// écrire dans son workspace. `codex exec` est déjà non-interactif (aucun prompt /// d'approbation possible), donc on ne passe **pas** `--ask-for-approval` : ce flag /// appartient à la commande interactive `codex`, pas à la sous-commande `exec` qui /// sort sur `error: unexpected argument '--ask-for-approval' found`. Défaut /// raisonnable, aligné sur l'autonomie projet (CLAUDE.md §12) ; à terme **piloté par /// les permissions de l'agent** (`.ideai/permissions.json` + sandbox OS) — non /// implémenté ici. 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("--sandbox".to_owned()); args.push("workspace-write".to_owned()); args.push(prompt.to_owned()); SpawnLine { command: self.command.clone(), args, cwd: self.cwd.clone(), env: Vec::new(), stdin: None, sandbox: self.sandbox.clone(), } } } #[async_trait] impl AgentSession for CodexExecSession { fn id(&self) -> SessionId { self.id } fn conversation_id(&self) -> Option { self.conversation_id.lock().expect("mutex sain").clone() } async fn send(&self, prompt: &str) -> Result { let spec = self.build_spawn_line(prompt); let raw_lines = run_turn(&spec, None, self.sandbox_enforcer.as_ref()).await?; let mut events = Vec::new(); let mut captured_id = None; 'lines: for line in &raw_lines { let parsed = parse_event(line)?; if let Some(id) = parsed.conversation_id { captured_id = Some(id); } // Le `Final` est **terminal** (contrat de port) : on arrête d'émettre dès // qu'on l'a vu, pour qu'aucun heartbeat de fin (`turn.completed` postérieur) // ne le suive dans le flux. for event in parsed.events { let is_final = matches!(event, ReplyEvent::Final { .. }); events.push(event); if is_final { break 'lines; } } } 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(()) } }