diff --git a/crates/app-tauri/src/chat.rs b/crates/app-tauri/src/chat.rs index b272593..dfcdecd 100644 --- a/crates/app-tauri/src/chat.rs +++ b/crates/app-tauri/src/chat.rs @@ -175,11 +175,17 @@ impl ChatBridge { /// Maps a domain [`ReplyEvent`] to its wire [`ReplyChunk`]. Pure translation, no /// I/O — the single point where the typed turn event becomes a serialisable chunk /// (kept here so the pump and any test share one mapping). +/// +/// Returns `None` for [`ReplyEvent::Heartbeat`]: a heartbeat is a non-terminal +/// liveness proof (readiness/heartbeat lot 1) with **no chat content**, so it maps +/// to no wire chunk — the pump simply skips it. Every content-bearing event still +/// maps to exactly one chunk. #[must_use] -pub fn chunk_from_event(event: ReplyEvent) -> ReplyChunk { +pub fn chunk_from_event(event: ReplyEvent) -> Option { match event { - ReplyEvent::TextDelta { text } => ReplyChunk::TextDelta { text }, - ReplyEvent::ToolActivity { label } => ReplyChunk::ToolActivity { label }, - ReplyEvent::Final { content } => ReplyChunk::Final { content }, + ReplyEvent::TextDelta { text } => Some(ReplyChunk::TextDelta { text }), + ReplyEvent::ToolActivity { label } => Some(ReplyChunk::ToolActivity { label }), + ReplyEvent::Final { content } => Some(ReplyChunk::Final { content }), + ReplyEvent::Heartbeat => None, } } diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index df31957..609e9e3 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -1196,7 +1196,11 @@ pub async fn agent_send( let bridge = std::sync::Arc::clone(&state.chat_bridge); std::thread::spawn(move || { for event in stream { - let chunk = crate::chat::chunk_from_event(event); + // Heartbeats carry no chat content (readiness/heartbeat lot 1) ⇒ no wire + // chunk; skip them while still draining so the turn runs to its `Final`. + let Some(chunk) = crate::chat::chunk_from_event(event) else { + continue; + }; // `send_output` always records into the conversation scrollback; the // boolean only reflects live delivery. If the view navigated away // (no channel at this generation) we keep draining so the turn still diff --git a/crates/app-tauri/src/events.rs b/crates/app-tauri/src/events.rs index b937b6f..26c512d 100644 --- a/crates/app-tauri/src/events.rs +++ b/crates/app-tauri/src/events.rs @@ -13,6 +13,7 @@ use serde::Serialize; use tauri::{AppHandle, Emitter}; use domain::events::{DomainEvent, OrchestrationSource}; +use domain::input::AgentLiveness; use infrastructure::TokioBroadcastEventBus; /// Name of the Tauri event carrying relayed [`DomainEvent`]s. @@ -84,6 +85,17 @@ pub enum DomainEventDto { /// Reply length in bytes (metric, not the payload). reply_len: usize, }, + /// An agent's liveness (alive/stalled) changed (lot 2, readiness/heartbeat). The + /// frontend can badge a frozen agent; `"stalled"` while no proof of liveness arrived + /// for longer than the profile's `stallAfterMs`, back to `"alive"` on a late + /// battement or when the turn ends. + #[serde(rename_all = "camelCase")] + AgentLivenessChanged { + /// Agent id (UUID string). + agent_id: String, + /// New liveness, as a lowercase string (`"alive"` / `"stalled"`). + liveness: AgentLivenessDto, + }, /// An agent's runtime profile was changed (hot-swap of the AI engine). #[serde(rename_all = "camelCase")] AgentProfileChanged { @@ -215,6 +227,26 @@ pub enum OrchestrationSourceDto { Mcp, } +/// Wire mirror of [`AgentLiveness`]: the alive/stalled liveness of an agent (lot 2), +/// serialised as a lowercase string (`"alive"` / `"stalled"`) the frontend badges. +#[derive(Debug, Clone, Copy, Serialize)] +#[serde(rename_all = "camelCase")] +pub enum AgentLivenessDto { + /// The agent shows proof of liveness (or is idle). + Alive, + /// No proof of liveness past the profile's `stallAfterMs` threshold. + Stalled, +} + +impl From for AgentLivenessDto { + fn from(liveness: AgentLiveness) -> Self { + match liveness { + AgentLiveness::Alive => Self::Alive, + AgentLiveness::Stalled => Self::Stalled, + } + } +} + impl From for OrchestrationSourceDto { fn from(source: OrchestrationSource) -> Self { match source { @@ -265,6 +297,12 @@ impl From<&DomainEvent> for DomainEventDto { agent_id: agent_id.to_string(), reply_len: *reply_len, }, + DomainEvent::AgentLivenessChanged { agent_id, liveness } => { + Self::AgentLivenessChanged { + agent_id: agent_id.to_string(), + liveness: (*liveness).into(), + } + } DomainEvent::AgentProfileChanged { agent_id, profile_id, @@ -376,3 +414,40 @@ pub fn spawn_relay(app: AppHandle, bus: &TokioBroadcastEventBus) { } }); } + +#[cfg(test)] +mod tests { + use super::*; + use domain::ids::AgentId; + + fn agent(n: u128) -> AgentId { + AgentId::from_uuid(uuid::Uuid::from_u128(n)) + } + + /// Lot 2 : un `AgentLivenessChanged{Stalled}` du domaine se relaie en DTO + /// `Stalled` portant le même agent, et se sérialise en `"stalled"` (le mot que + /// le front badge). Garantit le câblage présentation de la détection de stall. + #[test] + fn liveness_changed_stalled_relays_to_dto_and_wire() { + let dto = DomainEventDto::from(&DomainEvent::AgentLivenessChanged { + agent_id: agent(7), + liveness: AgentLiveness::Stalled, + }); + let json = serde_json::to_value(&dto).expect("serialisable"); + assert_eq!(json["type"], "agentLivenessChanged"); + assert_eq!(json["agentId"], agent(7).to_string()); + assert_eq!(json["liveness"], "stalled"); + } + + /// La reprise `Stalled→Alive` se relaie en DTO `Alive` ⇒ wire `"alive"`. + #[test] + fn liveness_changed_alive_relays_to_dto_and_wire() { + let dto = DomainEventDto::from(&DomainEvent::AgentLivenessChanged { + agent_id: agent(3), + liveness: AgentLiveness::Alive, + }); + let json = serde_json::to_value(&dto).expect("serialisable"); + assert_eq!(json["type"], "agentLivenessChanged"); + assert_eq!(json["liveness"], "alive"); + } +} diff --git a/crates/app-tauri/src/state.rs b/crates/app-tauri/src/state.rs index 1e0f64e..6b38e5d 100644 --- a/crates/app-tauri/src/state.rs +++ b/crates/app-tauri/src/state.rs @@ -827,7 +827,7 @@ impl AppState { // partagé pour `resolve`/`resolve_ticket`/`cancel_head` côté orchestrateur. let inmemory_mailbox = Arc::new(InMemoryMailbox::new()); let mailbox = Arc::clone(&inmemory_mailbox) as Arc; - let input_mediator = Arc::new( + let mediated_inbox = Arc::new( MediatedInbox::with_pty( Arc::clone(&inmemory_mailbox), Arc::new(SystemMillisClock), @@ -836,7 +836,28 @@ impl AppState { // Émet `AgentBusyChanged` à la source (Busy à l'enqueue qui démarre un // tour, Idle au mark_idle) ⇒ relayé au front en event Tauri (cadrage C4). .with_events(Arc::clone(&events_port)), - ) as Arc; + ); + // Détection de stagnation (lot 2, readiness/heartbeat) : une tâche périodique + // détenue au composition root appelle `sweep_stalled` (logique de décision pure, + // `now` fourni par l'horloge injectée). Bascule `Alive→Stalled` les agents `Busy` + // sans battement depuis `stall_after_ms` (profil) et émet `AgentLivenessChanged` + // une fois par transition. Tick d'1 s : largement assez fin pour des seuils en + // dizaines de secondes, négligeable en charge. Détaché ⇒ vit autant que l'app. + { + let sweeper = Arc::clone(&mediated_inbox); + // `build` runs in Tauri's `setup` hook (main thread, *no* ambient Tokio + // runtime) — `tokio::spawn` would panic with "there is no reactor running". + // Use Tauri's global async runtime, like `events::spawn_relay` does. + tauri::async_runtime::spawn(async move { + let mut tick = tokio::time::interval(std::time::Duration::from_secs(1)); + loop { + tick.tick().await; + sweeper.sweep_stalled(); + } + }); + } + let input_mediator = + Arc::clone(&mediated_inbox) as Arc; // Registre des conversations par paire (cadrage C3) : un fil par paire, session // vivante keyée par conversation (lève l'ambiguïté session/agent). let conversation_registry = Arc::new(InMemoryConversationRegistry::new()) @@ -870,7 +891,12 @@ impl AppState { .with_record_turn( Arc::new(AppRecordTurnProvider) as Arc, Arc::clone(&clock) as Arc, - ), + ) + // Registre des sessions IA structurées (§17.5) : permet à `ask_agent` de + // router une cible structurée (sans PTY) sur sa session et de la drainer via + // `drain_with_readiness` — le `Final` réveille le round-trip ET marque l'agent + // `Idle` (readiness/heartbeat lot 1). + .with_structured(Arc::clone(&structured_sessions)), ); // --- Windows (L10) --- @@ -1119,10 +1145,13 @@ impl AppState { let Some(profile) = profile_by_id.get(&agent.profile_id) else { continue; }; - if !is_claude_mcp_profile(profile) { - continue; + // Dispatch par profil : Claude (`.mcp.json` + settings) ou Codex + // (`config.toml` isolé via `CODEX_HOME`). Tout autre profil ⇒ rien à migrer. + if is_claude_mcp_profile(profile) { + let _ = migrate_claude_run_dir(project, &agent.id, profile).await; + } else if is_codex_mcp_profile(profile) { + let _ = migrate_codex_run_dir(project, &agent.id, profile).await; } - let _ = migrate_claude_run_dir(project, &agent.id, profile).await; } } } @@ -1229,6 +1258,115 @@ fn claude_mcp_config_target(profile: &AgentProfile) -> Option<&str> { } } +/// Pendant Codex de [`migrate_claude_run_dir`] : répare le `config.toml` MCP isolé +/// du run dir (table `[mcp_servers.idea]`) pour que Codex, qui lit ses serveurs MCP +/// dans `$CODEX_HOME/config.toml`, voie les outils `idea_*` après un redémarrage. +/// `CODEX_HOME` est isolé au run dir au lancement (jamais le `~/.codex` global), donc +/// ce fichier appartient entièrement à IdeA : il est régénéré (clobber) dès qu'un +/// runtime réel est disponible. Best-effort, idempotent. +async fn migrate_codex_run_dir( + project: &Project, + agent_id: &AgentId, + profile: &AgentProfile, +) -> Result<(), std::io::Error> { + let run_dir = project + .root + .as_str() + .trim_end_matches(['/', '\\']) + .to_owned() + + &format!("/.ideai/run/{agent_id}"); + let run_dir_path = Path::new(&run_dir); + if tokio::fs::metadata(run_dir_path).await.is_err() { + return Ok(()); + } + + let runtime = crate::mcp_endpoint::idea_exe_path().map(|exe| McpRuntime { + exe, + endpoint: crate::mcp_endpoint::mcp_endpoint(&project.id) + .as_cli_arg() + .to_owned(), + project_id: project.id.as_uuid().simple().to_string(), + requester: agent_id.to_string(), + }); + migrate_codex_mcp_config(run_dir_path, profile, runtime.as_ref()).await +} + +async fn migrate_codex_mcp_config( + run_dir: &Path, + profile: &AgentProfile, + runtime: Option<&McpRuntime>, +) -> Result<(), std::io::Error> { + let Some((target, _home_env)) = codex_mcp_config_target(profile) else { + return Ok(()); + }; + // Sans runtime réel, seule une déclaration minimale est disponible : on ne + // régénère pas (mirror de `migrate_claude_mcp_config`). + let Some(runtime) = runtime else { + return Ok(()); + }; + let toml_path = run_dir.join(target); + let desired = mcp_server_entry_toml(profile, Some(runtime)); + + // `config.toml` isolé = entièrement géré par IdeA ⇒ régénération (clobber) ; + // idempotent (no-op si le contenu est déjà à jour). + match tokio::fs::read_to_string(&toml_path).await { + Ok(existing) if existing == desired => return Ok(()), + Ok(_) => {} + Err(err) if err.kind() == std::io::ErrorKind::NotFound => {} + Err(err) => return Err(err), + } + if let Some(parent) = toml_path.parent() { + tokio::fs::create_dir_all(parent).await?; + } + write_atomically(&toml_path, &desired) +} + +fn is_codex_mcp_profile(profile: &AgentProfile) -> bool { + let is_codex = profile.structured_adapter == Some(StructuredAdapter::Codex) + || matches!( + &profile.context_injection, + ContextInjection::ConventionFile { target } + if target + .rsplit(['/', '\\']) + .next() + .unwrap_or(target) + .eq_ignore_ascii_case("AGENTS.md") + ); + is_codex && codex_mcp_config_target(profile).is_some() +} + +fn codex_mcp_config_target(profile: &AgentProfile) -> Option<(&str, &str)> { + match profile.mcp.as_ref().map(|mcp| &mcp.config) { + Some(McpConfigStrategy::TomlConfigHome { target, home_env }) => { + Some((target.as_str(), home_env.as_str())) + } + _ => None, + } +} + +/// Rend le contenu du `config.toml` Codex (table `[mcp_servers.idea]`) en réutilisant +/// l'encodeur TOML partagé [`domain::McpServerWiring::to_config_toml`] (D2) — même +/// source de wiring que la déclaration `.mcp.json` Claude, donc zéro dérive. +fn mcp_server_entry_toml(profile: &AgentProfile, runtime: Option<&McpRuntime>) -> String { + let transport = profile.mcp.as_ref().map_or(McpTransport::Stdio, |m| m.transport); + let (command, args) = match runtime { + Some(rt) => ( + rt.exe.clone(), + vec![ + "mcp-server".to_owned(), + "--endpoint".to_owned(), + rt.endpoint.clone(), + "--project".to_owned(), + rt.project_id.clone(), + "--requester".to_owned(), + rt.requester.clone(), + ], + ), + None => ("idea".to_owned(), vec!["mcp-server".to_owned()]), + }; + domain::McpServerWiring::new(command, args, transport).to_config_toml() +} + fn merge_claude_settings_json(existing: &str, project_root: &str) -> Option { let mut doc = match serde_json::from_str::(existing) { Ok(Value::Object(map)) => Value::Object(map), @@ -3284,6 +3422,88 @@ mod mcp_e2e_loopback_tests { (Arc::new(service), sessions, mailbox) } + /// Codex twin of [`build_service`]: identical wiring, but the single profile + /// targets the **Codex** structured adapter with a `TomlConfigHome` MCP strategy + /// (`$CODEX_HOME/config.toml`). This is the couple the bridge guard + /// (`materializes_idea_bridge`) must now let through — proving Codex is eligible + /// for `idea_ask_agent`, exactly like Claude. No real `codex` binary is ever + /// spawned: the runtime/PTY are the same in-memory fakes. + fn build_service_codex( + contexts: FakeContexts, + ) -> ( + Arc, + Arc, + Arc, + ) { + let profiles = Arc::new(FakeProfiles(Arc::new(vec![AgentProfile::new( + ProfileId::from_uuid(Uuid::from_u128(9)), + "Codex CLI", + "codex", + Vec::new(), + ContextInjection::stdin(), + None, + "{agentRunDir}", + None, + ) + .unwrap() + .with_structured_adapter(StructuredAdapter::Codex) + .with_mcp(McpCapability::new( + McpConfigStrategy::toml_config_home(".codex/config.toml", "CODEX_HOME").unwrap(), + McpTransport::Stdio, + ))]))); + let sessions = Arc::new(TerminalSessions::new()); + let mailbox = Arc::new(InMemoryMailbox::new()); + let bus = Arc::new(NoopBus); + let create = Arc::new(CreateAgentFromScratch::new( + Arc::new(contexts.clone()), + Arc::new(SeqIds(Mutex::new(1))), + bus.clone(), + )); + let launch = Arc::new(LaunchAgent::new( + Arc::new(contexts.clone()), + Arc::clone(&profiles) as Arc, + Arc::new(FakeRuntime), + Arc::new(FakeFs), + Arc::new(FakePty), + Arc::new(FakeSkills), + Arc::clone(&sessions), + bus.clone(), + Arc::new(SeqIds(Mutex::new(1))), + Arc::new(FakeRecall), + None, + )); + let list = Arc::new(ListAgents::new(Arc::new(contexts.clone()))); + let close = Arc::new(CloseTerminal::new(Arc::new(FakePty), Arc::clone(&sessions))); + let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts))); + let create_skill = Arc::new(CreateSkill::new( + Arc::new(FakeSkills) as Arc, + Arc::new(SeqIds(Mutex::new(1))), + )); + let input = Arc::new(MediatedInbox::with_pty( + Arc::clone(&mailbox), + Arc::new(SystemMillisClock), + Arc::new(FakePty) as Arc, + )) as Arc; + let conversations = Arc::new(InMemoryConversationRegistry::new()) + as Arc; + let service = OrchestratorService::new( + create, + launch, + list, + close, + update, + create_skill, + Arc::clone(&profiles) as Arc, + Arc::clone(&sessions), + ) + .with_input_mediator( + input, + Arc::clone(&mailbox) as Arc, + ) + .with_conversations(conversations); + (Arc::new(service), sessions, mailbox) + } + /// Pre-seeds a live PTY terminal session for `agent_id` so `ask_agent` reuses it. fn seed_live_pty(sessions: &TerminalSessions, agent_id: AgentId, session_id: SessionId) { sessions.insert( @@ -3519,6 +3739,101 @@ mod mcp_e2e_loopback_tests { handle.stop(); } + // ----------------------------------------------------------------------- + // 2b. Codex twin of the round-trip above: the *only* difference is the target + // profile (Codex + TomlConfigHome MCP). It proves the bridge guard now lets + // a Codex target through AND that the inline ask→reply loop still completes. + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn ask_then_reply_round_trips_inline_over_real_loopback_codex() { + let contexts = FakeContexts::new(); + let agent_id = contexts.seed_agent("architect"); + let (service, sessions, mailbox) = build_service_codex(contexts); + // Target already live in the PTY registry, so the ask reuses its terminal. + seed_live_pty( + &sessions, + agent_id, + SessionId::from_uuid(Uuid::from_u128(4242)), + ); + let proj = project(); + let (handle, endpoint) = start_real_server(service, &proj, None); + + // Connection A: the asker. + let conn_a = connect_client(&endpoint).await; + let (read_a, mut write_a) = tokio::io::split(conn_a); + let mut reader_a = BufReader::new(read_a); + write_a + .write_all(handshake_line(&project_id_arg(&proj), "agent-asker").as_bytes()) + .await + .unwrap(); + write_a + .write_all( + tools_call_line( + 7, + "idea_ask_agent", + json!({ "target": "architect", "task": "What is the answer?" }), + ) + .as_bytes(), + ) + .await + .unwrap(); + write_a.flush().await.unwrap(); + + // The ask is now blocked awaiting the reply — observe the pending ticket. + tokio::time::timeout(TIMEOUT, async { + while mailbox.pending(&agent_id) == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("ask must enqueue a ticket"); + + // Connection B: the target rendering its result. Its handshake requester is + // the target agent's id, which the server injects as the Reply `from`. + let conn_b = connect_client(&endpoint).await; + let (read_b, mut write_b) = tokio::io::split(conn_b); + let mut reader_b = BufReader::new(read_b); + write_b + .write_all(handshake_line(&project_id_arg(&proj), &agent_id.to_string()).as_bytes()) + .await + .unwrap(); + write_b + .write_all( + tools_call_line(8, "idea_reply", json!({ "result": "the answer is 42" })) + .as_bytes(), + ) + .await + .unwrap(); + write_b.flush().await.unwrap(); + let reply_resp = read_one_response(&mut reader_b) + .await + .expect("a reply ack line"); + assert_eq!( + reply_resp["result"]["isError"], + json!(false), + "got {reply_resp}" + ); + + // The asker now receives its inline result. + let resp = read_one_response(&mut reader_a) + .await + .expect("an ask response line"); + assert_eq!(resp["id"], json!(7)); + assert!(resp["error"].is_null(), "transport error: {resp}"); + let result = &resp["result"]; + assert_eq!(result["isError"], json!(false), "got {result}"); + assert_eq!( + result["content"][0]["text"].as_str().expect("text block"), + "the answer is 42", + "ask reply must be returned inline over the real loopback (Codex target); got {result}" + ); + + drop(write_a); + drop(write_b); + handle.stop(); + } + // ----------------------------------------------------------------------- // 3. idea_reply with no in-flight ask ⇒ typed tool error (isError:true), no // panic, connection stays healthy for a follow-up call. diff --git a/crates/app-tauri/tests/chat_bridge.rs b/crates/app-tauri/tests/chat_bridge.rs index 9d4fb5f..88f8667 100644 --- a/crates/app-tauri/tests/chat_bridge.rs +++ b/crates/app-tauri/tests/chat_bridge.rs @@ -57,7 +57,7 @@ fn final_chunk(s: &str) -> ReplyChunk { fn chunk_from_event_maps_text_delta() { assert_eq!( chunk_from_event(ReplyEvent::TextDelta { text: "hi".into() }), - ReplyChunk::TextDelta { text: "hi".into() } + Some(ReplyChunk::TextDelta { text: "hi".into() }) ); } @@ -67,9 +67,9 @@ fn chunk_from_event_maps_tool_activity() { chunk_from_event(ReplyEvent::ToolActivity { label: "reads file".into() }), - ReplyChunk::ToolActivity { + Some(ReplyChunk::ToolActivity { label: "reads file".into() - } + }) ); } @@ -79,12 +79,19 @@ fn chunk_from_event_maps_final() { chunk_from_event(ReplyEvent::Final { content: "done".into() }), - ReplyChunk::Final { + Some(ReplyChunk::Final { content: "done".into() - } + }) ); } +#[test] +fn chunk_from_event_drops_heartbeat() { + // A heartbeat is a non-terminal liveness proof with no chat content (lot 1) ⇒ no + // wire chunk; the pump skips it while still draining to the Final. + assert_eq!(chunk_from_event(ReplyEvent::Heartbeat), None); +} + // --------------------------------------------------------------------------- // agent_send pump behaviour: deltas* then exactly one Final (zone 2) // --------------------------------------------------------------------------- @@ -94,7 +101,10 @@ fn chunk_from_event_maps_final() { /// end. This is the body of the spawned pump thread, run inline. fn pump_turn(bridge: &ChatBridge, session: &SessionId, gen: u64, events: Vec) { for event in events { - let _ = bridge.send_output(session, chunk_from_event(event)); + // Mirror the real pump: heartbeats map to no chunk and are skipped. + if let Some(chunk) = chunk_from_event(event) { + let _ = bridge.send_output(session, chunk); + } } bridge.detach_if(session, gen); } diff --git a/crates/application/src/agent/catalogue.rs b/crates/application/src/agent/catalogue.rs index fa28e41..25df1c8 100644 --- a/crates/application/src/agent/catalogue.rs +++ b/crates/application/src/agent/catalogue.rs @@ -80,8 +80,11 @@ pub fn reference_profiles() -> Vec { .expect("codex reference profile is valid") .with_structured_adapter(StructuredAdapter::Codex) .with_mcp(McpCapability::new( - McpConfigStrategy::config_file(".mcp.json") - .expect(".mcp.json is a valid relative MCP config target"), + // Codex lit ses serveurs MCP dans `$CODEX_HOME/config.toml`, pas `.mcp.json` : + // IdeA écrit ce TOML DANS le run dir et pointe `CODEX_HOME` dessus pour + // isoler l'agent du `~/.codex` global (miroir du `.mcp.json` de Claude). + McpConfigStrategy::toml_config_home(".codex/config.toml", "CODEX_HOME") + .expect(".codex/config.toml + CODEX_HOME is a valid MCP config target"), McpTransport::Stdio, )), AgentProfile::new( @@ -157,18 +160,36 @@ mod mcp_tests { } #[test] - fn claude_and_codex_mcp_use_config_file_mcp_json() { - for slug in ["claude", "codex"] { - let mcp = profile(slug).mcp.expect("mcp present"); - assert_eq!( - mcp.config, - McpConfigStrategy::ConfigFile { - target: ".mcp.json".to_owned() - }, - "profile `{slug}` should declare `.mcp.json`" - ); - assert_eq!(mcp.transport, McpTransport::Stdio); - } + fn claude_mcp_uses_config_file_mcp_json() { + let mcp = profile("claude").mcp.expect("mcp present"); + assert_eq!( + mcp.config, + McpConfigStrategy::ConfigFile { + target: ".mcp.json".to_owned() + }, + "Claude should declare `.mcp.json`" + ); + assert_eq!(mcp.transport, McpTransport::Stdio); + } + + #[test] + fn codex_mcp_uses_toml_config_home_codex() { + // Codex lit `$CODEX_HOME/config.toml`, pas `.mcp.json` : le seed doit déclarer + // la stratégie TOML isolée par `CODEX_HOME` (pendant Codex de Claude). + let mcp = profile("codex").mcp.expect("mcp present"); + assert_eq!( + mcp.config, + McpConfigStrategy::TomlConfigHome { + target: ".codex/config.toml".to_owned(), + home_env: "CODEX_HOME".to_owned(), + }, + "Codex should declare `.codex/config.toml` + CODEX_HOME" + ); + assert_eq!(mcp.transport, McpTransport::Stdio); + assert!( + profile("codex").materializes_idea_bridge(), + "the Codex seed must materialise the idea bridge" + ); } #[test] diff --git a/crates/application/src/agent/lifecycle.rs b/crates/application/src/agent/lifecycle.rs index 6b0a84b..2192ff6 100644 --- a/crates/application/src/agent/lifecycle.rs +++ b/crates/application/src/agent/lifecycle.rs @@ -1746,6 +1746,36 @@ impl LaunchAgent { }, } } + domain::profile::McpConfigStrategy::TomlConfigHome { target, home_env } => { + // Pendant Codex de `ConfigFile` : on écrit un `config.toml` (table + // `[mcp_servers.idea]`, via l'encodeur TOML partagé D2) au chemin + // `/`, et on pousse `home_env` (ex. `CODEX_HOME`) vers + // le **dossier parent** de ce fichier. Codex lit alors ses serveurs MCP + // dans ce `config.toml` ISOLÉ au run dir, jamais le `~/.codex` global. + let path = RemotePath::new(join(run_dir, target)); + let declaration = mcp_server_wiring(mcp.transport, runtime).to_config_toml(); + // Même régime clobber/non-clobber que `.mcp.json` : avec un runtime réel + // (lancement app-tauri) on régénère/clobber à chaque (re)lancement (l'exe + // `$APPIMAGE` et l'endpoint dérivent entre runs) ; sans runtime + // (orchestrateur / hot-swap / tests) on reste non-clobbering pour ne pas + // écraser une déclaration réelle par la minimale. + match runtime { + Some(_) => { + let _ = self.fs.write(&path, declaration.as_bytes()).await; + } + None => match self.fs.exists(&path).await { + Ok(true) => {} + Ok(false) => { + let _ = self.fs.write(&path, declaration.as_bytes()).await; + } + Err(_) => {} + }, + } + // `home_env` pointe sur le DOSSIER PARENT de `target` (ex. + // `{runDir}/.codex`), pas sur le fichier — Codex y cherche `config.toml`. + let home_dir = parent_dir(run_dir, target); + spec.env.push((home_env.clone(), home_dir)); + } domain::profile::McpConfigStrategy::Flag { flag } => { // Pass the server via a launch flag (e.g. `--mcp-config {path}`). The // config path is the run dir itself (the CLI's cwd), where the server @@ -1858,58 +1888,40 @@ fn mcp_server_declaration( transport: domain::profile::McpTransport, runtime: Option<&McpRuntime>, ) -> String { - // Common `.mcp.json`-style shape (Claude Code et CLIs apparentées). The transport - // is surfaced so a socket-based CLI can be wired later without changing this seam. - let transport_label = match transport { - domain::profile::McpTransport::Stdio => "stdio", - domain::profile::McpTransport::Socket => "socket", - }; - - // `command` + extra args depend on whether OS/runtime facts were injected. - // Each `args` entry is emitted as an escaped JSON string so an exe path or - // endpoint with spaces/backslashes/quotes stays valid JSON. - let (command, extra_args) = match runtime { - Some(rt) => { - let arg = |label: &str, value: &str| { - format!( - ",\n {},\n {}", - json_string(label), - json_string(value) - ) - }; - let extra = format!( - "{}{}{}", - arg("--endpoint", &rt.endpoint), - arg("--project", &rt.project_id), - arg("--requester", &rt.requester), - ); - (rt.exe.as_str(), extra) - } - None => ("idea", String::new()), - }; - let command = json_string(command); - - format!( - r#"{{ - "mcpServers": {{ - "idea": {{ - "command": {command}, - "args": [ - "mcp-server"{extra_args} - ], - "transport": "{transport_label}" - }} - }} -}} -"# - ) + mcp_server_wiring(transport, runtime).to_mcp_json() } -/// Wraps a string as a JSON string literal (quotes + escaping), reusing the path -/// escaper so an exe path / endpoint with spaces, backslashes or quotes stays valid -/// JSON in the generated `.mcp.json`. -fn json_string(s: &str) -> String { - format!("\"{}\"", json_escape(s)) +/// Builds the IdeA MCP server **wiring** (`command` + `args` + transport) shared by +/// every materialisation format (cadrage Codex D2). It is the **single source of +/// truth** for *what* the bridge is launched as; the concrete bytes (`.mcp.json` +/// JSON for Claude, `config.toml` TOML for Codex) are produced by the domain +/// encoders [`domain::McpServerWiring::to_mcp_json`] / +/// [`domain::McpServerWiring::to_config_toml`], so the two sites can never drift. +/// +/// `Some(runtime)` ⇒ the **real** wiring (this IdeA exe + the project's loopback +/// endpoint/project/requester); `None` ⇒ the coherent **minimal** `idea mcp-server` +/// fallback (orchestrator / hot-swap / tests). +#[must_use] +fn mcp_server_wiring( + transport: domain::profile::McpTransport, + runtime: Option<&McpRuntime>, +) -> domain::McpServerWiring { + let (command, args) = match runtime { + Some(rt) => ( + rt.exe.clone(), + vec![ + "mcp-server".to_owned(), + "--endpoint".to_owned(), + rt.endpoint.clone(), + "--project".to_owned(), + rt.project_id.clone(), + "--requester".to_owned(), + rt.requester.clone(), + ], + ), + None => ("idea".to_owned(), vec!["mcp-server".to_owned()]), + }; + domain::McpServerWiring::new(command, args, transport) } /// Builds an absolute path string by joining a [`ProjectPath`] with a relative @@ -1919,6 +1931,18 @@ fn join(base: &ProjectPath, rel: &str) -> String { format!("{b}/{rel}") } +/// Resolves the **parent directory** (absolute) of `/` — used to point a +/// CLI's `home_env` (e.g. `CODEX_HOME`) at the directory *containing* the materialised +/// config file (`config.toml`), not the file itself. When `rel` has no separator +/// (file directly in the run dir), the parent is the run dir. +fn parent_dir(base: &ProjectPath, rel: &str) -> String { + let full = join(base, rel); + match full.rsplit_once(['/', '\\']) { + Some((parent, _)) if !parent.is_empty() => parent.to_owned(), + _ => base.as_str().trim_end_matches(['/', '\\']).to_owned(), + } +} + /// Computes an agent's isolated run directory `/.ideai/run//` /// (ARCHITECTURE §14.1). This is the PTY cwd for the agent — never the project /// root — guaranteeing that two distinct agents on the same project root get two diff --git a/crates/application/src/agent/mod.rs b/crates/application/src/agent/mod.rs index 1ddb711..deabf16 100644 --- a/crates/application/src/agent/mod.rs +++ b/crates/application/src/agent/mod.rs @@ -16,7 +16,7 @@ mod usecases; pub(crate) use lifecycle::unique_md_path; pub(crate) use lifecycle::ReattachDecision; -pub use structured::send_blocking; +pub use structured::{drain_with_readiness, send_blocking}; pub use catalogue::{reference_profile_id, reference_profiles, selectable_reference_profiles}; pub use inspect::{InspectConversation, InspectConversationInput, InspectConversationOutput}; diff --git a/crates/application/src/agent/structured.rs b/crates/application/src/agent/structured.rs index eeaa5ef..b2a452a 100644 --- a/crates/application/src/agent/structured.rs +++ b/crates/application/src/agent/structured.rs @@ -14,7 +14,10 @@ use std::time::Duration; +use domain::input::InputMediator; +use domain::ids::AgentId; use domain::ports::{AgentSession, AgentSessionError, ReplyEvent}; +use domain::readiness::{ReadinessPolicy, ReadinessSignal}; /// Envoie `prompt` à la session vivante puis **draine le flux de réponse jusqu'au /// [`ReplyEvent::Final`]**, et retourne son contenu agrégé. @@ -39,14 +42,83 @@ pub async fn send_blocking( session: &dyn AgentSession, prompt: &str, timeout: Option, +) -> Result { + drain_bounded_events(session, prompt, timeout, |_event| {}, |_signal| {}).await +} + +/// Comme [`send_blocking`], mais **branche la readiness** : à chaque événement du +/// tour, [`ReadinessPolicy::classify`] est consulté et, dès qu'il renvoie +/// [`ReadinessSignal::TurnEnded`] (le `Final`), le médiateur d'entrée est notifié +/// (`mark_idle(agent)`) pour que la FIFO de l'agent avance — **sans** dépendre d'un +/// `idea_reply` explicite ni du sniff de prompt PTY (chantier readiness/heartbeat, +/// lot 1, fix de la cause racine du blocage `Busy`). +/// +/// DRY : **un seul** chemin de lecture du flux (la boucle de [`drain_bounded`]) ; +/// cette fonction n'est que `send_blocking` muni d'un *sink* de readiness. Le `Final` +/// réveille donc à la fois le `pending` (via la valeur de retour) **et** la FIFO (via +/// `mark_idle`). `idea_reply` reste un signal alternatif (premier arrivé gagne) côté +/// orchestrateur. +/// +/// # Errors +/// Identiques à [`send_blocking`] (échec `send`/décodage, flux clos sans `Final`, +/// timeout). +pub async fn drain_with_readiness( + session: &dyn AgentSession, + prompt: &str, + timeout: Option, + mediator: &dyn InputMediator, + agent: AgentId, +) -> Result { + // `on_signal` ne reçoit QUE les événements terminaux (le `Final` ⇒ `TurnEnded`) : + // la readiness ne classe pas les non-terminaux. Pour le **battement** de vivacité + // (lot 2) on a besoin de notifier le médiateur à CHAQUE événement non terminal + // (delta / activité / heartbeat) ⇒ on passe un sink d'événement bruts `on_event`. + drain_bounded_events( + session, + prompt, + timeout, + |event| { + // Tout événement **non terminal** prouve la vivacité ⇒ un battement. + if !matches!(event, ReplyEvent::Final { .. }) { + mediator.mark_alive(agent); + } + }, + |signal| { + if signal == ReadinessSignal::TurnEnded { + mediator.mark_idle(agent); + } + }, + ) + .await +} + +/// Ouvre le flux du tour (`send`) et le **draine jusqu'au `Final`**, en appliquant +/// la borne temporelle `timeout`, en notifiant `on_event` à **chaque** événement brut +/// (pour le battement de vivacité, lot 2) et `on_signal` à chaque [`ReadinessSignal`] +/// dérivé par [`ReadinessPolicy`] (le `Final` ⇒ `TurnEnded`). +/// +/// **Chemin de lecture unique** (DRY) : `send_blocking` et `drain_with_readiness` +/// passent tous deux par ici, en différant seulement par leurs *sinks*. La session +/// **reste vivante** sur timeout (on ne `shutdown` rien ici, §17.1). +async fn drain_bounded_events( + session: &dyn AgentSession, + prompt: &str, + timeout: Option, + on_event: impl FnMut(&ReplyEvent), + on_signal: impl FnMut(ReadinessSignal), ) -> Result { match timeout { - Some(dur) => match tokio::time::timeout(dur, drain_to_final(session, prompt)).await { + Some(dur) => match tokio::time::timeout( + dur, + drain_to_final(session, prompt, on_event, on_signal), + ) + .await + { Ok(result) => result, // La session **reste vivante** : on ne `shutdown` rien ici (§17.1). Err(_elapsed) => Err(AgentSessionError::Timeout), }, - None => drain_to_final(session, prompt).await, + None => drain_to_final(session, prompt, on_event, on_signal).await, } } @@ -56,18 +128,124 @@ pub async fn send_blocking( /// après le `Final` il ne produit plus rien. On le parcourt donc simplement /// jusqu'à rencontrer le `Final` (et on retourne son contenu) ; si le flux /// s'épuise avant, c'est un tour interrompu → erreur [`AgentSessionError::Io`]. +/// +/// Chaque événement est classé par [`ReadinessPolicy`] et le signal éventuel est +/// remonté à `on_signal` (le `Final` ⇒ [`ReadinessSignal::TurnEnded`]). Deltas, +/// activités et heartbeats sont non terminaux ⇒ ignorés par le rendez-vous synchrone. async fn drain_to_final( session: &dyn AgentSession, prompt: &str, + mut on_event: impl FnMut(&ReplyEvent), + mut on_signal: impl FnMut(ReadinessSignal), ) -> Result { let stream = session.send(prompt).await?; for event in stream { + // Battement de vivacité (lot 2) : notifié pour CHAQUE événement brut, avant le + // classement readiness. Le sink décide (les non-terminaux prouvent la vivacité). + on_event(&event); + if let Some(signal) = ReadinessPolicy::classify(&event) { + on_signal(signal); + } if let ReplyEvent::Final { content } = event { return Ok(content); } - // TextDelta / ToolActivity : ignorés par le rendez-vous synchrone. + // TextDelta / ToolActivity / Heartbeat : non terminaux, ignorés ici. } Err(AgentSessionError::Io( "le flux de réponse s'est terminé sans événement Final".to_string(), )) } + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Mutex; + + use domain::ids::SessionId; + use domain::input::{AgentBusyState, SubmitConfig}; + use domain::mailbox::{PendingReply, Ticket}; + use domain::ports::PtyHandle; + + fn agent(n: u128) -> AgentId { + AgentId::from_uuid(uuid::Uuid::from_u128(n)) + } + + /// Session factice : `send` rejoue une liste fixe d'événements (terminée par un + /// `Final`). + struct FakeSession { + events: Vec, + } + #[async_trait::async_trait] + impl AgentSession for FakeSession { + fn id(&self) -> SessionId { + SessionId::from_uuid(uuid::Uuid::from_u128(1)) + } + fn conversation_id(&self) -> Option { + None + } + async fn send( + &self, + _prompt: &str, + ) -> Result { + Ok(Box::new(self.events.clone().into_iter())) + } + async fn shutdown(&self) -> Result<(), AgentSessionError> { + Ok(()) + } + } + + /// Médiateur factice qui enregistre l'ordre des `mark_alive` / `mark_idle`. + #[derive(Default)] + struct RecordingMediator { + calls: Mutex>, + } + impl InputMediator for RecordingMediator { + fn enqueue(&self, _agent: AgentId, _ticket: Ticket) -> PendingReply { + unreachable!("non utilisé par drain_with_readiness") + } + fn preempt(&self, _agent: AgentId) {} + fn mark_idle(&self, _agent: AgentId) { + self.calls.lock().unwrap().push("idle"); + } + fn mark_alive(&self, _agent: AgentId) { + self.calls.lock().unwrap().push("alive"); + } + fn busy_state(&self, _agent: AgentId) -> AgentBusyState { + AgentBusyState::Idle + } + fn bind_handle(&self, _agent: AgentId, _handle: PtyHandle) {} + fn bind_handle_with_prompt( + &self, + _agent: AgentId, + _handle: PtyHandle, + _pattern: Option, + _submit: SubmitConfig, + ) { + } + } + + #[tokio::test] + async fn drain_marks_alive_on_each_non_terminal_then_idle_on_final() { + let session = FakeSession { + events: vec![ + ReplyEvent::TextDelta { text: "a".into() }, + ReplyEvent::ToolActivity { label: "lit".into() }, + ReplyEvent::Heartbeat, + ReplyEvent::Final { + content: "fini".into(), + }, + ], + }; + let mediator = RecordingMediator::default(); + let out = drain_with_readiness(&session, "go", None, &mediator, agent(1)) + .await + .expect("drain ok"); + assert_eq!(out, "fini"); + // Trois battements (delta, activité, heartbeat) PUIS l'idle sur le Final. + assert_eq!( + *mediator.calls.lock().unwrap(), + vec!["alive", "alive", "alive", "idle"], + "un battement par événement non terminal, idle au Final (pas de battement sur le Final)" + ); + } +} diff --git a/crates/application/src/lib.rs b/crates/application/src/lib.rs index c4d405d..86525fd 100644 --- a/crates/application/src/lib.rs +++ b/crates/application/src/lib.rs @@ -28,8 +28,8 @@ pub mod terminal; pub mod window; pub use agent::{ - reference_profile_id, reference_profiles, selectable_reference_profiles, send_blocking, - ChangeAgentProfile, ChangeAgentProfileInput, ChangeAgentProfileOutput, ConfigureProfiles, + drain_with_readiness, reference_profile_id, reference_profiles, selectable_reference_profiles, + send_blocking, ChangeAgentProfile, ChangeAgentProfileInput, ChangeAgentProfileOutput, ConfigureProfiles, ConfigureProfilesInput, ConfigureProfilesOutput, CreateAgentFromScratch, CreateAgentInput, CreateAgentOutput, DeleteAgent, DeleteAgentInput, DeleteProfile, DeleteProfileInput, DetectProfiles, DetectProfilesInput, DetectProfilesOutput, FirstRunState, FirstRunStateOutput, diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index 3e23ce5..6c32450 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/crates/application/src/orchestrator/service.rs @@ -32,8 +32,9 @@ use domain::{ use crate::conversation::RecordTurn; use crate::agent::{ - CreateAgentFromScratch, CreateAgentInput, LaunchAgent, LaunchAgentInput, ListAgents, - ListAgentsInput, McpRuntime, ReattachDecision, UpdateAgentContext, UpdateAgentContextInput, + drain_with_readiness, CreateAgentFromScratch, CreateAgentInput, LaunchAgent, LaunchAgentInput, + ListAgents, ListAgentsInput, McpRuntime, ReattachDecision, UpdateAgentContext, + UpdateAgentContextInput, }; use crate::error::AppError; use crate::orchestrator::{ @@ -41,7 +42,7 @@ use crate::orchestrator::{ ReadMemoryInput, WriteMemory, WriteMemoryInput, }; use crate::skill::{CreateSkill, CreateSkillInput}; -use crate::terminal::{CloseTerminal, CloseTerminalInput, TerminalSessions}; +use crate::terminal::{CloseTerminal, CloseTerminalInput, StructuredSessions, TerminalSessions}; /// Default terminal geometry for an orchestrator-launched agent cell. The UI /// resizes the PTY to the real cell size on attach; these are sane starting rows @@ -57,7 +58,13 @@ const DEFAULT_COLS: u16 = 80; /// **without killing the session**, so the requester can retry. Internal and /// intentionally not yet config-exposed (it may become a per-project setting /// without changing the contract). -const ASK_AGENT_TIMEOUT: Duration = Duration::from_secs(300); +/// +/// TEMPORAIRE (demande utilisateur 2026-06-13) : la borne de 300 s coupait des tours +/// délégués légitimement très longs (gros refactors + tests). On la relève donc à 24 h +/// (≈ « pas de limite ») le temps de concevoir un vrai signal de vivacité (savoir si +/// l'agent travaille encore) qui remplacera ce timeout brut. NE PAS considérer comme +/// définitif : à reconvertir en borne courte + heartbeat once that liveness signal exists. +const ASK_AGENT_TIMEOUT: Duration = Duration::from_secs(24 * 60 * 60); /// Borne d'attente **en file** pour acquérir le verrou de tour d'un agent (A0, /// cadrage v5 §4). @@ -72,7 +79,20 @@ const ASK_AGENT_TIMEOUT: Duration = Duration::from_secs(300); /// [`domain::ports::AgentSessionError::Timeout`]) ; le tour en cours n'est pas /// affecté. Largeur = un tour complet + sa propre file ⇒ on autorise deux tours /// pleins d'attente. -const ASK_QUEUE_WAIT_CAP: Duration = Duration::from_secs(600); +/// +/// TEMPORAIRE (cf. [`ASK_AGENT_TIMEOUT`]) : relevé à 48 h (≈ deux tours « sans limite ») +/// en cohérence avec la levée temporaire de la borne de tour, le temps d'un vrai +/// signal de vivacité. +const ASK_QUEUE_WAIT_CAP: Duration = Duration::from_secs(48 * 60 * 60); + +/// Borne de tour effective (lot 2, timeouts pilotés par profil) : le +/// `turn_timeout_ms` du profil de la cible **quand il existe**, sinon le défaut +/// historique [`ASK_AGENT_TIMEOUT`] (zéro régression pour un profil sans `liveness` / +/// sans `turn_timeout_ms`). Pure et testable sans wiring de service. +#[must_use] +fn resolve_turn_timeout(turn_timeout_ms: Option) -> Duration { + turn_timeout_ms.map_or(ASK_AGENT_TIMEOUT, |ms| Duration::from_millis(u64::from(ms))) +} /// Fournit les faits OS/runtime (exe + endpoint projet) pour écrire la déclaration MCP /// réelle quand l'orchestrateur (re)lance une cible sur le chemin `ask`. Implémenté dans @@ -182,6 +202,15 @@ pub struct OrchestratorService { /// injectée avec [`Self::record_turn`]. La couche `application` reste **pure** : pas /// de `SystemTime::now()` brut ici, le temps vient du port injecté au composition root. clock: Option>, + /// Registre des **sessions IA structurées** vivantes (§17.5), pour router un `ask` + /// vers une cible à `structured_adapter` directement sur sa session + /// ([`drain_with_readiness`](crate::agent::drain_with_readiness)) — le `Final` + /// déterministe réveille le `pending` **et** marque l'agent `Idle` (chantier + /// readiness/heartbeat, lot 1, fix du blocage `Busy`). Injecté via + /// [`Self::with_structured`] ; `None` ⇒ on conserve **uniquement** le chemin + /// PTY/mailbox legacy (zéro régression pour les call sites/tests qui ne le branchent + /// pas). + structured: Option>, } /// Bundle des quatre use cases C7 sous [`domain::fileguard::FileGuard`], injectés @@ -244,9 +273,19 @@ impl OrchestratorService { context_guard: None, record_turn: None, clock: None, + structured: None, } } + /// Branche le registre des [`StructuredSessions`] (§17.5) pour router un `ask` + /// vers une cible **structurée** sur sa propre session. Builder additif (signature + /// de [`Self::new`] inchangée) ; `None` ⇒ chemin PTY/mailbox legacy uniquement. + #[must_use] + pub fn with_structured(mut self, structured: Arc) -> Self { + self.structured = Some(structured); + self + } + /// Branche les use cases C7 (`context.*`/`memory.*`) sous /// [`domain::fileguard::FileGuard`]. Builder additif (signature de [`Self::new`] /// inchangée). @@ -750,6 +789,42 @@ impl OrchestratorService { // Poser l'arête d'attente A→B (retirée en fin de tour par le RAII `_edge`). let _edge = requester.map(|from| WaitEdgeGuard::new(self, from, agent_id)); + // ── Chemin **structuré** (readiness/heartbeat lot 1) ────────────────────── + // Une cible à `structured_adapter` n'a **pas** de PTY : `ensure_live_pty` + // échouerait, et le tour ne pourrait se débloquer que par un `idea_reply` + // explicite (cause racine du blocage `Busy`). Quand le registre structuré est + // câblé et que la cible a une session vivante, on draine son tour via + // `drain_with_readiness` : le `Final` déterministe réveille le `pending` (valeur + // de retour) **et** marque l'agent `Idle` (`mark_idle`). On enregistre tout de + // même un ticket dans la FIFO pour la comptabilité busy et pour préserver + // `idea_reply` comme signal **alternatif** (premier arrivé gagne). + if let Some(structured) = self.structured.as_ref() { + // Lot 1b — auto-lancement d'une cible **froide** structurée : si le registre + // est câblé, qu'aucune session ne vit encore pour la cible, mais que son + // profil porte un `structured_adapter`, on **démarre** sa session via le + // launcher (qui route §17.4 vers `launch_structured` et l'insère dans CE + // même registre, avec la conf MCP matérialisée) plutôt que de tomber dans + // `ensure_live_pty` — chemin PTY qui échouerait pour une cible sans PTY. + // Une cible SANS `structured_adapter` (agent PTY/TUI legacy) conserve le + // chemin `ensure_live_pty` ci-dessous (zéro régression). + if let Some(session) = self + .ensure_structured_session(project, agent_id, &agent.profile_id, structured) + .await? + { + return self + .ask_structured( + project, + agent_id, + &target, + conversation_id, + requester, + task, + session.as_ref(), + ) + .await; + } + } + // 1. Garantir la cible vivante en PTY pour CE fil ; lier sa session à la // conversation, et brancher son handle d'entrée sur le médiateur (livraison). let handle = self @@ -786,6 +861,10 @@ impl OrchestratorService { } None => Ticket::from_human(ticket_id, conversation_id, requester_label, task), }; + // Timeout de tour piloté par profil (lot 2) : `turn_timeout_ms` de la cible si + // défini, sinon le défaut [`ASK_AGENT_TIMEOUT`]. Arme aussi le seuil de stall sur + // le médiateur AVANT l'enqueue (consommé au start_turn). + let turn_timeout = self.turn_timeout_for(project, agent_id).await; let pending = input.enqueue(agent_id, ticket); // Delivery is the mediator's responsibility (`InputMediator::enqueue` writes the // turn into the bound handle). The service no longer writes the PTY directly — @@ -793,7 +872,7 @@ impl OrchestratorService { // 3. Attendre la réponse, bornée. Timeout/canal fermé ⇒ retirer le ticket // (cible laissée vivante) et renvoyer une erreur typée. - match tokio::time::timeout(ASK_AGENT_TIMEOUT, pending).await { + match tokio::time::timeout(turn_timeout, pending).await { Ok(Ok(result)) => { // Checkpoint Response (P6b, best-effort) : persister la réponse AVANT de // déplacer `result` dans `reply_outcome`. Source = la **cible** (c'est @@ -821,6 +900,120 @@ impl OrchestratorService { } } + /// Chemin `ask` **structuré** (readiness/heartbeat lot 1) : la cible a un + /// `structured_adapter` et une session vivante ([`StructuredSessions`]). On pilote + /// son tour **directement** sur le port [`domain::ports::AgentSession`] et on le + /// draine via [`drain_with_readiness`] : le [`domain::ports::ReplyEvent::Final`] + /// déterministe rend le contenu (réveille le `pending`) **et** marque l'agent `Idle` + /// (`mark_idle`), sans dépendre d'un `idea_reply` explicite ni d'un PTY (que la cible + /// n'a pas). C'est le fix de la cause racine du blocage `Busy`. + /// + /// On enregistre tout de même un ticket dans la FIFO (`enqueue`) pour la comptabilité + /// busy et pour **préserver `idea_reply` comme signal alternatif** : on attend la + /// **première** des deux issues (réponse de la session OU résolution du `pending`). + /// Le checkpoint Prompt/Response best-effort est conservé à l'identique du chemin PTY. + #[allow(clippy::too_many_arguments)] + async fn ask_structured( + &self, + project: &Project, + agent_id: AgentId, + target: &str, + conversation_id: domain::conversation::ConversationId, + requester: Option, + task: String, + session: &dyn domain::ports::AgentSession, + ) -> Result { + let (input, mailbox) = match (&self.input, &self.mailbox) { + (Some(i), Some(m)) => (i, m), + _ => return Err(AppError::Invalid( + "la messagerie inter-agents (idea_ask_agent) n'est pas disponible : \ + médiateur d'entrée non câblé" + .to_owned(), + )), + }; + + // Checkpoint Prompt (best-effort), AVANT de déplacer `task` dans le ticket. + let prompt_source = match requester { + Some(from) => InputSource::agent(from), + None => InputSource::Human, + }; + self.record_turn_best_effort( + &project.root, + conversation_id, + prompt_source, + TurnRole::Prompt, + task.clone(), + ) + .await; + + // Ticket dans la FIFO : comptabilité busy + `idea_reply` comme signal alternatif. + let requester_label = self.requester_label(project, requester).await; + let ticket_id = TicketId::new_random(); + let ticket = match requester { + Some(from) => Ticket::from_agent( + ticket_id, + from, + conversation_id, + requester_label, + task.clone(), + ), + None => Ticket::from_human(ticket_id, conversation_id, requester_label, task.clone()), + }; + // Timeout de tour piloté par profil (lot 2) + armement du seuil de stall, AVANT + // l'enqueue qui démarre le tour (le médiateur arme alors sa fenêtre de vivacité). + let turn_timeout = self.turn_timeout_for(project, agent_id).await; + let pending = input.enqueue(agent_id, ticket); + + // Drainer le tour structuré (le `Final` ⇒ contenu + `mark_idle`), borné par le + // **même** garde-fou que le chemin PTY (profil ou défaut). On attend la + // **première** issue : le tour structuré OU un `idea_reply` explicite. + let drain = drain_with_readiness( + session, + &task, + Some(turn_timeout), + input.as_ref(), + agent_id, + ); + + let result = tokio::select! { + biased; + // Issue déterministe : la session a rendu son `Final`. + drained = drain => match drained { + Ok(content) => content, + Err(err) => { + mailbox.cancel_head(agent_id, ticket_id); + return Err(AppError::from(err)); + } + }, + // Issue alternative : un `idea_reply` explicite a résolu le ticket d'abord. + replied = pending => match replied { + Ok(content) => { + // La session draine encore en arrière-plan ; on s'assure que la FIFO + // avance même si le `Final` n'est pas encore observé. + input.mark_idle(agent_id); + content + } + Err(_cancelled) => { + mailbox.cancel_head(agent_id, ticket_id); + return Err(AppError::Process(format!( + "agent {target} : canal de réponse fermé avant un résultat" + ))); + } + }, + }; + + // Checkpoint Response (best-effort), AVANT de déplacer `result`. + self.record_turn_best_effort( + &project.root, + conversation_id, + InputSource::agent(agent_id), + TurnRole::Response, + result.clone(), + ) + .await; + Ok(self.reply_outcome(agent_id, target, result)) + } + /// `SubmitHumanInput` (cadrage C4 §5.3) — the **human** Envoyer path. /// /// The operator types into IdeA's mediated input; this resolves the `User↔Agent` @@ -1083,6 +1276,74 @@ impl OrchestratorService { }) } + /// Lot 1b — garantit une **session structurée vivante** pour la cible d'un `ask`. + /// + /// Retourne : + /// - `Ok(Some(session))` si la cible a déjà une session vivante, **ou** si son + /// profil porte un `structured_adapter` et qu'on vient de la (re)lancer ; + /// - `Ok(None)` si la cible n'est **pas** structurée (profil sans `structured_adapter`, + /// ou profil introuvable) ⇒ l'appelant retombe sur le chemin PTY `ensure_live_pty`. + /// + /// L'auto-lancement réutilise le **même** [`LaunchAgent`] que le chemin PTV + /// ([`Self::ensure_live_pty`]) avec le **même** `mcp_runtime` matérialisé : pour un + /// profil structuré, le launcher route §17.4 vers `launch_structured`, démarre la + /// session via la fabrique et l'insère dans le registre [`StructuredSessions`] + /// **partagé** (le même `Arc` que `self.structured`, câblé au composition root). + /// La conf MCP est donc matérialisée comme pour une cellule chat lancée à la main, + /// si bien que la cible voit les outils `idea_*` pour répondre. + async fn ensure_structured_session( + &self, + project: &Project, + agent_id: AgentId, + profile_id: &ProfileId, + structured: &Arc, + ) -> Result>, AppError> { + // Cible déjà chaude : route directe (aucun lancement). + if let Some(session) = structured.session_for_agent(&agent_id) { + return Ok(Some(session)); + } + + // Cible froide : ne (re)lancer que si le profil sait être piloté en mode + // structuré. Sinon (agent PTY/TUI legacy, ou profil introuvable) ⇒ chemin PTY. + let is_structured = self + .profiles + .list() + .await? + .into_iter() + .find(|p| &p.id == profile_id) + .is_some_and(|p| p.is_selectable()); + if !is_structured { + return Ok(None); + } + + // Démarrer la session via le launcher (route §17.4 → `launch_structured`, + // insère dans CE registre). Mêmes faits MCP que le chemin PTY pour que le pont + // `idea_*` de la cible se branche. `conversation_id: None` ⇒ le launcher dérive + // l'id de paire (User↔agent) ou réutilise celui de la cellule (P8a), comme pour + // un lancement direct utilisateur. + self.launch_agent + .execute(LaunchAgentInput { + project: project.clone(), + agent_id, + rows: DEFAULT_ROWS, + cols: DEFAULT_COLS, + node_id: None, + conversation_id: None, + mcp_runtime: self + .mcp_runtime_provider + .as_ref() + .and_then(|p| p.runtime_for(project, agent_id)), + }) + .await?; + + // Le launcher a inséré la session dans le registre partagé : la relire. + structured.session_for_agent(&agent_id).map(Some).ok_or_else(|| { + AppError::Process(format!( + "agent {agent_id} : aucune session structurée vivante après lancement" + )) + }) + } + /// Binds `conversation` to `session` in both the terminal registry (fast /// `session_for`) and the [`ConversationRegistry`] (domain thread state), when the /// latter is wired. Idempotent. @@ -1315,8 +1576,6 @@ impl OrchestratorService { profile_id: &ProfileId, target: &str, ) -> Result<(), AppError> { - use domain::profile::{McpConfigStrategy, StructuredAdapter}; - let Some(profile) = self .profiles .list() @@ -1327,21 +1586,20 @@ impl OrchestratorService { return Ok(()); }; - let honours_mcp_json = matches!( - profile.mcp.as_ref().map(|c| &c.config), - Some(McpConfigStrategy::ConfigFile { target }) if target == ".mcp.json" - ); - let is_claude = profile.structured_adapter == Some(StructuredAdapter::Claude); - - if honours_mcp_json && is_claude { + // Source de vérité UNIQUE (domaine) : un profil supporte le pont `idea_*` ssi + // IdeA matérialise réellement sa config MCP pour la CLI qu'il pilote — Claude + // via `.mcp.json`, Codex via `config.toml`/`CODEX_HOME`. Tout autre couple + // (y compris MCP absent) ⇒ repli fichier, pont non branché. + if profile.materializes_idea_bridge() { return Ok(()); } Err(AppError::Invalid(format!( "la cible '{target}' (profil '{}', adaptateur {:?}) ne supporte pas encore le \ - pont idea_* : la délégation inter-agents passe par un serveur MCP déclaré en \ - .mcp.json, que seul un profil Claude consomme aujourd'hui. Cible un agent au \ - profil Claude.", + pont idea_* : la délégation inter-agents passe par un serveur MCP qu'IdeA \ + matérialise pour la CLI cible, et ce profil ne déclare pas un couple \ + (adaptateur × stratégie MCP) pris en charge. Cible un agent dont le profil \ + expose le pont MCP (Claude ou Codex).", profile.name, profile.structured_adapter ))) } @@ -1383,6 +1641,54 @@ impl OrchestratorService { (profile.prompt_ready_pattern, submit) } + /// Résout la [`domain::profile::LivenessStrategy`] du profil de la cible (lot 2) : + /// le seuil de stagnation (`stall_after_ms`) et le timeout de tour + /// (`turn_timeout_ms`). `None`/absent ⇒ `(None, None)` : pas de détection de stall et + /// repli sur les bornes par défaut codées en dur (zéro régression pour un profil sans + /// bloc `liveness`). + async fn liveness_for_agent( + &self, + project: &Project, + agent_id: AgentId, + ) -> (Option, Option) { + let Some(agent) = self + .list_agents + .execute(ListAgentsInput { + project: project.clone(), + }) + .await + .ok() + .and_then(|out| out.agents.into_iter().find(|a| a.id == agent_id)) + else { + return (None, None); + }; + let Some(profile) = self + .profiles + .list() + .await + .ok() + .and_then(|ps| ps.into_iter().find(|p| p.id == agent.profile_id)) + else { + return (None, None); + }; + match profile.liveness { + Some(l) => (l.stall_after_ms, l.turn_timeout_ms), + None => (None, None), + } + } + + /// Borne de tour effective pour un `ask` : le `turn_timeout_ms` du profil de la cible + /// **quand il existe**, sinon le défaut historique [`ASK_AGENT_TIMEOUT`] (lot 2, + /// timeouts pilotés par profil). Arme aussi le seuil de stagnation sur le médiateur + /// avant l'enqueue (battement/`sweep_stalled`). + async fn turn_timeout_for(&self, project: &Project, agent_id: AgentId) -> Duration { + let (stall_after_ms, turn_timeout_ms) = self.liveness_for_agent(project, agent_id).await; + if let Some(input) = &self.input { + input.set_stall_threshold(agent_id, stall_after_ms); + } + resolve_turn_timeout(turn_timeout_ms) + } + /// Resolves a human-friendly profile reference (slug like `claude-code`, /// command like `claude`, or display name like `Claude Code`) to a configured /// [`ProfileId`]. Matching is universal — never hard-coded to one AI — by @@ -1459,6 +1765,18 @@ mod tests { use domain::profile::{AgentProfile, ContextInjection}; use domain::ProfileId; + // (c) — timeouts pilotés par profil (lot 2) : `turn_timeout_ms` prime sur le défaut. + #[test] + fn profile_turn_timeout_overrides_default() { + // Seuil de profil défini ⇒ il prime sur ASK_AGENT_TIMEOUT. + assert_eq!( + resolve_turn_timeout(Some(120_000)), + Duration::from_millis(120_000) + ); + // Absent (profil sans liveness / sans turn_timeout_ms) ⇒ repli sur le défaut. + assert_eq!(resolve_turn_timeout(None), ASK_AGENT_TIMEOUT); + } + fn profile(id: u128, name: &str, command: &str) -> AgentProfile { AgentProfile::new( ProfileId::from_uuid(uuid::Uuid::from_u128(id)), diff --git a/crates/application/tests/drain_with_readiness_lot1.rs b/crates/application/tests/drain_with_readiness_lot1.rs new file mode 100644 index 0000000..28b58f7 --- /dev/null +++ b/crates/application/tests/drain_with_readiness_lot1.rs @@ -0,0 +1,338 @@ +//! Lot 1 (readiness/heartbeat model-agnostique) — tests unitaires QA **indépendants** +//! du helper applicatif `drain_with_readiness`, **100 % fakes**. +//! +//! Couvre les points 5 et 6 du périmètre QA : +//! +//! - **Point 5** : un flux se terminant par `Final` appelle `mark_idle(agent)` +//! **exactement une fois** ; les `Heartbeat`/deltas/activités intermédiaires +//! n'appellent **jamais** `mark_idle`. +//! - **Point 6 (le bug d'origine)** : un agent cible qui **ne fait que renvoyer un +//! `Final`** (sans jamais appeler `idea_reply`) débloque bien la file via +//! `mark_idle` — c'est exactement la cause racine du blocage `Busy`. +//! +//! On teste au niveau `drain_with_readiness` (le rendez-vous synchrone qu'`ask_agent` +//! utilise sur la session structurée d'une cible) avec : +//! - un fake `AgentSession` scriptable (calqué sur `send_blocking_d1.rs`) ; +//! - un fake `InputMediator` qui **compte** les `mark_idle` par agent et journalise +//! l'ordre des appels, sans seconde file ni I/O. + +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Mutex; +use std::time::Duration; + +use async_trait::async_trait; + +use application::drain_with_readiness; +use domain::ids::AgentId; +use domain::input::{AgentBusyState, InputMediator}; +use domain::mailbox::{PendingReply, Ticket}; +use domain::ports::{AgentSession, AgentSessionError, ReplyEvent, ReplyStream}; +use domain::SessionId; +use uuid::Uuid; + +fn sid(n: u128) -> SessionId { + SessionId::from_uuid(Uuid::from_u128(n)) +} + +fn aid(n: u128) -> AgentId { + AgentId::from_uuid(Uuid::from_u128(n)) +} + +// --------------------------------------------------------------------------- +// Fake AgentSession scriptable (mono-usage), repris de send_blocking_d1.rs +// --------------------------------------------------------------------------- + +enum Script { + Stream(Vec), + Err(AgentSessionError), +} + +struct ScriptedSession { + id: SessionId, + script: Mutex>, + shutdowns: AtomicUsize, +} + +impl ScriptedSession { + fn new(script: Script) -> Self { + Self { + id: sid(1), + script: Mutex::new(Some(script)), + shutdowns: AtomicUsize::new(0), + } + } + fn shutdown_count(&self) -> usize { + self.shutdowns.load(Ordering::SeqCst) + } +} + +#[async_trait] +impl AgentSession for ScriptedSession { + fn id(&self) -> SessionId { + self.id + } + fn conversation_id(&self) -> Option { + None + } + async fn send(&self, _prompt: &str) -> Result { + let script = self + .script + .lock() + .unwrap() + .take() + .expect("send scripté une seule fois"); + match script { + Script::Stream(events) => Ok(Box::new(events.into_iter())), + Script::Err(e) => Err(e), + } + } + async fn shutdown(&self) -> Result<(), AgentSessionError> { + self.shutdowns.fetch_add(1, Ordering::SeqCst); + Ok(()) + } +} + +// --------------------------------------------------------------------------- +// Fake InputMediator : compte les mark_idle par agent + journalise les appels. +// --------------------------------------------------------------------------- + +struct RecordingMediator { + /// (agent, "mark_idle") dans l'ordre des appels. + calls: Mutex>, +} + +impl RecordingMediator { + fn new() -> Self { + Self { + calls: Mutex::new(Vec::new()), + } + } + fn mark_idle_count(&self, agent: AgentId) -> usize { + self.calls + .lock() + .unwrap() + .iter() + .filter(|(a, kind)| *a == agent && *kind == "mark_idle") + .count() + } + fn total_calls(&self) -> usize { + self.calls.lock().unwrap().len() + } +} + +impl InputMediator for RecordingMediator { + fn enqueue(&self, _agent: AgentId, _ticket: Ticket) -> PendingReply { + // Jamais utilisé par drain_with_readiness ; un future qui ne résout pas. + PendingReply::new(Box::pin(std::future::pending())) + } + fn preempt(&self, agent: AgentId) { + self.calls.lock().unwrap().push((agent, "preempt")); + } + fn mark_idle(&self, agent: AgentId) { + self.calls.lock().unwrap().push((agent, "mark_idle")); + } + fn busy_state(&self, _agent: AgentId) -> AgentBusyState { + AgentBusyState::Idle + } +} + +// --------------------------------------------------------------------------- +// Helpers d'événements +// --------------------------------------------------------------------------- + +fn delta(t: &str) -> ReplyEvent { + ReplyEvent::TextDelta { text: t.to_owned() } +} +fn tool(l: &str) -> ReplyEvent { + ReplyEvent::ToolActivity { + label: l.to_owned(), + } +} +fn heartbeat() -> ReplyEvent { + ReplyEvent::Heartbeat +} +fn final_(c: &str) -> ReplyEvent { + ReplyEvent::Final { + content: c.to_owned(), + } +} + +// =========================================================================== +// Point 6 (LE point qui compte) : un Final SEUL débloque la file via mark_idle, +// sans aucun idea_reply. +// =========================================================================== + +#[tokio::test] +async fn final_only_stream_unblocks_queue_via_mark_idle() { + // L'agent cible ne fait QUE renvoyer un Final (jamais d'idea_reply). + let agent = aid(42); + let session = ScriptedSession::new(Script::Stream(vec![final_("done")])); + let mediator = RecordingMediator::new(); + + let out = drain_with_readiness(&session, "tâche", None, &mediator, agent).await; + + assert_eq!(out, Ok("done".to_owned()), "le Final rend bien son contenu"); + assert_eq!( + mediator.mark_idle_count(agent), + 1, + "le Final déterministe DOIT marquer l'agent Idle (fix de la cause racine du blocage Busy)" + ); + // Aucun autre appel parasite, et la session reste vivante (pas de shutdown). + assert_eq!(mediator.total_calls(), 1); + assert_eq!(session.shutdown_count(), 0); +} + +// =========================================================================== +// Point 5 : exactement UN mark_idle ; les events intermédiaires n'en émettent pas. +// =========================================================================== + +#[tokio::test] +async fn intermediate_events_do_not_mark_idle_only_final_does() { + let agent = aid(7); + // Heartbeats + deltas + activité PUIS le Final. + let session = ScriptedSession::new(Script::Stream(vec![ + heartbeat(), + delta("hel"), + tool("lit un fichier"), + heartbeat(), + delta("lo"), + final_("hello"), + ])); + let mediator = RecordingMediator::new(); + + let out = drain_with_readiness(&session, "x", None, &mediator, agent).await; + + assert_eq!(out, Ok("hello".to_owned())); + assert_eq!( + mediator.mark_idle_count(agent), + 1, + "mark_idle exactement UNE fois (au Final), jamais sur heartbeat/delta/activité" + ); + assert_eq!( + mediator.total_calls(), + 1, + "aucun appel mediator hormis l'unique mark_idle" + ); +} + +#[tokio::test] +async fn heartbeats_alone_never_mark_idle_before_final() { + // Que des heartbeats avant le Final : aucun ne doit marquer Idle. + let agent = aid(9); + let session = ScriptedSession::new(Script::Stream(vec![ + heartbeat(), + heartbeat(), + heartbeat(), + final_("fini"), + ])); + let mediator = RecordingMediator::new(); + + let out = drain_with_readiness(&session, "x", None, &mediator, agent).await; + assert_eq!(out, Ok("fini".to_owned())); + assert_eq!(mediator.mark_idle_count(agent), 1); +} + +// =========================================================================== +// Bords : pas de Final ⇒ pas de mark_idle ; erreur de send propagée ; mauvais agent. +// =========================================================================== + +#[tokio::test] +async fn stream_without_final_does_not_mark_idle_and_is_io_error() { + let agent = aid(3); + let session = ScriptedSession::new(Script::Stream(vec![heartbeat(), delta("a"), tool("b")])); + let mediator = RecordingMediator::new(); + + let out = drain_with_readiness(&session, "x", None, &mediator, agent).await; + assert!( + matches!(out, Err(AgentSessionError::Io(_))), + "flux épuisé sans Final ⇒ Io, obtenu {out:?}" + ); + assert_eq!( + mediator.mark_idle_count(agent), + 0, + "sans Final, on ne marque JAMAIS Idle (jamais de faux Idle)" + ); + assert_eq!(mediator.total_calls(), 0); +} + +#[tokio::test] +async fn send_error_is_propagated_and_no_mark_idle() { + let agent = aid(5); + let session = + ScriptedSession::new(Script::Err(AgentSessionError::Decode("bad json".to_owned()))); + let mediator = RecordingMediator::new(); + + let out = drain_with_readiness(&session, "x", None, &mediator, agent).await; + assert_eq!(out, Err(AgentSessionError::Decode("bad json".to_owned()))); + assert_eq!(mediator.mark_idle_count(agent), 0); +} + +#[tokio::test] +async fn mark_idle_targets_the_drained_agent_only() { + // Le mark_idle doit porter sur l'agent passé à drain_with_readiness, pas un autre. + let drained = aid(100); + let other = aid(200); + let session = ScriptedSession::new(Script::Stream(vec![final_("ok")])); + let mediator = RecordingMediator::new(); + + let _ = drain_with_readiness(&session, "x", None, &mediator, drained).await; + assert_eq!(mediator.mark_idle_count(drained), 1); + assert_eq!( + mediator.mark_idle_count(other), + 0, + "mark_idle ne doit cibler que l'agent drainé" + ); +} + +// =========================================================================== +// Timeout (garde-fou du tour) : Timeout renvoyé, pas de mark_idle, session vivante. +// (Le helper borne via tokio::time::timeout ; on force l'expiration avec un Final +// reporté — réutilise un fake retardé minimal local.) +// =========================================================================== + +struct DelayedSession { + delay: Duration, + shutdowns: AtomicUsize, +} + +#[async_trait] +impl AgentSession for DelayedSession { + fn id(&self) -> SessionId { + sid(2) + } + fn conversation_id(&self) -> Option { + None + } + async fn send(&self, _prompt: &str) -> Result { + tokio::time::sleep(self.delay).await; + Ok(Box::new(vec![final_("too late")].into_iter())) + } + async fn shutdown(&self) -> Result<(), AgentSessionError> { + self.shutdowns.fetch_add(1, Ordering::SeqCst); + Ok(()) + } +} + +#[tokio::test] +async fn timeout_returns_timeout_no_mark_idle_session_alive() { + let agent = aid(11); + let session = DelayedSession { + delay: Duration::from_secs(1), + shutdowns: AtomicUsize::new(0), + }; + let mediator = RecordingMediator::new(); + + let out = + drain_with_readiness(&session, "x", Some(Duration::from_millis(20)), &mediator, agent).await; + assert_eq!(out, Err(AgentSessionError::Timeout)); + assert_eq!( + mediator.mark_idle_count(agent), + 0, + "un timeout ne marque pas Idle (le tour n'a pas rendu son Final)" + ); + assert_eq!( + session.shutdowns.load(Ordering::SeqCst), + 0, + "timeout ne tue PAS la session (§17.1)" + ); +} diff --git a/crates/application/tests/orchestrator_service.rs b/crates/application/tests/orchestrator_service.rs index c823d9e..f2b9ff6 100644 --- a/crates/application/tests/orchestrator_service.rs +++ b/crates/application/tests/orchestrator_service.rs @@ -2825,3 +2825,257 @@ async fn p6b_failing_record_does_not_degrade_ask() { "no turns stored when append fails" ); } + +// =========================================================================== +// Lot 1b — auto-lancement d'une cible **froide** structurée sur le chemin `ask` +// =========================================================================== +// +// Avant le Lot 1b, `ask_agent` ne routait vers `ask_structured` que si la cible avait +// **déjà** une session structurée vivante ; une cible structurée froide tombait dans +// `ensure_live_pty` (chemin PTY) — or une cible structurée n'a pas de PTY, donc le tour +// restait bloqué `Busy` (débloquable seulement par un `idea_reply` explicite). Le Lot 1b +// démarre la session structurée via le **même** launcher que le chemin PTY (qui route +// §17.4 vers `launch_structured` et l'insère dans le registre **partagé**), puis route +// vers `ask_structured` : le `Final` déterministe de la session débloque le tour et +// marque l'agent `Idle` (`mark_idle`), **sans** `idea_reply` ni PTY. + +use std::sync::atomic::{AtomicUsize, Ordering}; + +use application::StructuredSessions; +use domain::ports::{ + AgentSession, AgentSessionError, AgentSessionFactory, ReplyEvent, ReplyStream, +}; + +/// Session structurée fake : `send()` rend un unique [`ReplyEvent::Final`] qui **écho** +/// le prompt reçu — c'est ce `Final` que `drain_with_readiness` transforme en réponse +/// du tour **et** en `mark_idle`. Aucun `idea_reply` n'est nécessaire. +struct EchoSession { + id: SessionId, +} +#[async_trait] +impl AgentSession for EchoSession { + fn id(&self) -> SessionId { + self.id + } + fn conversation_id(&self) -> Option { + None + } + async fn send(&self, prompt: &str) -> Result { + let stream: ReplyStream = Box::new( + vec![ReplyEvent::Final { + content: format!("structured: {prompt}"), + }] + .into_iter(), + ); + Ok(stream) + } + async fn shutdown(&self) -> Result<(), AgentSessionError> { + Ok(()) + } +} + +/// Fabrique fake : `supports` = profil structuré, et `start` **compte** les démarrages +/// (preuve qu'une session froide a bien été auto-lancée) en rendant une [`EchoSession`]. +#[derive(Clone, Default)] +struct CountingFactory { + starts: Arc, + next_id: Arc>, +} +impl CountingFactory { + fn new() -> Self { + Self { + starts: Arc::new(AtomicUsize::new(0)), + next_id: Arc::new(Mutex::new(9000)), + } + } + fn start_count(&self) -> usize { + self.starts.load(Ordering::SeqCst) + } +} +#[async_trait] +impl AgentSessionFactory for CountingFactory { + fn supports(&self, profile: &AgentProfile) -> bool { + profile.structured_adapter.is_some() + } + async fn start( + &self, + _profile: &AgentProfile, + _ctx: &PreparedContext, + _cwd: &ProjectPath, + _session: &SessionPlan, + ) -> Result, AgentSessionError> { + self.starts.fetch_add(1, Ordering::SeqCst); + let id = { + let mut n = self.next_id.lock().unwrap(); + let id = SessionId::from_uuid(Uuid::from_u128(*n)); + *n += 1; + id + }; + Ok(Arc::new(EchoSession { id })) + } +} + +/// Tout le câblage `ask` **structuré** : launcher + service voient le **même** registre +/// `StructuredSessions`, et le launcher porte la fabrique (routage §17.4). +struct StructuredAskFixture { + service: Arc, + structured: Arc, + factory: CountingFactory, + pty: FakePty, +} + +fn structured_ask_fixture(contexts: FakeContexts) -> StructuredAskFixture { + let profiles = Arc::new(FakeProfiles::new(vec![claude_profile()])); + let sessions = Arc::new(TerminalSessions::new()); + let pty = FakePty::new(sid(777)); + let bus = SpyBus::default(); + let mailbox = Arc::new(TestMailbox::new()); + let structured = Arc::new(StructuredSessions::new()); + let factory = CountingFactory::new(); + + let create = Arc::new(CreateAgentFromScratch::new( + Arc::new(contexts.clone()), + Arc::new(SeqIds::new()), + Arc::new(bus.clone()), + )); + // Launcher câblé **structuré** : pour un profil à `structured_adapter`, `execute` + // route §17.4 → `launch_structured`, démarre la session via la fabrique et l'insère + // dans CE registre partagé (le même `Arc` que le service ci-dessous). + let launch = Arc::new( + LaunchAgent::new( + Arc::new(contexts.clone()), + Arc::clone(&profiles) as Arc, + Arc::new(FakeRuntime), + Arc::new(FakeFs), + Arc::new(pty.clone()), + Arc::new(FakeSkills), + Arc::clone(&sessions), + Arc::new(bus.clone()), + Arc::new(SeqIds::new()), + Arc::new(FakeRecall), + None, + ) + .with_structured( + Arc::new(factory.clone()) as Arc, + Arc::clone(&structured), + ), + ); + let list = Arc::new(ListAgents::new(Arc::new(contexts.clone()))); + let close = Arc::new(CloseTerminal::new( + Arc::new(pty.clone()), + Arc::clone(&sessions), + )); + let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts.clone()))); + let create_skill = Arc::new(CreateSkill::new( + Arc::new(RecordingSkills::default()) as Arc, + Arc::new(SeqIds::new()), + )); + let mediator = Arc::new(TestMediator::new( + Arc::clone(&mailbox), + Arc::new(pty.clone()) as Arc, + )); + let conversations = Arc::new(TestConversations::new()) as Arc; + + let service = Arc::new( + OrchestratorService::new( + create, + launch, + list, + close, + update, + create_skill, + Arc::clone(&profiles) as Arc, + Arc::clone(&sessions), + ) + .with_input_mediator( + Arc::clone(&mediator) as Arc, + Arc::clone(&mailbox) as Arc, + ) + .with_conversations(conversations) + .with_events(Arc::new(bus.clone())) + // Même registre que le launcher : `ask_agent` y relit la session fraîchement + // insérée par `launch_structured`. + .with_structured(Arc::clone(&structured)), + ); + + StructuredAskFixture { + service, + structured, + factory, + pty, + } +} + +/// Cible **froide** structurée (adaptateur Claude, aucune session vivante) : `ask_agent` +/// auto-lance sa session structurée via le launcher (factory.start appelé **une** fois), +/// puis la draine — le `Final` débloque le round-trip **sans** `idea_reply` ni PTY. +#[tokio::test] +async fn ask_cold_structured_target_autolaunches_session_and_final_unblocks() { + let agent = scratch_agent(aid(1), "architect", "agents/architect.md"); + let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona")); + + // Pré-condition : aucune session structurée vivante pour la cible (elle est froide). + assert!( + fx.structured.session_for_agent(&aid(1)).is_none(), + "cible structurée froide : aucune session avant l'ask" + ); + + let out = fx + .service + .dispatch(&project(), cmd(ASK_JSON)) + .await + .expect("ask ok"); + + // Le `Final` de la session a rendu le contenu — sans aucun `idea_reply` dispatché. + assert_eq!( + out.reply.as_deref(), + Some("structured: Analyse §17"), + "le Final de la session auto-lancée débloque le tour" + ); + // Exactement **un** démarrage de session structurée (la cible froide a été lancée). + assert_eq!( + fx.factory.start_count(), + 1, + "une session structurée a été auto-lancée pour la cible froide" + ); + // La session est désormais vivante dans le registre partagé (insérée par le launcher). + assert!( + fx.structured.session_for_agent(&aid(1)).is_some(), + "la session auto-lancée est enregistrée dans StructuredSessions" + ); + // Chemin structuré ⇒ **aucun** PTY spawné (la cible n'a pas de terminal). + assert!( + fx.pty.spawns().is_empty(), + "cible structurée : aucun PTY spawné (chemin chat, pas terminal)" + ); +} + +/// Réutilisation : une **deuxième** délégation vers la **même** cible (désormais chaude) +/// ne relance **pas** de session — elle réutilise celle insérée au premier `ask`. Prouve +/// que le `session_for_agent` court-circuite l'auto-lancement (idempotence du chemin). +#[tokio::test] +async fn ask_warm_structured_target_reuses_session_without_relaunch() { + let agent = scratch_agent(aid(1), "architect", "agents/architect.md"); + let fx = structured_ask_fixture(FakeContexts::with_agent(&agent, "# persona")); + + // 1er ask : auto-lance (start_count == 1). + let _ = fx + .service + .dispatch(&project(), cmd(ASK_JSON)) + .await + .expect("first ask ok"); + assert_eq!(fx.factory.start_count(), 1, "1er ask auto-lance la session"); + + // 2e ask : cible chaude ⇒ aucune relance. + let out = fx + .service + .dispatch(&project(), cmd(ASK_JSON)) + .await + .expect("second ask ok"); + assert_eq!(out.reply.as_deref(), Some("structured: Analyse §17")); + assert_eq!( + fx.factory.start_count(), + 1, + "cible chaude : la session est réutilisée, aucune relance" + ); +} diff --git a/crates/domain/src/events.rs b/crates/domain/src/events.rs index 1f9b053..215634e 100644 --- a/crates/domain/src/events.rs +++ b/crates/domain/src/events.rs @@ -71,6 +71,19 @@ pub enum DomainEvent { /// `true` when a turn is in flight, `false` when the agent is idle. busy: bool, }, + /// An agent's **liveness** (alive/stalled) changed (lot 2, chantier + /// readiness/heartbeat). Emitted **once per transition** by the stall detector: + /// `Alive→Stalled` when no proof of liveness arrived for longer than the profile's + /// `stall_after_ms`, and `Stalled→Alive` when a late battement (delta / tool + /// activity / heartbeat) revives it or the agent returns to `Idle`. Discrete, + /// low-frequency beacon (no spam) relayed to the front so the mediated-input view + /// can badge a frozen agent. Purely advisory — the FIFO and the turn keep running. + AgentLivenessChanged { + /// The agent whose liveness changed. + agent_id: AgentId, + /// The new liveness state. + liveness: crate::input::AgentLiveness, + }, /// An agent's runtime profile was changed (hot-swap of the AI engine). AgentProfileChanged { /// The agent. diff --git a/crates/domain/src/input.rs b/crates/domain/src/input.rs index 21b8897..b0bc982 100644 --- a/crates/domain/src/input.rs +++ b/crates/domain/src/input.rs @@ -80,6 +80,29 @@ pub enum AgentBusyState { }, } +/// Whether an agent currently shows **signs of life** during a turn (lot 2, +/// chantier readiness/heartbeat — détection « Stalled »). +/// +/// Orthogonal to [`AgentBusyState`]: an agent can be `Busy` **and** `Alive` (it is +/// working, producing deltas/heartbeats) or `Busy` **and** `Stalled` (no proof of +/// liveness for longer than its profile's `stall_after_ms`). Only meaningful while +/// `Busy`; an `Idle` agent is considered `Alive` (its turn ended cleanly). +/// +/// Published to the front (`AgentLivenessChanged`) **once per transition** so the UI +/// can badge a frozen agent without event spam. Derived from the per-agent +/// `last_seen_ms` heartbeat, never from parsing the model output. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", tag = "liveness")] +pub enum AgentLiveness { + /// The agent is producing proof of liveness (deltas / tool activity / + /// heartbeats) within its `stall_after_ms` window — or is idle. + Alive, + /// No proof of liveness for longer than the profile's `stall_after_ms`: the + /// agent is presumed frozen. Advisory only — the FIFO and the turn keep running + /// (a late heartbeat flips it back to [`Self::Alive`]). + Stalled, +} + impl AgentBusyState { /// Whether a turn is currently in flight. #[must_use] @@ -195,6 +218,29 @@ pub trait InputMediator: Send + Sync { /// Marks `agent` free (prompt-ready or explicit signal) so its FIFO advances. fn mark_idle(&self, agent: AgentId); + /// Records a **proof of liveness** (« battement ») for `agent` — called on every + /// non-terminal turn event (text delta, tool activity, [`crate::ports::ReplyEvent::Heartbeat`]) + /// by the drain loop. Refreshes the per-agent `last_seen` timestamp so the stall + /// detector ([`crate::events::DomainEvent::AgentLivenessChanged`], lot 2) knows the + /// agent is still working; a battement arriving after a `Stalled` transition flips it + /// back to [`AgentLiveness::Alive`]. + /// + /// Default: no-op (a mediator that does not track liveness). The infra adapter + /// `MediatedInbox` overrides it to refresh `last_seen` and publish an + /// `AgentLivenessChanged{Stalled→Alive}` recovery on the first late battement. + fn mark_alive(&self, _agent: AgentId) {} + + /// Declares the agent's **stall threshold** (its profile's + /// [`crate::profile::LivenessStrategy::stall_after_ms`]) so the stall detector knows + /// how long without a battement means "frozen" (lot 2). The orchestrator resolves it + /// from the target's profile and calls this **before** the enqueue that starts a + /// turn; the adapter stashes it and arms a fresh liveness window when the turn + /// begins. `None` ⇒ no stall detection for this agent (legacy / no `liveness` block — + /// zero regression). + /// + /// Default: no-op (a mediator that does not track liveness). + fn set_stall_threshold(&self, _agent: AgentId, _stall_after_ms: Option) {} + /// The current [`AgentBusyState`] of `agent`. fn busy_state(&self, agent: AgentId) -> AgentBusyState; } diff --git a/crates/domain/src/lib.rs b/crates/domain/src/lib.rs index 393672e..1c53cdf 100644 --- a/crates/domain/src/lib.rs +++ b/crates/domain/src/lib.rs @@ -47,6 +47,7 @@ pub mod orchestrator; pub mod ports; pub mod profile; pub mod project; +pub mod readiness; pub mod remote; pub mod skill; pub mod template; @@ -74,7 +75,8 @@ pub use skill::{Skill, SkillRef, SkillScope}; pub use template::{AgentTemplate, TemplateVersion}; pub use profile::{ - AgentProfile, ContextInjection, EmbedderProfile, EmbedderStrategy, SessionStrategy, + AgentProfile, ContextInjection, EmbedderProfile, EmbedderStrategy, LivenessStrategy, + McpServerWiring, SessionStrategy, }; pub use mailbox::{AgentMailbox, MailboxError, PendingReply, Ticket, TicketId}; @@ -84,7 +86,9 @@ pub use conversation::{ ConversationSession, SessionRef, WaitForGraph, }; -pub use input::{AgentBusyState, InputMediator, InputSource}; +pub use input::{AgentBusyState, AgentLiveness, InputMediator, InputSource}; + +pub use readiness::{ReadinessPolicy, ReadinessSignal}; pub use conversation_log::{ ConversationLog, ConversationTurn, Handoff, HandoffStore, HandoffSummarizer, diff --git a/crates/domain/src/ports.rs b/crates/domain/src/ports.rs index af49b1c..0f31670 100644 --- a/crates/domain/src/ports.rs +++ b/crates/domain/src/ports.rs @@ -209,6 +209,17 @@ pub enum ReplyEvent { /// Libellé humain-lisible de l'activité. label: String, }, + /// **Preuve de vivacité non terminale** (model-agnostique) : l'agent est + /// toujours en train de travailler. Émis par l'adapter quand le moteur signale + /// un battement de cœur natif **sans contenu utile** (handshake/init, fenêtre de + /// limite de débit, début/fin de tour côté moteur…). Sert à distinguer « agent + /// vivant mais lent » d'« agent bloqué » (readiness/heartbeat, lot 1). + /// + /// **Jamais terminal** : un `Heartbeat` ne clôt **pas** le flux — il s'intercale + /// comme un delta (ignoré par les consommateurs synchrones) et le flux continue + /// jusqu'au [`ReplyEvent::Final`]. Un tour comporte ≥0 `Heartbeat`, jamais + /// d'obligation d'en émettre. + Heartbeat, /// **Événement terminal déterministe** d'un tour : l'adapter l'émet quand il a /// lu le message `result` documenté de la CLI. Porte le contenu final agrégé. /// Après `Final`, le flux se termine (plus aucun événement). @@ -220,8 +231,10 @@ pub enum ReplyEvent { /// Flux borné d'événements de réponse d'UN tour (ARCHITECTURE §17.1). Se termine /// après le [`ReplyEvent::Final`] (ou sur erreur). Calqué sur [`OutputStream`], -/// mais **typé** : deltas de texte → activités d'outil → un `Final` déterministe, -/// plutôt que des octets bruts. +/// mais **typé** : deltas de texte → activités d'outil → battements de cœur* +/// ([`ReplyEvent::Heartbeat`], non terminaux, en nombre quelconque et entrelacés) → +/// **un** `Final` terminal déterministe, plutôt que des octets bruts. Seul le `Final` +/// clôt le flux ; deltas, activités et heartbeats sont tous non terminaux. pub type ReplyStream = Box + Send>; // --------------------------------------------------------------------------- diff --git a/crates/domain/src/profile.rs b/crates/domain/src/profile.rs index e886930..0c53d09 100644 --- a/crates/domain/src/profile.rs +++ b/crates/domain/src/profile.rs @@ -118,6 +118,61 @@ impl SessionStrategy { } } +/// Réglages de **vivacité** (readiness/heartbeat) d'un profil IA — place ménagée +/// pour le lot 2 (chantier readiness/heartbeat). Donnée **déclarative** (pas de code +/// par CLI — Open/Closed), calquée sur [`SessionStrategy`] / [`McpCapability`]. +/// +/// Deux seuils optionnels : +/// - `stall_after_ms` : délai sans **aucune** preuve de vivacité (delta / activité / +/// `ReplyEvent::Heartbeat`) au bout duquel l'agent est présumé **bloqué** +/// (`ReadinessSignal::Stalled`) — détection au lot 2 ; +/// - `turn_timeout_ms` : durée maximale d'**un tour** avant le garde-fou +/// (`ReadinessSignal::TimedOut`) — remplacement des timeouts en dur au lot 2. +/// +/// **Lot 1 : champs présents mais non consommés** — on ne fait que ménager la place +/// (le modèle de sérialisation et l'API sont figés ici pour éviter une migration au +/// lot 2). `None` (défaut, et valeur des profils existants) ⇒ comportement actuel. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LivenessStrategy { + /// Délai (ms) sans preuve de vivacité avant de présumer l'agent bloqué. + /// `None` ⇒ pas de détection de stagnation (lot 2). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub stall_after_ms: Option, + /// Durée maximale (ms) d'un tour avant le garde-fou de timeout. `None` ⇒ pas de + /// garde-fou par profil (lot 2). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub turn_timeout_ms: Option, +} + +impl LivenessStrategy { + /// Construit une stratégie de vivacité validée (parse-don't-validate, comme + /// [`SessionStrategy::new`]). + /// + /// # Errors + /// Renvoie [`DomainError::EmptyField`] si un seuil fourni vaut `0` (un seuil de + /// `0 ms` n'a pas de sens : `None` est la façon d'exprimer « pas de seuil »). + pub const fn new( + stall_after_ms: Option, + turn_timeout_ms: Option, + ) -> Result { + if let Some(0) = stall_after_ms { + return Err(DomainError::EmptyField { + field: "liveness.stallAfterMs", + }); + } + if let Some(0) = turn_timeout_ms { + return Err(DomainError::EmptyField { + field: "liveness.turnTimeoutMs", + }); + } + Ok(Self { + stall_after_ms, + turn_timeout_ms, + }) + } +} + /// Adapter d'**exécution structurée** qui pilote un profil IA (ARCHITECTURE §17). /// /// Déclaratif, Open/Closed (comme [`EmbedderStrategy`]) : un profil déclare quel @@ -193,6 +248,22 @@ pub enum McpConfigStrategy { /// Nom de la variable d'environnement. var: String, }, + /// Écrire un fichier de conf MCP **TOML** au chemin (relatif au run dir isolé + /// §14.1) attendu par la CLI, et pousser `home_env` (le nom d'une variable + /// d'environnement) vers le **dossier parent** de `target` pour isoler la CLI + /// de sa config globale. C'est le pendant Codex de [`Self::ConfigFile`] : Codex + /// lit ses serveurs MCP dans `$CODEX_HOME/config.toml` (défaut `~/.codex`), table + /// TOML `[mcp_servers.]`. IdeA écrit ce `config.toml` DANS le run dir de + /// l'agent et pointe `CODEX_HOME={runDir}/.codex` pour ne JAMAIS toucher au + /// `~/.codex` global (isolation par agent, miroir du `.mcp.json` de Claude). + TomlConfigHome { + /// Chemin relatif sûr du fichier `config.toml` (convention : + /// `".codex/config.toml"`). + target: String, + /// Nom de la variable d'environnement pointée sur le **dossier parent** de + /// `target` (ex. `"CODEX_HOME"`). + home_env: String, + }, } impl McpConfigStrategy { @@ -227,6 +298,24 @@ impl McpConfigStrategy { crate::validation::valid_env_var(&var)?; Ok(Self::Env { var }) } + + /// Constructeur validé `TomlConfigHome` (pendant Codex de [`Self::config_file`]). + /// + /// # Errors + /// - [`DomainError::PathNotRelativeSafe`] si `target` est absolu ou contient `..` + /// (même validation que [`Self::config_file`]), + /// - [`DomainError::InvalidEnvVar`] si `home_env` n'est pas un identifiant de + /// variable d'environnement valide. + pub fn toml_config_home( + target: impl Into, + home_env: impl Into, + ) -> Result { + let target = target.into(); + let home_env = home_env.into(); + crate::validation::relative_safe(&target)?; + crate::validation::valid_env_var(&home_env)?; + Ok(Self::TomlConfigHome { target, home_env }) + } } /// Capacité MCP d'un profil : COMMENT déclarer le serveur MCP IdeA à cette CLI, @@ -253,6 +342,128 @@ impl McpCapability { } } +/// Donnée de **wiring** du serveur MCP IdeA (`command` + `args` + `transport`), +/// factorisée pour servir de **source unique** aux deux sérialisations qui en +/// dérivaient séparément (et risquaient de diverger) : la déclaration `.mcp.json` +/// de Claude (JSON) et la table `[mcp_servers.idea]` de Codex (TOML). +/// +/// Pure (aucune I/O, aucune dépendance) : les deux encodeurs construisent une +/// chaîne à la main, donc le domaine reste sans dépendance (`serde_json`/`toml` +/// non requis). Le contenu (exe, endpoint, …) est calculé par l'appelant — le +/// domaine ne fait que la **mise en forme**. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct McpServerWiring { + /// Commande à lancer (le binaire IdeA en mode `mcp-server`, ou `"idea"` en + /// déclaration minimale dégradée). + pub command: String, + /// Arguments passés à `command` (ex. `["mcp-server", "--endpoint", …]`). + pub args: Vec, + /// Transport du serveur MCP (surfacé dans les deux formats). + pub transport: McpTransport, +} + +impl McpServerWiring { + /// Construit le wiring depuis ses parties. + #[must_use] + pub const fn new(command: String, args: Vec, transport: McpTransport) -> Self { + Self { + command, + args, + transport, + } + } + + /// Étiquette stable du transport, identique pour les deux formats. + #[must_use] + const fn transport_label(&self) -> &'static str { + match self.transport { + McpTransport::Stdio => "stdio", + McpTransport::Socket => "socket", + } + } + + /// Encode un document **`.mcp.json`** complet (Claude Code et CLIs apparentées) : + /// `{ "mcpServers": { "idea": { command, args, transport } } }`. Chaque chaîne est + /// échappée en littéral JSON (chemins avec espaces/backslash/quotes restent + /// valides). + #[must_use] + pub fn to_mcp_json(&self) -> String { + let command = json_string(&self.command); + let args = if self.args.is_empty() { + String::new() + } else { + let joined = self + .args + .iter() + .map(|a| format!("\n {}", json_string(a))) + .collect::>() + .join(","); + format!("{joined}\n ") + }; + let transport = self.transport_label(); + format!( + r#"{{ + "mcpServers": {{ + "idea": {{ + "command": {command}, + "args": [{args}], + "transport": "{transport}" + }} + }} +}} +"# + ) + } + + /// Encode la table **`[mcp_servers.idea]`** d'un `config.toml` Codex : + /// `command`, `args` (tableau TOML), `transport`. Chaque chaîne est échappée en + /// chaîne basique TOML (équivalent du `json_string` : espaces/backslash/quotes + /// restent valides). + #[must_use] + pub fn to_config_toml(&self) -> String { + let command = toml_string(&self.command); + let args = self + .args + .iter() + .map(|a| toml_string(a)) + .collect::>() + .join(", "); + let transport = self.transport_label(); + format!( + "[mcp_servers.idea]\ncommand = {command}\nargs = [{args}]\ntransport = \"{transport}\"\n" + ) + } +} + +/// Échappe `s` en **littéral de chaîne JSON** (guillemets inclus) pour les chemins +/// exe/endpoint avec espaces, backslash ou quotes. +fn json_string(s: &str) -> String { + let mut out = String::with_capacity(s.len() + 2); + out.push('"'); + for c in s.chars() { + match c { + '"' => out.push_str("\\\""), + '\\' => out.push_str("\\\\"), + '\n' => out.push_str("\\n"), + '\r' => out.push_str("\\r"), + '\t' => out.push_str("\\t"), + c if (c as u32) < 0x20 => out.push_str(&format!("\\u{:04x}", c as u32)), + c => out.push(c), + } + } + out.push('"'); + out +} + +/// Échappe `s` en **chaîne basique TOML** (guillemets inclus). Les chaînes +/// basiques TOML utilisent les mêmes séquences d'échappement que JSON pour `"`, +/// `\`, et les contrôles. +fn toml_string(s: &str) -> String { + // Le jeu d'échappement requis par une chaîne basique TOML coïncide avec celui de + // JSON pour les caractères qui nous concernent (chemins, flags). + json_string(s) +} + /// Declarative runtime configuration for one AI CLI. /// /// Invariants: @@ -316,6 +527,13 @@ pub struct AgentProfile { /// échappé présent dans la sortie. Un moteur regex pourra être ajouté plus tard /// comme variante déclarative (Open/Closed) si le besoin se confirme. /// + /// **Rang (chantier readiness/heartbeat, lot 1) : signal de repli n°3.** Depuis + /// l'introduction de la fin-de-tour structurée ([`crate::ports::ReplyEvent::Final`] + /// ⇒ [`crate::readiness::ReadinessSignal::TurnEnded`], signal n°1) et du signal + /// explicite `idea_reply` (n°2), ce sniff littéral est **rétrogradé** au rang de + /// repli : il ne sert plus que pour les agents **TUI/PTY sans adapter structuré**. + /// Conservé tel quel pour la rétro-compat (jamais supprimé). + /// /// `None` (défaut, et valeur des profils existants) ⇒ **aucune** détection par /// motif : l'agent ne repasse `Idle` que sur signal explicite (`idea_reply`) ou via /// le garde-fou du timeout par tour. Conforme au fallback « en cas de doute → reste @@ -325,6 +543,15 @@ pub struct AgentProfile { /// un profil sans motif sérialise exactement comme avant. #[serde(default, skip_serializing_if = "Option::is_none")] pub prompt_ready_pattern: Option, + /// Réglages de **vivacité** (readiness/heartbeat, chantier lot 1). `None` (défaut, + /// et valeur des profils existants) ⇒ comportement actuel. **Lot 1** : champ + /// présent mais **non consommé** (place ménagée pour les seuils de stagnation / + /// timeout de tour du lot 2). + /// + /// `skip_serializing_if = Option::is_none` ⇒ **zéro régression** de sérialisation : + /// un profil sans cette clé sérialise exactement comme avant. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub liveness: Option, /// Séquence de soumission écrite **après** le texte d'une délégation pour la /// faire valider par la CLI (§20.3, fix Bug 1). Le portail d'écriture (front) /// écrit d'abord le texte (sans `\n`, pour esquiver la détection de paste de @@ -483,6 +710,7 @@ impl AgentProfile { structured_adapter: None, mcp: None, prompt_ready_pattern: None, + liveness: None, submit_sequence: None, submit_delay_ms: None, }) @@ -515,6 +743,15 @@ impl AgentProfile { self } + /// Builder : fixe la [`LivenessStrategy`] (readiness/heartbeat, lot 1) et renvoie + /// le profil. Laisse [`AgentProfile::new`] stable (zéro régression d'appel) : les + /// profils sans réglage de vivacité ne l'appellent simplement pas. + #[must_use] + pub const fn with_liveness(mut self, liveness: LivenessStrategy) -> Self { + self.liveness = Some(liveness); + self + } + /// Builder : fixe la [`Self::submit_sequence`] (§20.3, fix Bug 1) et renvoie le /// profil. Laisse [`AgentProfile::new`] stable (zéro régression d'appel) : les /// profils qui s'en remettent au défaut `"\r"` ne l'appellent simplement pas. @@ -546,6 +783,33 @@ impl AgentProfile { pub fn is_selectable(&self) -> bool { self.structured_adapter.is_some() } + + /// **Source de vérité UNIQUE** de la whitelist des couples (adaptateur structuré + /// × stratégie MCP) qu'IdeA **matérialise réellement** pour exposer les outils + /// `idea_*` à la CLI — donc les seuls couples vers lesquels la délégation + /// inter-agents (`idea_ask_agent`/`idea_reply`) peut router une cible. + /// + /// Un profil déclare bien une [`McpConfigStrategy`], mais ce n'est honoré que si + /// IdeA sait écrire la config que **cette** CLI lit nativement : + /// - `Claude` + `ConfigFile { target == ".mcp.json" }` ⇒ Claude lit `.mcp.json` + /// dans son cwd (run dir isolé) ; + /// - `Codex` + `TomlConfigHome { .. }` ⇒ Codex lit `$CODEX_HOME/config.toml`, + /// qu'IdeA isole dans le run dir via `home_env` ; + /// - tout autre couple (y compris `mcp` absent) ⇒ `false` : repli fichier + /// `.ideai/requests` + prose, le pont natif n'est pas branché. + /// + /// Cette fonction centralise le critère pour que la garde applicative + /// ([`crate`] côté application) et la matérialisation ne puissent pas diverger. + #[must_use] + pub fn materializes_idea_bridge(&self) -> bool { + match (self.structured_adapter, self.mcp.as_ref().map(|c| &c.config)) { + (Some(StructuredAdapter::Claude), Some(McpConfigStrategy::ConfigFile { target })) => { + target == ".mcp.json" + } + (Some(StructuredAdapter::Codex), Some(McpConfigStrategy::TomlConfigHome { .. })) => true, + _ => false, + } + } } #[cfg(test)] @@ -832,4 +1096,200 @@ mod mcp_tests { assert_eq!(back.submit_sequence.as_deref(), Some("\r")); assert_eq!(back.submit_delay_ms, Some(60)); } + + // -- Lot 1 : liveness (readiness/heartbeat) — non-régression de sérialisation -- + + #[test] + fn profile_default_has_no_liveness() { + // Profils existants (via `new`) : aucun réglage de vivacité. + assert!(profile_without_mcp().liveness.is_none()); + } + + #[test] + fn profile_without_liveness_omits_key_in_json() { + let json = serde_json::to_string(&profile_without_mcp()).expect("serialise"); + assert!( + !json.contains("liveness"), + "a profile without liveness must NOT serialise the key (zero regression); got: {json}" + ); + } + + #[test] + fn legacy_json_without_liveness_deserialises_to_none() { + let legacy = r#"{ + "id": "00000000-0000-0000-0000-000000000000", + "name": "Dev", + "command": "claude", + "args": [], + "contextInjection": { "strategy": "conventionFile", "target": "CLAUDE.md" }, + "detect": null, + "cwdTemplate": "{agentRunDir}" + }"#; + let profile: AgentProfile = serde_json::from_str(legacy).expect("legacy deserialise"); + assert!(profile.liveness.is_none()); + } + + #[test] + fn with_liveness_sets_and_round_trips_camel_case() { + let liveness = LivenessStrategy::new(Some(30_000), Some(600_000)).expect("valid liveness"); + let profile = profile_without_mcp().with_liveness(liveness); + assert_eq!(profile.liveness, Some(liveness)); + + let json = serde_json::to_string(&profile).expect("serialise"); + assert!(json.contains("liveness"), "key present: {json}"); + assert!(json.contains("stallAfterMs"), "camelCase field: {json}"); + assert!(json.contains("turnTimeoutMs"), "camelCase field: {json}"); + + let back: AgentProfile = serde_json::from_str(&json).expect("deserialise"); + assert_eq!(profile, back); + assert_eq!(back.liveness, Some(liveness)); + } + + #[test] + fn liveness_omits_unset_thresholds_in_json() { + // Un seul seuil fixé : l'autre est `None` ⇒ sa clé est omise. + let liveness = LivenessStrategy::new(None, Some(600_000)).expect("valid liveness"); + let json = serde_json::to_string(&liveness).expect("serialise"); + assert!( + !json.contains("stallAfterMs"), + "an unset stall threshold must be omitted; got: {json}" + ); + assert!(json.contains("turnTimeoutMs"), "set threshold present: {json}"); + } + + #[test] + fn liveness_new_rejects_zero_thresholds() { + assert!(matches!( + LivenessStrategy::new(Some(0), None).unwrap_err(), + DomainError::EmptyField { .. } + )); + assert!(matches!( + LivenessStrategy::new(None, Some(0)).unwrap_err(), + DomainError::EmptyField { .. } + )); + // Les deux None : valide (= « aucun seuil »). + assert!(LivenessStrategy::new(None, None).is_ok()); + } + + // -- Codex : surface MCP `TomlConfigHome` (pont inter-agents Codex) ---------- + + #[test] + fn toml_config_home_round_trips_with_tagged_strategy() { + let strategy = McpConfigStrategy::toml_config_home(".codex/config.toml", "CODEX_HOME") + .expect("valid toml config home"); + let json = serde_json::to_string(&strategy).expect("serialise"); + + // Wire contract: tagged enum (`strategy` tag) + camelCase variant & fields. + assert!( + json.contains("\"strategy\":\"tomlConfigHome\""), + "tagged camelCase variant expected; got: {json}" + ); + assert!( + json.contains("\"target\":\".codex/config.toml\""), + "target field expected; got: {json}" + ); + // NB: `rename_all = "camelCase"` on this enum renames *variants*, not the + // fields of a struct variant, so `home_env` stays snake_case on the wire. + assert!( + json.contains("\"home_env\":\"CODEX_HOME\""), + "home_env field expected; got: {json}" + ); + + let back: McpConfigStrategy = serde_json::from_str(&json).expect("deserialise"); + assert_eq!(strategy, back); + } + + #[test] + fn toml_config_home_rejects_absolute_and_parent_target() { + let abs = McpConfigStrategy::toml_config_home("/abs/x", "CODEX_HOME").unwrap_err(); + assert!(matches!(abs, DomainError::PathNotRelativeSafe { .. })); + + let parent = McpConfigStrategy::toml_config_home("../escape", "CODEX_HOME").unwrap_err(); + assert!(matches!(parent, DomainError::PathNotRelativeSafe { .. })); + } + + #[test] + fn toml_config_home_rejects_invalid_home_env() { + // Empty home_env: first char is None ⇒ invalid identifier. + let empty = McpConfigStrategy::toml_config_home(".codex/config.toml", "").unwrap_err(); + assert!(matches!(empty, DomainError::InvalidEnvVar { .. })); + + // Illegal character in the env var name. + let illegal = + McpConfigStrategy::toml_config_home(".codex/config.toml", "BAD-NAME").unwrap_err(); + assert!(matches!(illegal, DomainError::InvalidEnvVar { .. })); + } + + #[test] + fn materializes_idea_bridge_matrix() { + // Claude + `.mcp.json` ConfigFile ⇒ bridge materialised. + let claude = profile_without_mcp() + .with_structured_adapter(StructuredAdapter::Claude) + .with_mcp(McpCapability::new( + McpConfigStrategy::config_file(".mcp.json").expect("valid target"), + McpTransport::Stdio, + )); + assert!(claude.materializes_idea_bridge()); + + // Codex + TomlConfigHome ⇒ bridge materialised (the key point: Codex used to + // be refused before this surface existed). + let codex = profile_without_mcp() + .with_structured_adapter(StructuredAdapter::Codex) + .with_mcp(McpCapability::new( + McpConfigStrategy::toml_config_home(".codex/config.toml", "CODEX_HOME") + .expect("valid toml config home"), + McpTransport::Stdio, + )); + assert!(codex.materializes_idea_bridge()); + + // Codex WITHOUT any MCP capability ⇒ no bridge. + let codex_no_mcp = + profile_without_mcp().with_structured_adapter(StructuredAdapter::Codex); + assert!(!codex_no_mcp.materializes_idea_bridge()); + + // Codex + wrong strategy (`.mcp.json` ConfigFile, Claude's shape) ⇒ no bridge. + let codex_wrong_strategy = profile_without_mcp() + .with_structured_adapter(StructuredAdapter::Codex) + .with_mcp(McpCapability::new( + McpConfigStrategy::config_file(".mcp.json").expect("valid target"), + McpTransport::Stdio, + )); + assert!(!codex_wrong_strategy.materializes_idea_bridge()); + } + + #[test] + fn mcp_server_wiring_encodes_expected_toml() { + // A command path with a space and a backslash exercises TOML escaping. + let wiring = McpServerWiring::new( + "/opt/My Apps\\idea".to_owned(), + vec![ + "mcp-server".to_owned(), + "--endpoint".to_owned(), + "/tmp/sock 1".to_owned(), + ], + McpTransport::Stdio, + ); + let toml = wiring.to_config_toml(); + + // Table header present. + assert!( + toml.contains("[mcp_servers.idea]"), + "table header expected; got: {toml}" + ); + // Command path escaped as a TOML basic string (backslash doubled, space kept). + assert!( + toml.contains("command = \"/opt/My Apps\\\\idea\""), + "escaped command path expected; got: {toml}" + ); + // Args preserved in order, as a TOML array. + assert!( + toml.contains("args = [\"mcp-server\", \"--endpoint\", \"/tmp/sock 1\"]"), + "args in order expected; got: {toml}" + ); + // Transport surfaced. + assert!( + toml.contains("transport = \"stdio\""), + "transport expected; got: {toml}" + ); + } } diff --git a/crates/domain/src/readiness.rs b/crates/domain/src/readiness.rs new file mode 100644 index 0000000..71b6559 --- /dev/null +++ b/crates/domain/src/readiness.rs @@ -0,0 +1,111 @@ +//! Politique de **readiness** (« fin-de-tour ») model-agnostique (chantier +//! readiness/heartbeat, lot 1). +//! +//! Objet **pur** (aucune I/O, aucune dépendance externe) qui classe un signal +//! observable d'un tour d'agent en un [`ReadinessSignal`] normalisé. Le but : que +//! l'application puisse décider de marquer un agent `Idle` +//! ([`crate::input::InputMediator::mark_idle`]) sur un **signal déterministe** +//! (`Final` du flux structuré) plutôt que de dépendre uniquement d'un `idea_reply` +//! explicite ou d'un sniff littéral de prompt PTY. +//! +//! # Hiérarchie des signaux de fin-de-tour (rappel cadrage) +//! +//! 1. **Signal n°1 — fin de tour structurée** : [`ReplyEvent::Final`] émis par +//! l'adapter (Claude `type:"result"`, Codex `agent_message`/`item.completed`). +//! Déterministe, model-agnostique ⇒ classé [`ReadinessSignal::TurnEnded`]. +//! 2. **Signal n°2 — `idea_reply` explicite** : l'agent appelle l'outil MCP +//! [`crate::ports`]/délégation. Premier arrivé gagne avec le n°1. +//! 3. **Signal n°3 — repli `prompt_ready_pattern`** : sniff littéral du sigil de +//! prompt dans la sortie PTY ([`crate::profile::AgentProfile::prompt_ready_pattern`]). +//! **Rétrogradé** au rang de repli depuis ce lot : il ne sert que pour les agents +//! TUI/PTY sans adapter structuré (rétro-compat, jamais supprimé). +//! +//! Les variantes [`ReadinessSignal::Stalled`]/[`ReadinessSignal::TimedOut`] sont la +//! place réservée au **lot 2** (détection de stagnation, remplacement des timeouts) : +//! elles existent dans le vocabulaire mais ne sont **pas** produites par +//! [`ReadinessPolicy::classify`] dans ce lot. + +use crate::ports::ReplyEvent; + +/// Signal de readiness normalisé, model-agnostique, qu'une [`ReadinessPolicy`] +/// déduit d'un événement observable du tour. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReadinessSignal { + /// Le tour est **déterministiquement terminé** : l'agent a rendu son `Final`. + /// C'est le signal n°1, model-agnostique — il doit réveiller le `pending` et + /// marquer l'agent `Idle`. + TurnEnded, + /// Un `idea_reply` explicite a été observé (signal n°2). N'est **pas** produit + /// par [`ReadinessPolicy::classify`] (qui ne voit que des [`ReplyEvent`]) : il + /// est porté par le chemin de délégation, présent ici pour compléter le + /// vocabulaire et le rendre explicite. + ExplicitReply, + /// Le sigil de prompt PTY (repli n°3) est apparu. Idem : non produit par + /// `classify`, présent pour nommer le signal de repli legacy. + PromptReady, + /// L'agent semble **bloqué** (aucune preuve de vivacité depuis un seuil). Place + /// réservée au **lot 2** — non produit dans ce lot. + Stalled, + /// Le garde-fou de durée de tour a expiré. Place réservée au **lot 2** — non + /// produit dans ce lot. + TimedOut, +} + +/// Politique **pure** de classification d'un événement de tour en +/// [`ReadinessSignal`]. Sans état, sans I/O : un simple `match` sur le contrat de +/// port universel [`ReplyEvent`], pour que la décision « ce tour est-il fini ? » +/// vive dans le **domaine** et reste testable sans process ni réseau. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct ReadinessPolicy; + +impl ReadinessPolicy { + /// Classe un [`ReplyEvent`] en signal de readiness. + /// + /// - [`ReplyEvent::Final`] ⇒ `Some(`[`ReadinessSignal::TurnEnded`]`)` : seul + /// événement terminal, il signe la fin de tour déterministe. + /// - [`ReplyEvent::TextDelta`] / [`ReplyEvent::ToolActivity`] / + /// [`ReplyEvent::Heartbeat`] ⇒ `None` : tous **non terminaux** (le flux + /// continue). Un heartbeat prouve la vivacité mais ne termine pas le tour. + #[must_use] + pub const fn classify(event: &ReplyEvent) -> Option { + match event { + ReplyEvent::Final { .. } => Some(ReadinessSignal::TurnEnded), + ReplyEvent::TextDelta { .. } + | ReplyEvent::ToolActivity { .. } + | ReplyEvent::Heartbeat => None, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn final_classifies_as_turn_ended() { + let ev = ReplyEvent::Final { + content: "fini".to_owned(), + }; + assert_eq!( + ReadinessPolicy::classify(&ev), + Some(ReadinessSignal::TurnEnded) + ); + } + + #[test] + fn deltas_activities_and_heartbeats_are_non_terminal() { + assert_eq!( + ReadinessPolicy::classify(&ReplyEvent::TextDelta { text: "x".into() }), + None + ); + assert_eq!( + ReadinessPolicy::classify(&ReplyEvent::ToolActivity { label: "lit".into() }), + None + ); + assert_eq!( + ReadinessPolicy::classify(&ReplyEvent::Heartbeat), + None, + "un heartbeat prouve la vivacité mais ne termine JAMAIS le tour" + ); + } +} diff --git a/crates/infrastructure/src/input/mod.rs b/crates/infrastructure/src/input/mod.rs index 205eb0c..e16bc2c 100644 --- a/crates/infrastructure/src/input/mod.rs +++ b/crates/infrastructure/src/input/mod.rs @@ -25,7 +25,7 @@ use std::sync::{Arc, Mutex}; use domain::events::DomainEvent; use domain::ids::AgentId; -use domain::input::{AgentBusyState, InputMediator, SubmitConfig}; +use domain::input::{AgentBusyState, AgentLiveness, InputMediator, SubmitConfig}; use domain::mailbox::{AgentMailbox, PendingReply, Ticket, TicketId}; use domain::ports::{EventBus, PtyHandle, PtyPort}; @@ -40,13 +40,34 @@ use crate::mailbox::InMemoryMailbox; /// every path (explicit `mark_idle`, prompt-ready match) stays consistent. struct BusyTracker { busy: Mutex>, + /// Per-agent **liveness** bookkeeping (lot 2) : dernier battement observé, seuil de + /// stagnation issu du profil, et état de vivacité courant pour n'émettre + /// `AgentLivenessChanged` qu'**une fois par transition** (pas de spam). + liveness: Mutex>, events: Option>, } +/// État de vivacité d'un agent maintenu par le [`BusyTracker`] (lot 2). +/// +/// Pur (aucune I/O) : un timestamp `last_seen_ms` rafraîchi à chaque battement, le +/// seuil `stall_after_ms` du profil (cf. [`domain::profile::LivenessStrategy`]) et +/// l'état courant `current` pour décider d'une **transition** (et donc d'un seul +/// événement). `stall_after_ms = None` ⇒ pas de détection (comportement legacy). +#[derive(Debug, Clone, Copy)] +struct LivenessState { + /// Epoch-millis du dernier battement (ou du démarrage du tour). + last_seen_ms: u64, + /// Seuil de stagnation du profil de l'agent. `None` ⇒ détection désactivée. + stall_after_ms: Option, + /// État de vivacité courant (pour n'émettre qu'à la transition). + current: AgentLiveness, +} + impl BusyTracker { fn new(events: Option>) -> Self { Self { busy: Mutex::new(HashMap::new()), + liveness: Mutex::new(HashMap::new()), events, } } @@ -57,6 +78,100 @@ impl BusyTracker { .unwrap_or_else(std::sync::PoisonError::into_inner) } + fn lock_liveness(&self) -> std::sync::MutexGuard<'_, HashMap> { + self.liveness + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + /// Publie un `AgentLivenessChanged` (si un bus est câblé). + fn publish_liveness(&self, agent: AgentId, liveness: AgentLiveness) { + if let Some(events) = &self.events { + events.publish(DomainEvent::AgentLivenessChanged { + agent_id: agent, + liveness, + }); + } + } + + /// Enregistre le seuil de stagnation du profil d'un agent au moment où son tour + /// démarre (depuis l'enqueue) : (re)initialise `last_seen` à `now` et repart d'un + /// état `Alive`. Sans seuil (`None`), l'entrée existe quand même mais le sweep ne + /// la déclarera jamais `Stalled` (zéro régression pour un profil sans liveness). + fn arm_liveness(&self, agent: AgentId, stall_after_ms: Option, now_ms: u64) { + self.lock_liveness().insert( + agent, + LivenessState { + last_seen_ms: now_ms, + stall_after_ms, + current: AgentLiveness::Alive, + }, + ); + } + + /// Rafraîchit le `last_seen` d'un agent (un **battement**) et, s'il était + /// `Stalled`, le ramène à `Alive` en émettant l'unique transition de reprise. No-op + /// si l'agent n'a pas d'entrée de vivacité armée (tour non structuré / legacy). + fn touch(&self, agent: AgentId, now_ms: u64) { + let recovered = { + let mut map = self.lock_liveness(); + let Some(state) = map.get_mut(&agent) else { + return; + }; + state.last_seen_ms = now_ms; + if state.current == AgentLiveness::Stalled { + state.current = AgentLiveness::Alive; + true + } else { + false + } + }; + if recovered { + self.publish_liveness(agent, AgentLiveness::Alive); + } + } + + /// Balaye tous les agents et, pour chacun dont le tour stagne + /// (`now - last_seen > stall_after_ms` et seuil défini), bascule `Alive→Stalled` + /// en émettant **une seule** transition. **Pure et testable sans horloge réelle** : + /// `now_ms` est passé en paramètre. Idempotente : un agent déjà `Stalled` ne ré-émet + /// pas. Un agent sans seuil (`None`) ou redevenu `Idle` n'est jamais déclaré stalled. + fn sweep_stalled(&self, now_ms: u64) { + let newly_stalled: Vec = { + let mut map = self.lock_liveness(); + map.iter_mut() + .filter_map(|(agent, state)| { + let threshold = state.stall_after_ms?; + if state.current != AgentLiveness::Alive { + return None; // déjà Stalled : pas de ré-émission. + } + let elapsed = now_ms.saturating_sub(state.last_seen_ms); + if elapsed > u64::from(threshold) { + state.current = AgentLiveness::Stalled; + Some(*agent) + } else { + None + } + }) + .collect() + }; + for agent in newly_stalled { + self.publish_liveness(agent, AgentLiveness::Stalled); + } + } + + /// Retire l'entrée de vivacité d'un agent (fin de tour). Si l'agent était `Stalled`, + /// la fin de tour est en soi un retour à `Alive` ⇒ on émet la transition de reprise. + fn clear_liveness(&self, agent: AgentId) { + let was_stalled = self + .lock_liveness() + .remove(&agent) + .is_some_and(|s| s.current == AgentLiveness::Stalled); + if was_stalled { + self.publish_liveness(agent, AgentLiveness::Alive); + } + } + fn busy_state(&self, agent: AgentId) -> AgentBusyState { self.lock() .get(&agent) @@ -94,6 +209,9 @@ impl BusyTracker { }); } } + // Fin de tour : retirer l'entrée de vivacité (émet la reprise si l'agent + // était `Stalled`). Indépendant de `was_busy` pour rester idempotent. + self.clear_liveness(agent); } } @@ -143,6 +261,10 @@ pub struct MediatedInbox { /// stashed at bind time (§20.3) and echoed on the `DelegationReady` event when a /// turn starts. Absent ⇒ both `None` (the front applies its defaults). submit: Mutex>, + /// Per-agent stall threshold (`LivenessStrategy::stall_after_ms`, lot 2), stashed by + /// `set_stall_threshold` and consumed to arm a fresh liveness window on the enqueue + /// that starts a turn. Absent ⇒ `None` (no stall detection — legacy behaviour). + stall: Mutex>>, /// Agents whose prompt-ready watcher thread is already armed, so re-binding the same /// handle does not spawn a duplicate watcher (lot C5). Shared (`Arc`) because each /// watcher thread un-arms its own entry on exit. @@ -161,6 +283,7 @@ impl MediatedInbox { pty: None, handles: Mutex::new(HashMap::new()), submit: Mutex::new(HashMap::new()), + stall: Mutex::new(HashMap::new()), watched: Arc::new(Mutex::new(HashSet::new())), } } @@ -190,6 +313,7 @@ impl MediatedInbox { pty: Some(pty), handles: Mutex::new(HashMap::new()), submit: Mutex::new(HashMap::new()), + stall: Mutex::new(HashMap::new()), watched: Arc::new(Mutex::new(HashSet::new())), } } @@ -215,6 +339,22 @@ impl MediatedInbox { .unwrap_or_else(std::sync::PoisonError::into_inner) } + fn stall(&self) -> std::sync::MutexGuard<'_, HashMap>> { + self.stall + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + /// **Détection de stagnation** (lot 2) : balaie les agents `Busy` et bascule + /// `Alive→Stalled` ceux dont le dernier battement remonte à plus de + /// `stall_after_ms`. Émet `AgentLivenessChanged` **une fois par transition**. La + /// logique de décision est **pure** ([`BusyTracker::sweep_stalled`]) : `now_ms` est + /// fourni par l'horloge injectée, donc testable sans horloge réelle. Destinée à + /// être appelée périodiquement par une tâche détenue au composition root. + pub fn sweep_stalled(&self) { + self.tracker.sweep_stalled(self.clock.now_ms()); + } + /// The underlying mailbox (e.g. for `cancel_head` / `resolve` from the orchestrator). #[must_use] pub fn mailbox(&self) -> Arc { @@ -305,13 +445,22 @@ impl InputMediator for MediatedInbox { // If the agent is Idle, this enqueue starts its turn ⇒ go Busy. If already // Busy, we still accept (queue grows; the turn advances on mark_idle) — never // reject the sender (forward fallback). + let now_ms = self.clock.now_ms(); let started_turn = self.tracker.start_turn( agent, AgentBusyState::Busy { ticket: ticket_id, - since_ms: self.clock.now_ms(), + since_ms: now_ms, }, ); + // Sur l'enqueue qui **démarre** un tour : armer une fenêtre de vivacité fraîche + // avec le seuil de stagnation stashé du profil (lot 2). `last_seen = now`, état + // `Alive`. Un profil sans seuil (`None`) arme une entrée jamais déclarée stalled + // (zéro régression). Un re-enqueue pendant `Busy` ne ré-arme pas (le tour court). + if started_turn { + let stall_after_ms = self.stall().get(&agent).copied().flatten(); + self.tracker.arm_liveness(agent, stall_after_ms, now_ms); + } // Publish only on the enqueue that **starts** a turn (Idle→Busy); a second // enqueue while Busy queues behind without re-announcing (cadrage C4 §4.2, // ARCHITECTURE §20). Published outside the busy mutex (the tracker released it @@ -391,10 +540,23 @@ impl InputMediator for MediatedInbox { fn mark_idle(&self, agent: AgentId) { // Single authority (also used by the prompt-ready watcher): real Busy→Idle only, - // publishing AgentBusyChanged{busy:false} once. + // publishing AgentBusyChanged{busy:false} once. Also clears the liveness entry + // (fin de tour) — émet la reprise si l'agent était `Stalled`. self.tracker.mark_idle(agent); } + fn mark_alive(&self, agent: AgentId) { + // Un battement (delta / activité / heartbeat) rafraîchit `last_seen` et ramène + // l'agent à `Alive` s'il était `Stalled` (lot 2). + self.tracker.touch(agent, self.clock.now_ms()); + } + + fn set_stall_threshold(&self, agent: AgentId, stall_after_ms: Option) { + // Stashé par l'orchestrateur depuis le profil de la cible AVANT l'enqueue qui + // démarre le tour ; consommé par `arm_liveness` au start_turn (lot 2). + self.stall().insert(agent, stall_after_ms); + } + fn busy_state(&self, agent: AgentId) -> AgentBusyState { self.tracker.busy_state(agent) } @@ -985,4 +1147,205 @@ mod tests { std::thread::sleep(std::time::Duration::from_millis(60)); assert!(inbox.busy_state(a).is_busy(), "no match ⇒ stays Busy"); } + + // ==================================================================== + // Lot 2 — détection de stagnation (last_seen / sweep_stalled / liveness) + // ==================================================================== + + use std::sync::atomic::{AtomicU64, Ordering}; + use domain::input::AgentLiveness; + + /// Horloge millis **mutable** (atomique) pour piloter le temps dans les tests de + /// stagnation sans horloge réelle. + struct MutClock(AtomicU64); + impl MutClock { + fn new(start: u64) -> Arc { + Arc::new(Self(AtomicU64::new(start))) + } + fn set(&self, now: u64) { + self.0.store(now, Ordering::SeqCst); + } + } + impl MillisClock for MutClock { + fn now_ms(&self) -> u64 { + self.0.load(Ordering::SeqCst) + } + } + + /// Construit un inbox sur l'horloge mutable + un bus enregistreur, renvoie les deux. + fn inbox_with_clock(clock: Arc) -> (MediatedInbox, Arc) { + let bus = Arc::new(RecordingBus::default()); + let inbox = MediatedInbox::new(Arc::new(InMemoryMailbox::new()), clock) + .with_events(Arc::clone(&bus) as Arc); + (inbox, bus) + } + + impl RecordingBus { + /// Toutes les transitions de vivacité publiées, dans l'ordre. + fn liveness_events(&self) -> Vec<(AgentId, AgentLiveness)> { + self.0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .iter() + .filter_map(|e| match e { + DomainEvent::AgentLivenessChanged { agent_id, liveness } => { + Some((*agent_id, *liveness)) + } + _ => None, + }) + .collect() + } + } + + // (a) absence de battement > stall_after_ms ⇒ Stalled + #[test] + fn no_heartbeat_past_threshold_marks_stalled() { + let clock = MutClock::new(1_000); + let (inbox, bus) = inbox_with_clock(Arc::clone(&clock)); + let a = agent(1); + + // Profil : seuil de stagnation à 30_000 ms. Armé avant le tour. + inbox.set_stall_threshold(a, Some(30_000)); + inbox.enqueue(a, ticket(10, "task")); // démarre le tour à t=1_000 ⇒ last_seen=1_000 + + // Dans la fenêtre : pas de transition. + clock.set(1_000 + 30_000); + inbox.sweep_stalled(); + assert_eq!(bus.liveness_events(), vec![], "à la limite exacte, pas encore stalled"); + + // Au-delà du seuil : exactement une transition Stalled. + clock.set(1_000 + 30_001); + inbox.sweep_stalled(); + assert_eq!(bus.liveness_events(), vec![(a, AgentLiveness::Stalled)]); + + // Idempotent : un second sweep ne ré-émet pas. + clock.set(1_000 + 60_000); + inbox.sweep_stalled(); + assert_eq!( + bus.liveness_events(), + vec![(a, AgentLiveness::Stalled)], + "déjà stalled ⇒ pas de spam" + ); + } + + // (b) un battement dans la fenêtre réinitialise last_seen (pas de stall) + #[test] + fn heartbeat_within_window_resets_last_seen() { + let clock = MutClock::new(1_000); + let (inbox, bus) = inbox_with_clock(Arc::clone(&clock)); + let a = agent(1); + inbox.set_stall_threshold(a, Some(30_000)); + inbox.enqueue(a, ticket(10, "task")); // last_seen=1_000 + + // Battement à t=20_000 ⇒ last_seen=20_000. + clock.set(20_000); + inbox.mark_alive(a); + + // À t=45_000 : 45_000 - 20_000 = 25_000 < 30_000 ⇒ toujours Alive. + clock.set(45_000); + inbox.sweep_stalled(); + assert_eq!( + bus.liveness_events(), + vec![], + "un battement a réinitialisé la fenêtre ⇒ pas de stall" + ); + } + + // (d) AgentLivenessChanged émis une seule fois par transition (Alive→Stalled→Alive) + #[test] + fn liveness_event_once_per_transition() { + let clock = MutClock::new(0); + let (inbox, bus) = inbox_with_clock(Arc::clone(&clock)); + let a = agent(1); + inbox.set_stall_threshold(a, Some(10_000)); + inbox.enqueue(a, ticket(10, "t")); // last_seen=0 + + // Stall. + clock.set(10_001); + inbox.sweep_stalled(); + // Sweeps répétés : pas de ré-émission. + inbox.sweep_stalled(); + inbox.sweep_stalled(); + assert_eq!(bus.liveness_events(), vec![(a, AgentLiveness::Stalled)]); + + // Battement tardif ⇒ une seule reprise Alive. + clock.set(20_000); + inbox.mark_alive(a); + inbox.mark_alive(a); // second battement : déjà Alive ⇒ rien. + assert_eq!( + bus.liveness_events(), + vec![(a, AgentLiveness::Stalled), (a, AgentLiveness::Alive)] + ); + + // Re-stall possible après reprise (nouvelle transition). + clock.set(20_000 + 10_001); + inbox.sweep_stalled(); + assert_eq!( + bus.liveness_events(), + vec![ + (a, AgentLiveness::Stalled), + (a, AgentLiveness::Alive), + (a, AgentLiveness::Stalled), + ] + ); + } + + // mark_idle (fin de tour) sur un agent Stalled émet la reprise et retire l'entrée. + #[test] + fn mark_idle_on_stalled_emits_recovery_and_clears() { + let clock = MutClock::new(0); + let (inbox, bus) = inbox_with_clock(Arc::clone(&clock)); + let a = agent(1); + inbox.set_stall_threshold(a, Some(5_000)); + inbox.enqueue(a, ticket(10, "t")); + clock.set(5_001); + inbox.sweep_stalled(); + assert_eq!(bus.liveness_events(), vec![(a, AgentLiveness::Stalled)]); + + inbox.mark_idle(a); // fin de tour ⇒ reprise Alive + entrée retirée. + assert_eq!( + bus.liveness_events(), + vec![(a, AgentLiveness::Stalled), (a, AgentLiveness::Alive)] + ); + + // Plus d'entrée : un sweep ultérieur ne ré-émet rien (même très tard). + clock.set(1_000_000); + inbox.sweep_stalled(); + assert_eq!( + bus.liveness_events(), + vec![(a, AgentLiveness::Stalled), (a, AgentLiveness::Alive)] + ); + } + + // (e) agent sans LivenessStrategy (stall_after_ms = None) ⇒ comportement legacy : + // jamais Stalled, aucun AgentLivenessChanged. + #[test] + fn agent_without_threshold_is_never_stalled() { + let clock = MutClock::new(0); + let (inbox, bus) = inbox_with_clock(Arc::clone(&clock)); + let a = agent(1); + // Pas de set_stall_threshold (ou None) : armé sans seuil au start_turn. + inbox.enqueue(a, ticket(10, "t")); + clock.set(10_000_000); // très loin dans le futur. + inbox.sweep_stalled(); + assert_eq!( + bus.liveness_events(), + vec![], + "sans seuil ⇒ jamais stalled (zéro régression)" + ); + // Et un mark_alive sur un agent jamais stalled n'émet rien non plus. + inbox.mark_alive(a); + assert_eq!(bus.liveness_events(), vec![]); + } + + // Un battement n'a aucun effet sur un agent dont aucune entrée n'est armée. + #[test] + fn mark_alive_on_unarmed_agent_is_noop() { + let clock = MutClock::new(0); + let (inbox, bus) = inbox_with_clock(Arc::clone(&clock)); + let a = agent(1); + inbox.mark_alive(a); // jamais de tour démarré. + inbox.sweep_stalled(); + assert_eq!(bus.liveness_events(), vec![]); + } } diff --git a/crates/infrastructure/src/session/claude.rs b/crates/infrastructure/src/session/claude.rs index 6ffeb15..3be6da8 100644 --- a/crates/infrastructure/src/session/claude.rs +++ b/crates/infrastructure/src/session/claude.rs @@ -47,10 +47,10 @@ pub struct ParsedLine { /// Le flux est du **JSONL** (un objet JSON par ligne). Types réels : /// /// - `{"type":"system","subtype":"init","session_id":"","cwd":…,"tools":…,…}` -/// ⇒ capture le `session_id` (= id de conversation pour la reprise), **aucun** -/// événement émis. +/// ⇒ capture le `session_id` (= id de conversation pour la reprise) **et** émet un +/// [`ReplyEvent::Heartbeat`] (preuve de vivacité non terminale : la CLI a démarré). /// - `{"type":"rate_limit_event","rate_limit_info":{…},"session_id":"…"}` -/// ⇒ **ignoré** (comme tout `type` inconnu). +/// ⇒ [`ReplyEvent::Heartbeat`] (pas de contenu, mais le moteur est vivant). /// - `{"type":"assistant","message":{"role":"assistant","content":[ /// {"type":"text","text":"…"} | {"type":"tool_use","name":"…", …} /// ], …},"session_id":"…","parent_tool_use_id":null}` @@ -81,7 +81,12 @@ pub fn parse_event(line: &str) -> Result { .map(str::to_owned); let events = match value.get("type").and_then(Value::as_str) { - Some("system") => Vec::new(), // init/handshake : on ne capte que le session_id. + // init/handshake : on capte le session_id ET on émet un battement de cœur + // (preuve de vivacité non terminale : la CLI a démarré et répond). + Some("system") => vec![ReplyEvent::Heartbeat], + // Fenêtre de limite de débit : pas de contenu, mais le moteur est vivant ⇒ + // battement de cœur (readiness/heartbeat lot 1), plus ignoré. + Some("rate_limit_event") => vec![ReplyEvent::Heartbeat], Some("assistant") => assistant_events(&value), Some("result") => value .get("result") @@ -92,7 +97,7 @@ pub fn parse_event(line: &str) -> Result { }] }) .unwrap_or_default(), - _ => Vec::new(), // type inconnu / non pertinent (rate_limit_event, …) : ignoré. + _ => Vec::new(), // type inconnu / non pertinent : ignoré (robustesse). }; Ok(ParsedLine { events, session_id }) @@ -216,13 +221,22 @@ impl AgentSession for ClaudeSdkSession { let mut events = Vec::new(); let mut captured_id = None; - for line in &raw_lines { + 'lines: for line in &raw_lines { let parsed = parse_event(line)?; if let Some(id) = parsed.session_id { captured_id = Some(id); } // Aplatit : une ligne `assistant` multi-blocs rend plusieurs événements. - events.extend(parsed.events); + // 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 (`rate_limit_event` tardif…) + // 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; + } + } } // Persiste le session_id capté (pivot de reprise) avant de rendre le flux. if let Some(id) = captured_id { diff --git a/crates/infrastructure/src/session/codex.rs b/crates/infrastructure/src/session/codex.rs index b3c683b..94042a8 100644 --- a/crates/infrastructure/src/session/codex.rs +++ b/crates/infrastructure/src/session/codex.rs @@ -43,13 +43,13 @@ pub struct ParsedLine { /// /// - `{"type":"thread.started","thread_id":""}` ⇒ capte le `thread_id` /// (= id de conversation pour la reprise), **aucun** événement émis. -/// - `{"type":"turn.started"}` ⇒ ignoré. +/// - `{"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":{…}}` ⇒ ignoré (le `Final` vient de -/// l'`agent_message`). +/// - `{"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é). @@ -75,6 +75,10 @@ pub fn parse_event(line: &str) -> Result { .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) { @@ -94,7 +98,7 @@ pub fn parse_event(line: &str) -> Result { } } } - // turn.started / turn.completed / type inconnu : ignoré (robustesse). + // type inconnu : ignoré (robustesse). _ => {} } @@ -192,12 +196,21 @@ impl AgentSession for CodexExecSession { let mut events = Vec::new(); let mut captured_id = None; - for line in &raw_lines { + 'lines: 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); + // 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); diff --git a/crates/infrastructure/src/session/conformance.rs b/crates/infrastructure/src/session/conformance.rs index a3b2ad7..79bc53c 100644 --- a/crates/infrastructure/src/session/conformance.rs +++ b/crates/infrastructure/src/session/conformance.rs @@ -208,14 +208,17 @@ pub(crate) mod harness { } other => panic!("le dernier événement doit être Final, vu: {other:?}"), } - // Les événements avant le Final ne sont que des deltas / activités. + // Les événements avant le Final ne sont que des deltas / activités / heartbeats + // (tous **non terminaux** ; le heartbeat est une preuve de vivacité, lot 1). for e in &events[..events.len() - 1] { assert!( matches!( e, - ReplyEvent::TextDelta { .. } | ReplyEvent::ToolActivity { .. } + ReplyEvent::TextDelta { .. } + | ReplyEvent::ToolActivity { .. } + | ReplyEvent::Heartbeat ), - "avant le Final, seuls deltas/activités sont permis, vu: {e:?}" + "avant le Final, seuls deltas/activités/heartbeats sont permis, vu: {e:?}" ); } diff --git a/crates/infrastructure/src/session/mod.rs b/crates/infrastructure/src/session/mod.rs index d064b40..89e3951 100644 --- a/crates/infrastructure/src/session/mod.rs +++ b/crates/infrastructure/src/session/mod.rs @@ -106,23 +106,24 @@ mod tests { // -- parse_event Claude (format RÉEL vérifié 2026-06-09) -------------- #[test] - fn claude_parse_init_captures_session_id_without_event() { + fn claude_parse_init_captures_session_id_and_heartbeats() { let parsed = claude::parse_event( r#"{"type":"system","subtype":"init","session_id":"conv-123","cwd":"/tmp","tools":[],"model":"claude-opus-4-8"}"#, ) .expect("parse ok"); assert_eq!(parsed.session_id.as_deref(), Some("conv-123")); - assert!(parsed.events.is_empty()); + // L'init capte le session_id ET émet un heartbeat (vivacité non terminale, lot 1). + assert_eq!(parsed.events, vec![ReplyEvent::Heartbeat]); } #[test] - fn claude_parse_rate_limit_event_is_ignored() { + fn claude_parse_rate_limit_event_is_heartbeat() { let parsed = claude::parse_event( r#"{"type":"rate_limit_event","rate_limit_info":{"x":1},"session_id":"conv-123"}"#, ) .expect("parse ok"); - // Type inconnu/non pertinent : ignoré (mais session_id tout de même capté). - assert!(parsed.events.is_empty()); + // Plus ignoré : preuve de vivacité ⇒ heartbeat (mais session_id tout de même capté). + assert_eq!(parsed.events, vec![ReplyEvent::Heartbeat]); assert_eq!(parsed.session_id.as_deref(), Some("conv-123")); } @@ -221,10 +222,15 @@ mod tests { let sess = codex::parse_event(r#"{"type":"thread.started","thread_id":"cx-9"}"#).expect("ok"); assert_eq!(sess.conversation_id.as_deref(), Some("cx-9")); + // Le handshake ne capte que le thread_id, sans événement (pas un heartbeat). assert!(sess.events.is_empty()); + // turn.started / turn.completed ⇒ heartbeat (vivacité non terminale, lot 1). let started = codex::parse_event(r#"{"type":"turn.started"}"#).expect("ok"); - assert!(started.events.is_empty()); + assert_eq!(started.events, vec![ReplyEvent::Heartbeat]); + let completed = + codex::parse_event(r#"{"type":"turn.completed","usage":{}}"#).expect("ok"); + assert_eq!(completed.events, vec![ReplyEvent::Heartbeat]); let msg = codex::parse_event( r#"{"type":"item.completed","item":{"id":"item_0","type":"agent_message","text":"fini"}}"#, @@ -236,11 +242,6 @@ mod tests { content: "fini".to_owned() }] ); - - let completed = - codex::parse_event(r#"{"type":"turn.completed","usage":{"input_tokens":10}}"#) - .expect("ok"); - assert!(completed.events.is_empty()); } #[test] @@ -728,30 +729,26 @@ mod tests { assert!(p.events.is_empty()); } - /// Ligne vide / type inconnu / turn.started / turn.completed ⇒ ignorés sans erreur. + /// Ligne vide / type inconnu ⇒ ignorés sans erreur. (turn.started/completed sont + /// désormais des heartbeats : couverts par `codex_parse_thread_started_message_and_final`.) #[test] fn codex_empty_and_unknown_ignored() { assert_eq!(codex::parse_event("").unwrap(), Default::default()); - assert!(codex::parse_event(r#"{"type":"heartbeat"}"#) + // Un `type` inconnu reste ignoré (robustesse), pas un heartbeat. + assert!(codex::parse_event(r#"{"type":"telemetry"}"#) .unwrap() .events .is_empty()); - assert!(codex::parse_event(r#"{"type":"turn.started"}"#) - .unwrap() - .events - .is_empty()); - assert!( - codex::parse_event(r#"{"type":"turn.completed","usage":{}}"#) - .unwrap() - .events - .is_empty() - ); } // ---- Machinerie process via FakeCli --------------------------------- - /// Deltas PUIS Final : le flux ne contient rien après le `Final` (déjà couvert - /// pour Claude ; ici on le prouve aussi pour Codex, substituabilité Liskov). + /// Deltas PUIS Final : **exactement un** `Final`, et aucun autre `Final` après lui + /// (substituabilité Liskov). Note (lot 1) : un `turn.completed` postérieur émet un + /// `Heartbeat` non terminal — légitimement après le `Final` —, donc on ne teste plus + /// « rien après le Final » mais « pas de second Final, et seul un heartbeat peut + /// suivre ». Le rendez-vous synchrone (`drain_to_final`) s'arrête de toute façon au + /// premier `Final`. #[tokio::test] async fn codex_stream_closed_after_final() { let fake = FakeCli::printing(&[ @@ -762,12 +759,19 @@ mod tests { ]); let s = CodexExecSession::new(SessionId::new_random(), fake.command(), "/", None); let events: Vec<_> = s.send("x").await.expect("send").collect(); - let after = events + let finals = events + .iter() + .filter(|e| matches!(e, ReplyEvent::Final { .. })) + .count(); + assert_eq!(finals, 1, "exactement un Final"); + // Après le Final, seuls des événements non terminaux (heartbeat) peuvent suivre. + let after_final_terminals = events .iter() .skip_while(|e| !matches!(e, ReplyEvent::Final { .. })) .skip(1) + .filter(|e| matches!(e, ReplyEvent::Final { .. })) .count(); - assert_eq!(after, 0); + assert_eq!(after_final_terminals, 0, "aucun second Final après le premier"); } /// LIMITE/ÉCART (à arbitrer) : un flux SANS `Final` ne provoque PAS d'erreur au @@ -1097,6 +1101,8 @@ mod tests { assert_eq!( events, vec![ + // L'init `system` émet un heartbeat (vivacité non terminale, lot 1). + ReplyEvent::Heartbeat, ReplyEvent::TextDelta { text: "a".into() }, ReplyEvent::ToolActivity { label: "T".into() }, ReplyEvent::TextDelta { text: "b".into() }, @@ -1104,7 +1110,7 @@ mod tests { content: "final-ok".into() }, ], - "le flot complet doit aplatir les 3 blocs PUIS un seul Final" + "heartbeat d'init, puis les 3 blocs aplatis, PUIS un seul Final" ); // Un seul Final, en dernière position (redondant mais explicite). assert_eq!(