818 lines
31 KiB
Rust
818 lines
31 KiB
Rust
//! [`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, ReplyProgress, ReplyProgressKind,
|
|
ReplyProgressSource, ReplyProgressStage, ReplyStream,
|
|
};
|
|
use domain::sandbox::{SandboxEnforcer, SandboxPlan};
|
|
use domain::SessionId;
|
|
|
|
use super::process::{
|
|
is_missing_provider_session, resume_fallback_prompt, run_turn_cancellable, SpawnLine,
|
|
TurnCancel,
|
|
};
|
|
|
|
/// 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>"}` /
|
|
/// `{"type":"codex.thread.started","thread_id":"<id>"}` ⇒ capte le `thread_id`
|
|
/// (= id de conversation pour la reprise), **aucun** événement émis.
|
|
/// - `{"type":"turn.started"}` ⇒ [`ReplyEvent::Heartbeat`] (vivacité non terminale).
|
|
/// - `{"type":"conversation.item.added","item":{…}}` /
|
|
/// `{"type":"item.started","item":{…}}` ⇒ progression non terminale `Started`
|
|
/// pour les items outil/MCP dès que Codex annonce leur existence.
|
|
/// - `{"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) ; si `item.type=="error"` ⇒ [`ReplyEvent::Error`] avec message
|
|
/// extrait de `item.text`, `item.message`, `item.error.message`, `item.error`, puis
|
|
/// des champs top-level `stderr`/`message`/`error.message`/`error` ; sinon
|
|
/// (`reasoning`/`command`/autre) ⇒ progression canonique `Completed` (`label` =
|
|
/// `item.type`).
|
|
/// - `{"type":"conversation.item.done","item":{…}}` est traité comme `item.completed`
|
|
/// (Codex CLI ≥ 0.147).
|
|
/// - `{"type":"turn.completed","usage":{…}}` ⇒ [`ReplyEvent::Heartbeat`] (le `Final`
|
|
/// vient de l'`agent_message`, pas de `turn.completed`).
|
|
/// - `{"type":"turn.failed","error":{…}}` / `{"type":"error","message":…}` ⇒
|
|
/// [`ReplyEvent::Error`] avec le message fournisseur. Une erreur de namespace MCP
|
|
/// trop long reçoit en plus le code lisible `MCP_NAMESPACE_TOO_LONG`.
|
|
///
|
|
/// 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" | "codex.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") => {
|
|
events.push(ReplyEvent::Progress {
|
|
progress: provider_progress(
|
|
ReplyProgressKind::Turn,
|
|
ReplyProgressStage::Started,
|
|
"tour démarré",
|
|
"turn.started",
|
|
),
|
|
});
|
|
events.push(ReplyEvent::Heartbeat);
|
|
}
|
|
Some("turn.completed") => {
|
|
events.push(ReplyEvent::Progress {
|
|
progress: provider_progress(
|
|
ReplyProgressKind::Turn,
|
|
ReplyProgressStage::Completed,
|
|
"tour terminé côté provider",
|
|
"turn.completed",
|
|
),
|
|
});
|
|
events.push(ReplyEvent::Heartbeat);
|
|
}
|
|
Some("turn.failed" | "error") => {
|
|
events.push(ReplyEvent::Error {
|
|
message: codex_terminal_error_message(&value),
|
|
});
|
|
}
|
|
Some("item.started" | "conversation.item.added" | "response.output_item.added") => {
|
|
if let Some(item) = value.get("item") {
|
|
append_item_progress(
|
|
&mut events,
|
|
item,
|
|
ReplyProgressStage::Started,
|
|
value
|
|
.get("type")
|
|
.and_then(Value::as_str)
|
|
.unwrap_or("item.started"),
|
|
);
|
|
}
|
|
}
|
|
Some("item.updated" | "conversation.item.updated") => {
|
|
if let Some(item) = value.get("item") {
|
|
append_item_progress(
|
|
&mut events,
|
|
item,
|
|
ReplyProgressStage::Delta,
|
|
value
|
|
.get("type")
|
|
.and_then(Value::as_str)
|
|
.unwrap_or("item.updated"),
|
|
);
|
|
}
|
|
}
|
|
Some("item.completed" | "conversation.item.done" | "response.output_item.done") => {
|
|
if let Some(item) = value.get("item") {
|
|
match item_type(item).as_deref() {
|
|
Some("agent_message") => {
|
|
let content = item_text(item).unwrap_or_default();
|
|
events.push(ReplyEvent::Final { content });
|
|
}
|
|
Some("error") => {
|
|
events.push(ReplyEvent::Error {
|
|
message: codex_error_message(&value, item),
|
|
});
|
|
}
|
|
// reasoning / command / tout autre item ⇒ progression canonique.
|
|
Some(kind) => {
|
|
let label = item_label(item).unwrap_or_else(|| kind.to_owned());
|
|
events.push(ReplyEvent::Progress {
|
|
progress: item_progress(
|
|
item,
|
|
&label,
|
|
ReplyProgressStage::Completed,
|
|
value
|
|
.get("type")
|
|
.and_then(Value::as_str)
|
|
.unwrap_or("item.completed"),
|
|
),
|
|
});
|
|
}
|
|
None => {}
|
|
}
|
|
}
|
|
}
|
|
// type inconnu : ignoré (robustesse).
|
|
_ => {}
|
|
}
|
|
|
|
Ok(ParsedLine {
|
|
events,
|
|
conversation_id,
|
|
})
|
|
}
|
|
|
|
fn provider_progress(
|
|
kind: ReplyProgressKind,
|
|
stage: ReplyProgressStage,
|
|
label: impl Into<String>,
|
|
native_event: impl Into<String>,
|
|
) -> ReplyProgress {
|
|
ReplyProgress::new(ReplyProgressSource::ProviderNative, kind, stage, label)
|
|
.with_provider_event("codex", native_event)
|
|
}
|
|
|
|
fn item_progress(
|
|
item: &Value,
|
|
label: &str,
|
|
stage: ReplyProgressStage,
|
|
native_event: impl Into<String>,
|
|
) -> ReplyProgress {
|
|
let progress_id = item_progress_id(item);
|
|
let progress = if label.starts_with("idea_") {
|
|
ReplyProgress::new(
|
|
ReplyProgressSource::IdeaLocal,
|
|
ReplyProgressKind::Mcp,
|
|
stage,
|
|
label.to_owned(),
|
|
)
|
|
.with_provider_event("codex", native_event)
|
|
.with_tool_name(label.to_owned())
|
|
} else {
|
|
provider_progress(
|
|
ReplyProgressKind::Tool,
|
|
stage,
|
|
label.to_owned(),
|
|
native_event,
|
|
)
|
|
.with_tool_name(label.to_owned())
|
|
};
|
|
let progress = if let Some(progress_id) = progress_id {
|
|
progress.with_progress_id(progress_id)
|
|
} else {
|
|
progress
|
|
};
|
|
if let Some(preview) = item_preview(item) {
|
|
progress.with_text(preview)
|
|
} else {
|
|
progress
|
|
}
|
|
}
|
|
|
|
fn agent_message_progress(item: &Value, native_event: &str) -> Option<ReplyEvent> {
|
|
let content = item_text(item)?;
|
|
let mut progress = ReplyProgress::new(
|
|
ReplyProgressSource::ProviderNative,
|
|
ReplyProgressKind::Message,
|
|
ReplyProgressStage::Delta,
|
|
"message intermédiaire",
|
|
)
|
|
.with_provider_event("codex", native_event)
|
|
.with_text(content);
|
|
if let Some(progress_id) = item_progress_id(item) {
|
|
progress = progress.with_progress_id(progress_id);
|
|
}
|
|
Some(ReplyEvent::Progress { progress })
|
|
}
|
|
|
|
fn item_progress_id(item: &Value) -> Option<String> {
|
|
non_blank_string(item.get("id"))
|
|
.or_else(|| non_blank_string(item.get("item_id")))
|
|
.or_else(|| non_blank_string(item.get("call_id")))
|
|
.map(|id| format!("codex:item:{id}"))
|
|
}
|
|
|
|
fn append_item_progress(
|
|
events: &mut Vec<ReplyEvent>,
|
|
item: &Value,
|
|
stage: ReplyProgressStage,
|
|
native_event: &str,
|
|
) {
|
|
let Some(kind) = item_type(item) else {
|
|
return;
|
|
};
|
|
if matches!(kind.as_str(), "agent_message" | "error") {
|
|
return;
|
|
}
|
|
let label = item_label(item).unwrap_or(kind);
|
|
events.push(ReplyEvent::Progress {
|
|
progress: item_progress(item, &label, stage, native_event),
|
|
});
|
|
}
|
|
|
|
fn item_type(item: &Value) -> Option<String> {
|
|
non_blank_string(item.get("type"))
|
|
.or_else(|| non_blank_string(item.get("item_type")))
|
|
.or_else(|| non_blank_string(item.get("kind")))
|
|
}
|
|
|
|
fn item_label(item: &Value) -> Option<String> {
|
|
non_blank_string(item.get("name"))
|
|
.or_else(|| non_blank_string(item.get("tool_name")))
|
|
.or_else(|| non_blank_string(item.get("tool")))
|
|
.or_else(|| non_blank_string(item.get("action")))
|
|
.or_else(|| nested_non_blank(item, &["call", "name"]))
|
|
.or_else(|| nested_non_blank(item, &["function", "name"]))
|
|
.or_else(|| nested_non_blank(item, &["function_call", "name"]))
|
|
.or_else(|| item_type(item))
|
|
}
|
|
|
|
fn item_text(item: &Value) -> Option<String> {
|
|
non_blank_string(item.get("text"))
|
|
.or_else(|| non_blank_string(item.get("message")))
|
|
.or_else(|| nested_non_blank(item, &["content", "text"]))
|
|
}
|
|
|
|
fn item_preview(item: &Value) -> Option<String> {
|
|
non_blank_string(item.get("arguments"))
|
|
.or_else(|| non_blank_string(item.get("input")))
|
|
.or_else(|| non_blank_string(item.get("command")))
|
|
.or_else(|| non_blank_string(item.get("text")))
|
|
.or_else(|| non_blank_string(item.get("message")))
|
|
.or_else(|| nested_non_blank(item, &["function", "arguments"]))
|
|
.or_else(|| nested_non_blank(item, &["function_call", "arguments"]))
|
|
.or_else(|| nested_non_blank(item, &["call", "arguments"]))
|
|
.or_else(|| value_preview(item.get("arguments")))
|
|
.or_else(|| value_preview(item.get("input")))
|
|
}
|
|
|
|
fn value_preview(value: Option<&Value>) -> Option<String> {
|
|
value
|
|
.filter(|value| value.is_object() || value.is_array())
|
|
.map(Value::to_string)
|
|
.filter(|text| !text.trim().is_empty())
|
|
}
|
|
|
|
fn nested_non_blank(root: &Value, path: &[&str]) -> Option<String> {
|
|
let mut cursor = root;
|
|
for key in path {
|
|
cursor = cursor.get(*key)?;
|
|
}
|
|
non_blank_string(Some(cursor))
|
|
}
|
|
|
|
const CODEX_ERROR_FALLBACK: &str = "Codex a renvoyé une erreur.";
|
|
|
|
fn non_blank_string(value: Option<&Value>) -> Option<String> {
|
|
value
|
|
.and_then(Value::as_str)
|
|
.map(str::trim)
|
|
.filter(|s| !s.is_empty())
|
|
.map(str::to_owned)
|
|
}
|
|
|
|
fn error_value_message(value: Option<&Value>) -> Option<String> {
|
|
let value = value?;
|
|
non_blank_string(value.get("message")).or_else(|| non_blank_string(Some(value)))
|
|
}
|
|
|
|
fn codex_error_message(root: &Value, item: &Value) -> String {
|
|
let message = non_blank_string(item.get("text"))
|
|
.or_else(|| non_blank_string(item.get("message")))
|
|
.or_else(|| error_value_message(item.get("error")))
|
|
.or_else(|| non_blank_string(root.get("stderr")))
|
|
.or_else(|| non_blank_string(root.get("message")))
|
|
.or_else(|| error_value_message(root.get("error")))
|
|
.unwrap_or_else(|| CODEX_ERROR_FALLBACK.to_owned());
|
|
classify_codex_error(message)
|
|
}
|
|
|
|
fn codex_terminal_error_message(root: &Value) -> String {
|
|
let message = non_blank_string(root.get("message"))
|
|
.or_else(|| error_value_message(root.get("error")))
|
|
.or_else(|| non_blank_string(root.get("stderr")))
|
|
.unwrap_or_else(|| CODEX_ERROR_FALLBACK.to_owned());
|
|
classify_codex_error(message)
|
|
}
|
|
|
|
fn classify_codex_error(message: String) -> String {
|
|
let lowercase = message.to_ascii_lowercase();
|
|
if lowercase.contains("namespace")
|
|
&& (lowercase.contains("string_above_max_length")
|
|
|| lowercase.contains("maximum length 64")
|
|
|| lowercase.contains("max length 64"))
|
|
{
|
|
format!("MCP_NAMESPACE_TOO_LONG: {message}")
|
|
} else {
|
|
message
|
|
}
|
|
}
|
|
|
|
/// 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,
|
|
/// Codex CLI sandbox mode passed to `codex exec --sandbox`.
|
|
sandbox_mode: String,
|
|
/// Project/workspace roots that must be writable in Codex's CLI sandbox.
|
|
writable_roots: Vec<String>,
|
|
/// Structured policy projection of Codex workspace-write sandbox network access.
|
|
network_access: Option<bool>,
|
|
/// Profile-selected model forwarded as a Codex config override on every exec turn.
|
|
model: Option<String>,
|
|
/// Profile-selected reasoning effort forwarded as a Codex config override.
|
|
model_reasoning_effort: Option<String>,
|
|
/// Variables d'environnement préparées au lancement (ex. `CODEX_HOME` isolé).
|
|
env: Vec<(String, String)>,
|
|
/// Id de conversation **du moteur** Codex, capté au premier tour, `None` avant.
|
|
conversation_id: Mutex<Option<String>>,
|
|
/// Portable handoff injected only if the seeded provider session is stale.
|
|
resume_fallback_preamble: Option<String>,
|
|
/// Plan de sandbox OS **par lancement** (lot LP4-4), porté dans chaque
|
|
/// [`SpawnLine`]. `None` ⇒ aucun sandboxing (drain async natif).
|
|
sandbox: Option<SandboxPlan>,
|
|
/// 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<Arc<dyn SandboxEnforcer>>,
|
|
/// Cancel handle for the process backing the current `codex exec` turn.
|
|
turn_cancel: TurnCancel,
|
|
}
|
|
|
|
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<String>,
|
|
cwd: impl Into<String>,
|
|
seed_conversation_id: Option<String>,
|
|
writable_roots: Vec<String>,
|
|
env: Vec<(String, String)>,
|
|
sandbox: Option<SandboxPlan>,
|
|
sandbox_enforcer: Option<Arc<dyn SandboxEnforcer>>,
|
|
) -> Self {
|
|
Self::new_with_policy_and_overrides(
|
|
id,
|
|
command,
|
|
cwd,
|
|
seed_conversation_id,
|
|
"workspace-write",
|
|
writable_roots,
|
|
None,
|
|
None,
|
|
None,
|
|
env,
|
|
sandbox,
|
|
sandbox_enforcer,
|
|
)
|
|
}
|
|
|
|
/// Construit l'adapter avec la politique compatible `codex exec` résolue par
|
|
/// `LaunchAgent`.
|
|
#[must_use]
|
|
#[allow(clippy::too_many_arguments)]
|
|
pub fn new_with_policy(
|
|
id: SessionId,
|
|
command: impl Into<String>,
|
|
cwd: impl Into<String>,
|
|
seed_conversation_id: Option<String>,
|
|
sandbox_mode: impl Into<String>,
|
|
writable_roots: Vec<String>,
|
|
network_access: Option<bool>,
|
|
env: Vec<(String, String)>,
|
|
sandbox: Option<SandboxPlan>,
|
|
sandbox_enforcer: Option<Arc<dyn SandboxEnforcer>>,
|
|
) -> Self {
|
|
Self::new_with_policy_and_overrides(
|
|
id,
|
|
command,
|
|
cwd,
|
|
seed_conversation_id,
|
|
sandbox_mode,
|
|
writable_roots,
|
|
network_access,
|
|
None,
|
|
None,
|
|
env,
|
|
sandbox,
|
|
sandbox_enforcer,
|
|
)
|
|
}
|
|
|
|
/// Construit l'adapter avec politique + overrides de config issus du profil IdeA.
|
|
#[must_use]
|
|
#[allow(clippy::too_many_arguments)]
|
|
pub fn new_with_policy_and_overrides(
|
|
id: SessionId,
|
|
command: impl Into<String>,
|
|
cwd: impl Into<String>,
|
|
seed_conversation_id: Option<String>,
|
|
sandbox_mode: impl Into<String>,
|
|
writable_roots: Vec<String>,
|
|
network_access: Option<bool>,
|
|
model: Option<String>,
|
|
model_reasoning_effort: Option<String>,
|
|
env: Vec<(String, String)>,
|
|
sandbox: Option<SandboxPlan>,
|
|
sandbox_enforcer: Option<Arc<dyn SandboxEnforcer>>,
|
|
) -> Self {
|
|
Self {
|
|
id,
|
|
command: command.into(),
|
|
cwd: cwd.into(),
|
|
sandbox_mode: sandbox_mode.into(),
|
|
writable_roots,
|
|
network_access,
|
|
model,
|
|
model_reasoning_effort,
|
|
env,
|
|
conversation_id: Mutex::new(seed_conversation_id),
|
|
resume_fallback_preamble: None,
|
|
sandbox,
|
|
sandbox_enforcer,
|
|
turn_cancel: TurnCancel::new(),
|
|
}
|
|
}
|
|
|
|
/// Adds the portable context used by the one-shot stale-resume fallback.
|
|
#[must_use]
|
|
pub fn with_resume_fallback_preamble(mut self, preamble: Option<&str>) -> Self {
|
|
self.resume_fallback_preamble = preamble.map(str::to_owned);
|
|
self
|
|
}
|
|
|
|
/// 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 [--add-dir <explicit-root>...] <prompt>`.
|
|
/// - Reprise (id connu) : `codex exec --json
|
|
/// --skip-git-repo-check --sandbox workspace-write [--add-dir <explicit-root>...]
|
|
/// resume <thread_id> <prompt>`.
|
|
///
|
|
/// **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`. Comme l'agent
|
|
/// tourne depuis son run dir isolé, `--add-dir` expose uniquement les roots
|
|
/// writable explicitement fournis par la launch policy.
|
|
fn build_spawn_line(&self, prompt: &str) -> SpawnLine {
|
|
let mut args = vec!["exec".to_owned()];
|
|
let conversation_id = self.conversation_id.lock().expect("mutex sain").clone();
|
|
args.push("--json".to_owned());
|
|
args.push("--skip-git-repo-check".to_owned());
|
|
if !self.sandbox_mode.trim().is_empty() {
|
|
args.push("--sandbox".to_owned());
|
|
args.push(self.sandbox_mode.clone());
|
|
}
|
|
if self.sandbox_mode == "workspace-write" {
|
|
for root in self.writable_roots.iter().filter(|root| !root.is_empty()) {
|
|
args.push("--add-dir".to_owned());
|
|
args.push(root.clone());
|
|
}
|
|
}
|
|
if let Some(network_access) = self.network_access {
|
|
args.push("-c".to_owned());
|
|
args.push(format!(
|
|
"sandbox_workspace_write.network_access={network_access}"
|
|
));
|
|
}
|
|
if let Some(model) = self
|
|
.model
|
|
.as_deref()
|
|
.filter(|model| !model.trim().is_empty())
|
|
{
|
|
args.push("-c".to_owned());
|
|
args.push(format!("model={}", codex_toml_string(model)));
|
|
}
|
|
if let Some(effort) = self
|
|
.model_reasoning_effort
|
|
.as_deref()
|
|
.filter(|effort| !effort.trim().is_empty())
|
|
{
|
|
args.push("-c".to_owned());
|
|
args.push(format!(
|
|
"model_reasoning_effort={}",
|
|
codex_toml_string(effort)
|
|
));
|
|
}
|
|
if let Some(id) = conversation_id {
|
|
args.push("resume".to_owned());
|
|
args.push(id);
|
|
}
|
|
args.push(prompt.to_owned());
|
|
let mut env = self.env.clone();
|
|
if let Some(network_access) = self.network_access {
|
|
upsert_env(
|
|
&mut env,
|
|
"CODEX_SANDBOX_NETWORK_DISABLED",
|
|
if network_access { "0" } else { "1" },
|
|
);
|
|
}
|
|
SpawnLine {
|
|
command: self.command.clone(),
|
|
args,
|
|
cwd: self.cwd.clone(),
|
|
env,
|
|
stdin: None,
|
|
sandbox: self.sandbox.clone(),
|
|
}
|
|
}
|
|
}
|
|
|
|
fn codex_toml_string(value: &str) -> String {
|
|
serde_json::to_string(value).expect("string serialization cannot fail")
|
|
}
|
|
|
|
fn upsert_env(env: &mut Vec<(String, String)>, key: &str, value: &str) {
|
|
if let Some((_, existing)) = env.iter_mut().find(|(k, _)| k == key) {
|
|
*existing = value.to_owned();
|
|
} else {
|
|
env.push((key.to_owned(), value.to_owned()));
|
|
}
|
|
}
|
|
|
|
#[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> {
|
|
self.send_inner(prompt, None).await
|
|
}
|
|
|
|
async fn send_with_tap(
|
|
&self,
|
|
prompt: &str,
|
|
tap: std::sync::mpsc::Sender<ReplyEvent>,
|
|
) -> Result<ReplyStream, AgentSessionError> {
|
|
self.send_inner(prompt, Some(tap)).await
|
|
}
|
|
|
|
async fn shutdown(&self) -> Result<(), AgentSessionError> {
|
|
// « un run par tour » ⇒ pas de process long survivant : idempotent, no-op.
|
|
Ok(())
|
|
}
|
|
|
|
async fn cancel_current_turn(
|
|
&self,
|
|
) -> Result<domain::ports::CancelCurrentTurnOutcome, AgentSessionError> {
|
|
self.turn_cancel.cancel().await
|
|
}
|
|
}
|
|
|
|
impl CodexExecSession {
|
|
async fn send_inner(
|
|
&self,
|
|
prompt: &str,
|
|
tap: Option<std::sync::mpsc::Sender<ReplyEvent>>,
|
|
) -> Result<ReplyStream, AgentSessionError> {
|
|
let had_native_resume = self.conversation_id().is_some();
|
|
let spec = self.build_spawn_line(prompt);
|
|
let has_tap = tap.is_some();
|
|
let (line_tap, parser_thread) = tap.map_or((None, None), |event_tap| {
|
|
let (line_tx, line_rx) = std::sync::mpsc::channel::<String>();
|
|
let parser = std::thread::spawn(move || {
|
|
let mut pending_final: Option<String> = None;
|
|
for line in line_rx {
|
|
if let Ok(Some(progress)) = agent_message_progress_from_line(&line) {
|
|
let _ = event_tap.send(progress);
|
|
}
|
|
if let Ok(parsed) = parse_event(&line) {
|
|
for event in parsed.events {
|
|
match event {
|
|
ReplyEvent::Final { content } => {
|
|
let _ = pending_final.replace(content);
|
|
}
|
|
other => {
|
|
if event_supersedes_pending_final(&other) {
|
|
let _ = pending_final.take();
|
|
}
|
|
let _ = event_tap.send(other);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
});
|
|
(Some(line_tx), Some(parser))
|
|
});
|
|
let mut raw_result = run_turn_cancellable(
|
|
&spec,
|
|
None,
|
|
self.sandbox_enforcer.as_ref(),
|
|
line_tap.clone(),
|
|
&self.turn_cancel,
|
|
)
|
|
.await;
|
|
if had_native_resume && raw_result.as_ref().is_err_and(is_missing_provider_session) {
|
|
*self.conversation_id.lock().expect("mutex sain") = None;
|
|
let fallback_prompt =
|
|
resume_fallback_prompt(self.resume_fallback_preamble.as_deref(), prompt);
|
|
raw_result = run_turn_cancellable(
|
|
&self.build_spawn_line(&fallback_prompt),
|
|
None,
|
|
self.sandbox_enforcer.as_ref(),
|
|
line_tap,
|
|
&self.turn_cancel,
|
|
)
|
|
.await;
|
|
}
|
|
if let Some(parser) = parser_thread {
|
|
let _ = parser.join();
|
|
}
|
|
let raw_lines = raw_result?;
|
|
|
|
let mut events = Vec::new();
|
|
let mut captured_id = None;
|
|
// Fix canal (B2, bootstrap) : un tour `codex exec` peut émettre PLUSIEURS
|
|
// `agent_message` — un préambule (« je vais… ») puis la conclusion. On ne coupe
|
|
// donc plus au premier : on retient le **dernier** `agent_message` comme unique
|
|
// `Final` terminal (la conclusion du tour), et on dégrade les précédents en
|
|
// activité non terminale (readiness préservée). Ainsi le demandeur reçoit la
|
|
// conclusion, pas le préambule. Absence d'`agent_message` ⇒ aucun `Final`
|
|
// (no-reply inchangé).
|
|
let mut last_final: Option<String> = None;
|
|
for line in &raw_lines {
|
|
let parsed = parse_event(line)?;
|
|
if let Some(id) = parsed.conversation_id {
|
|
captured_id = Some(id);
|
|
}
|
|
for event in parsed.events {
|
|
match event {
|
|
ReplyEvent::Final { content } => {
|
|
// Un `agent_message` précédemment retenu est supersédé : il devient
|
|
// une annonce non terminale, on garde le plus récent en conclusion.
|
|
if last_final.is_some() {
|
|
if !has_tap {
|
|
events.push(ReplyEvent::Announcement {
|
|
text: last_final.take().expect("présent"),
|
|
});
|
|
} else {
|
|
let _ = last_final.take();
|
|
}
|
|
}
|
|
last_final = Some(content);
|
|
}
|
|
other => {
|
|
if last_final.is_some() && event_supersedes_pending_final(&other) {
|
|
if !has_tap {
|
|
events.push(ReplyEvent::Announcement {
|
|
text: last_final.take().expect("présent"),
|
|
});
|
|
} else {
|
|
let _ = last_final.take();
|
|
}
|
|
}
|
|
if !has_tap {
|
|
events.push(other);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
// Le dernier `agent_message` retenu est le seul `Final` terminal, émis en fin de
|
|
// flux (après les heartbeats/activités du tour).
|
|
if let Some(content) = last_final {
|
|
events.push(ReplyEvent::Final { content });
|
|
}
|
|
if let Some(id) = captured_id {
|
|
*self.conversation_id.lock().expect("mutex sain") = Some(id);
|
|
}
|
|
|
|
Ok(Box::new(events.into_iter()))
|
|
}
|
|
}
|
|
|
|
fn event_supersedes_pending_final(event: &ReplyEvent) -> bool {
|
|
match event {
|
|
ReplyEvent::Progress { progress } => {
|
|
progress.kind != ReplyProgressKind::Turn
|
|
&& matches!(
|
|
progress.stage,
|
|
ReplyProgressStage::Started | ReplyProgressStage::Delta
|
|
)
|
|
}
|
|
ReplyEvent::ToolActivity { .. } | ReplyEvent::Error { .. } => true,
|
|
ReplyEvent::Final { .. }
|
|
| ReplyEvent::TextDelta { .. }
|
|
| ReplyEvent::Announcement { .. }
|
|
| ReplyEvent::Heartbeat
|
|
| ReplyEvent::RateLimited { .. } => false,
|
|
}
|
|
}
|
|
|
|
fn agent_message_progress_from_line(line: &str) -> Result<Option<ReplyEvent>, AgentSessionError> {
|
|
let trimmed = line.trim();
|
|
if trimmed.is_empty() {
|
|
return Ok(None);
|
|
}
|
|
let value: Value = serde_json::from_str(trimmed)
|
|
.map_err(|e| AgentSessionError::Decode(format!("ligne JSON illisible: {e}")))?;
|
|
let Some("item.completed" | "conversation.item.done" | "response.output_item.done") =
|
|
value.get("type").and_then(Value::as_str)
|
|
else {
|
|
return Ok(None);
|
|
};
|
|
let Some(item) = value.get("item") else {
|
|
return Ok(None);
|
|
};
|
|
if item_type(item).as_deref() != Some("agent_message") {
|
|
return Ok(None);
|
|
}
|
|
Ok(agent_message_progress(
|
|
item,
|
|
value
|
|
.get("type")
|
|
.and_then(Value::as_str)
|
|
.unwrap_or("item.completed"),
|
|
))
|
|
}
|