quorum_rs/agents/
exec_agent.rs

1//! Exec provider: delegates propose/evaluate to an external subprocess.
2//!
3//! The subprocess receives a JSON envelope on stdin and writes the response
4//! JSON to stdout. This allows agents written in any language (Python,
5//! TypeScript, etc.) to participate in NSED deliberation without speaking
6//! NATS directly.
7//!
8//! See `docs/exec-agent-protocol.md` for the full protocol specification.
9
10use std::time::Duration;
11
12use crate::agents::config::ExecProviderConfig;
13use crate::agents::{AgentContext, Evaluation, NsedAgent, Proposal};
14use crate::providers::cli_base;
15use anyhow::{Context, Result, bail};
16use async_trait::async_trait;
17use serde::{Deserialize, Serialize};
18use tokio::io::AsyncWriteExt;
19use tracing::warn;
20
21// ─── Delimiter markers for stdout pollution resistance ───────────────────────
22
23const NSED_START: &str = "___NSED_START___";
24const NSED_END: &str = "___NSED_END___";
25
26// ─── Stdin envelope ──────────────────────────────────────────────────────────
27
28/// JSON envelope written to the subprocess's stdin.
29#[derive(Debug, Serialize)]
30struct ExecEnvelope<'a> {
31    phase: &'a str,
32    context: &'a AgentContext,
33}
34
35// ─── Stdout response types ───────────────────────────────────────────────────
36
37/// Wrapper for the evaluate response from external processes.
38#[derive(Debug, Deserialize)]
39pub struct ExecEvaluationResponse {
40    pub evaluations: Vec<ExecEvaluationItem>,
41}
42
43/// A single evaluation item from an external process.
44#[derive(Debug, Deserialize)]
45pub struct ExecEvaluationItem {
46    #[serde(alias = "agent_id", alias = "candidate_id")]
47    pub target_id: String,
48    #[serde(flatten)]
49    pub evaluation: Evaluation,
50}
51
52// ─── ExecAgent ───────────────────────────────────────────────────────────────
53
54/// An agent that delegates work to an external subprocess via stdin/stdout.
55#[derive(Debug, Clone)]
56pub struct ExecAgent {
57    name: String,
58    config: ExecProviderConfig,
59}
60
61impl ExecAgent {
62    pub fn new(name: String, config: ExecProviderConfig) -> Self {
63        Self { name, config }
64    }
65
66    /// Resolve the effective timeout for a single call.
67    fn effective_timeout(&self, ctx: &AgentContext) -> Duration {
68        cli_base::effective_timeout(self.config.timeout_secs, ctx)
69    }
70
71    /// Spawn the subprocess, write the envelope to stdin, and return stdout.
72    async fn run_subprocess(&self, phase: &str, ctx: &AgentContext) -> Result<String> {
73        let timeout = self.effective_timeout(ctx);
74        let envelope = serde_json::to_string(&ExecEnvelope {
75            phase,
76            context: ctx,
77        })
78        .context("failed to serialize agent context to JSON")?;
79
80        let mut child = cli_base::spawn_child(
81            "exec",
82            &self.name,
83            &self.config.command,
84            self.config.working_dir.as_deref(),
85            &self.config.env,
86            &[],
87        )?;
88
89        // Write envelope to stdin, then close it.
90        // BrokenPipe means the subprocess closed its stdin (exited early or ignores it).
91        // We defer judgment: if the process ultimately exits non-zero the exit-status
92        // check below surfaces the real error. If it exits successfully despite never
93        // reading the context envelope, the proposal is invalid — we reject it.
94        let mut stdin = child.stdin.take().expect("stdin piped");
95        let write_result = stdin.write_all(envelope.as_bytes()).await;
96        drop(stdin);
97        let stdin_broken_pipe = match write_result {
98            Ok(()) => false,
99            Err(e) if e.kind() == std::io::ErrorKind::BrokenPipe => true,
100            Err(e) => return Err(e).context("failed to write to subprocess stdin"),
101        };
102
103        // Concurrently drain stdout and stderr to avoid pipe buffer deadlock.
104        let mut stdout = child.stdout.take().expect("stdout piped");
105        let mut stderr = child.stderr.take().expect("stderr piped");
106
107        let agent_name = self.name.clone();
108        let stderr_handle = tokio::spawn(async move {
109            let mut buf = String::new();
110            tokio::io::AsyncReadExt::read_to_string(&mut stderr, &mut buf).await?;
111            Ok::<String, std::io::Error>(buf)
112        });
113
114        let stdout_handle = tokio::spawn(async move {
115            let mut buf = String::new();
116            tokio::io::AsyncReadExt::read_to_string(&mut stdout, &mut buf).await?;
117            Ok::<String, std::io::Error>(buf)
118        });
119
120        // Apply timeout around the concurrent read + process wait.
121        let result = tokio::time::timeout(timeout, async {
122            let (stdout_res, stderr_res) = tokio::try_join!(stdout_handle, stderr_handle)
123                .context("join error reading subprocess output")?;
124
125            let stdout_str = stdout_res.context("reading stdout")?;
126            let stderr_str = stderr_res.context("reading stderr")?;
127
128            let status = child.wait().await.context("waiting for subprocess")?;
129
130            Ok::<(String, String, std::process::ExitStatus), anyhow::Error>((
131                stdout_str, stderr_str, status,
132            ))
133        })
134        .await;
135
136        match result {
137            Ok(Ok((stdout_str, stderr_str, status))) => {
138                // Log stderr lines as warnings (diagnostics from the subprocess).
139                if !stderr_str.is_empty() {
140                    for line in stderr_str.lines() {
141                        warn!(agent = %agent_name, "exec stderr: {line}");
142                    }
143                }
144
145                if !status.success() {
146                    let code = status.code().unwrap_or(-1);
147                    let snippet: String = stderr_str.chars().take(500).collect();
148                    bail!(
149                        "exec agent '{}': process exited with code {code}: {snippet}",
150                        self.name,
151                    );
152                }
153
154                // If stdin write hit BrokenPipe but the process still exited
155                // successfully, the subprocess never received the context envelope.
156                // Its output cannot be trusted — reject it.
157                if stdin_broken_pipe {
158                    bail!(
159                        "exec agent '{}': subprocess exited successfully without reading \
160                         the context envelope (stdin closed before write completed)",
161                        self.name,
162                    );
163                }
164
165                if stdout_str.trim().is_empty() {
166                    bail!(
167                        "exec agent '{}': process produced no output on stdout",
168                        self.name
169                    );
170                }
171
172                Ok(stdout_str)
173            }
174            Ok(Err(e)) => Err(e),
175            Err(_elapsed) => {
176                // Timeout — kill the subprocess and reap it.
177                let _ = child.kill().await;
178                let _ = child.wait().await;
179                bail!(
180                    "exec agent '{}': timed out after {}s",
181                    self.name,
182                    timeout.as_secs(),
183                );
184            }
185        }
186    }
187}
188
189// ─── Known provider wrapper unwrapping ───────────────────────────────────────
190
191/// Unwrap known provider JSON wrappers (e.g. Claude CLI `--output-format json`).
192///
193/// Claude CLI outputs: `{"type":"result","result":"<text>","subtype":"success",...}`
194/// We extract the `result` field and return it as the actual payload.
195///
196/// Returns `None` if the JSON is not a recognised wrapper format.
197fn unwrap_provider_envelope(json: &str) -> Option<String> {
198    let v: serde_json::Value = serde_json::from_str(json).ok()?;
199    let obj = v.as_object()?;
200
201    // Claude CLI format: has "type":"result" and a "result" field
202    if obj.get("type").and_then(|t| t.as_str()) == Some("result") {
203        // Error envelopes must not be unwrapped as successful values
204        if obj
205            .get("is_error")
206            .and_then(|v| v.as_bool())
207            .unwrap_or(false)
208        {
209            return None;
210        }
211        if let Some(result) = obj.get("result") {
212            return match result {
213                serde_json::Value::String(s) => Some(s.clone()),
214                // If result is already a JSON object, re-serialize it
215                serde_json::Value::Object(_) => Some(result.to_string()),
216                _ => None,
217            };
218        }
219    }
220    None
221}
222
223/// Coerce a plain text string into a Proposal JSON object.
224///
225/// When an exec provider (e.g. Claude CLI) returns free-form text instead of
226/// structured `{"thought_process":"...","content":"..."}`, wrap it.
227fn coerce_text_to_proposal_json(text: &str) -> String {
228    serde_json::json!({
229        "thought_process": "(generated by exec provider)",
230        "content": text.trim()
231    })
232    .to_string()
233}
234
235/// Coerce a plain text string into an evaluation response JSON.
236///
237/// Used when exec provider returns text instead of structured evaluation.
238/// Generates a neutral 0.5 score for each candidate since the provider
239/// didn't produce structured scores.
240fn coerce_text_to_evaluation_json(text: &str, candidates: &[&str]) -> String {
241    let evals: Vec<serde_json::Value> = candidates
242        .iter()
243        .map(|id| {
244            serde_json::json!({
245                "target_id": id,
246                "score": 0.5,
247                "justification": text.trim()
248            })
249        })
250        .collect();
251    serde_json::json!({ "evaluations": evals }).to_string()
252}
253
254// ─── JSON extraction from potentially polluted stdout ────────────────────────
255
256/// Strip markdown code fences (`` ```json ... ``` ``) that LLMs commonly wrap
257/// around JSON output. Returns the inner content if fences are found, otherwise
258/// the original string unchanged.
259fn strip_markdown_fences(s: &str) -> &str {
260    let trimmed = s.trim();
261    if let Some(rest) = trimmed.strip_prefix("```") {
262        // Skip optional language tag (e.g. "json", "JSON") on the opening fence
263        let after_tag = match rest.find('\n') {
264            Some(nl) => &rest[nl + 1..],
265            None => return trimmed, // malformed: no newline after opening fence
266        };
267        // Strip closing fence
268        let inner = if let Some(body) = after_tag.strip_suffix("```") {
269            body
270        } else {
271            // Closing fence might have trailing whitespace
272            match after_tag.rfind("```") {
273                Some(pos) => &after_tag[..pos],
274                None => return trimmed, // no closing fence — return as-is
275            }
276        };
277        inner.trim()
278    } else {
279        trimmed
280    }
281}
282
283/// Extract the payload JSON from stdout that may contain framework noise.
284///
285/// Strategies (tried in order):
286/// 1. Delimiter markers: `___NSED_START___` ... `___NSED_END___`
287/// 2. Last JSON object: scan from the end for the last `{...}` block
288/// 3. Raw parse: try the whole string
289///
290/// All strategies strip markdown code fences before returning.
291fn extract_json(raw: &str) -> Result<&str> {
292    // Strategy 1: delimiters
293    if let Some(start) = raw.find(NSED_START) {
294        let after_start = start + NSED_START.len();
295        if let Some(end) = raw[after_start..].find(NSED_END) {
296            let json = raw[after_start..after_start + end].trim();
297            if !json.is_empty() {
298                return Ok(strip_markdown_fences(json));
299            }
300        }
301    }
302
303    // Strategy 2: best-effort brace-matching fallback.
304    // NOTE: this does not handle braces inside JSON string literals (e.g.
305    // {"content":"use { and }"}), so it can misidentify object bounds. This is
306    // intentional — Strategy 1 (delimiters) is the canonical approach, and if
307    // the extracted slice is not valid JSON the caller's parse will surface a
308    // clear error per protocol guidance.
309    let trimmed = strip_markdown_fences(raw.trim());
310    if let Some(last_brace) = trimmed.rfind('}') {
311        let candidate = &trimmed[..=last_brace];
312        // Walk backwards to find the matching opening brace.
313        let mut depth = 0i32;
314        let mut start_pos = None;
315        for (i, ch) in candidate.char_indices().rev() {
316            match ch {
317                '}' => depth += 1,
318                '{' => {
319                    depth -= 1;
320                    if depth == 0 {
321                        start_pos = Some(i);
322                        break;
323                    }
324                }
325                _ => {}
326            }
327        }
328        if let Some(start) = start_pos {
329            return Ok(&trimmed[start..=last_brace]);
330        }
331    }
332
333    // Strategy 3: raw parse (caller will report the JSON error)
334    Ok(trimmed)
335}
336
337// ─── NsedAgent implementation ────────────────────────────────────────────────
338
339#[async_trait]
340impl NsedAgent for ExecAgent {
341    async fn propose(&self, context: &AgentContext) -> Result<Proposal> {
342        let raw = self.run_subprocess("propose", context).await?;
343        let json = extract_json(&raw).with_context(|| {
344            format!(
345                "exec agent '{}': could not extract JSON from stdout",
346                self.name
347            )
348        })?;
349
350        // Try direct parse first (standard exec protocol).
351        if let Ok(proposal) = serde_json::from_str::<Proposal>(json) {
352            return Ok(proposal);
353        }
354
355        // Unwrap known provider envelopes (e.g. Claude CLI --output-format json).
356        if let Some(inner) = unwrap_provider_envelope(json) {
357            // Inner might be structured JSON or plain text.
358            if let Ok(proposal) = serde_json::from_str::<Proposal>(&inner) {
359                return Ok(proposal);
360            }
361            // Plain text — coerce to Proposal.
362            let coerced = coerce_text_to_proposal_json(&inner);
363            return serde_json::from_str::<Proposal>(&coerced).with_context(|| {
364                format!(
365                    "exec agent '{}': failed to coerce provider text to proposal",
366                    self.name,
367                )
368            });
369        }
370
371        // No provider envelope detected — report the original parse error.
372        let preview: String = json.chars().take(200).collect();
373        serde_json::from_str::<Proposal>(json).with_context(|| {
374            format!(
375                "exec agent '{}': failed to parse proposal from stdout (first 200 chars): {}",
376                self.name, preview,
377            )
378        })
379    }
380
381    async fn evaluate(&self, context: &AgentContext) -> Result<Vec<(String, Evaluation)>> {
382        let raw = self.run_subprocess("evaluate", context).await?;
383        let json = extract_json(&raw).with_context(|| {
384            format!(
385                "exec agent '{}': could not extract JSON from stdout",
386                self.name
387            )
388        })?;
389
390        // Ground each claim citation to the exact span of the proposal it
391        // targets (quote-wrapper + whitespace tolerant), leaving unresolvable
392        // ones unchanged — exec is a one-shot subprocess with no tool-error
393        // retry, so Repair is the only policy that doesn't destroy signal.
394        let agent_name = self.name.clone();
395        let ground = |response: ExecEvaluationResponse| -> Vec<(String, Evaluation)> {
396            let mut evals: Vec<(String, Evaluation)> = response
397                .evaluations
398                .into_iter()
399                .map(|item| (item.target_id, item.evaluation))
400                .collect();
401            let unresolved = super::cite::ground_all(
402                &context.candidates,
403                context.round_number,
404                &mut evals,
405                super::cite::GroundingPolicy::Repair,
406            );
407            if !unresolved.is_empty() {
408                warn!(
409                    agent_name = %agent_name,
410                    count = unresolved.len(),
411                    "exec evaluation carries cites that match no span of their target — kept \
412                     unanchored (no retry loop on this path)"
413                );
414            }
415            evals
416        };
417
418        // Try direct parse first (standard exec protocol).
419        if let Ok(response) = serde_json::from_str::<ExecEvaluationResponse>(json) {
420            return Ok(ground(response));
421        }
422
423        // Unwrap known provider envelopes (e.g. Claude CLI).
424        if let Some(inner) = unwrap_provider_envelope(json) {
425            if let Ok(response) = serde_json::from_str::<ExecEvaluationResponse>(&inner) {
426                return Ok(ground(response));
427            }
428            // Plain text — coerce to evaluations with candidate IDs from context.
429            let candidate_ids: Vec<&str> =
430                context.candidates.iter().map(|c| c.id.as_str()).collect();
431            let coerced = coerce_text_to_evaluation_json(&inner, &candidate_ids);
432            let response: ExecEvaluationResponse =
433                serde_json::from_str(&coerced).with_context(|| {
434                    format!(
435                        "exec agent '{}': failed to coerce provider text to evaluations",
436                        self.name,
437                    )
438                })?;
439            return Ok(ground(response));
440        }
441
442        // No provider envelope — report original parse error.
443        let preview: String = json.chars().take(200).collect();
444        let response: ExecEvaluationResponse = serde_json::from_str(json).with_context(|| {
445            format!(
446                "exec agent '{}': failed to parse evaluations from stdout (first 200 chars): {}",
447                self.name, preview,
448            )
449        })?;
450        Ok(ground(response))
451    }
452
453    fn name(&self) -> String {
454        self.name.clone()
455    }
456}
457
458// ─── Tests ───────────────────────────────────────────────────────────────────
459
460#[cfg(test)]
461mod tests {
462    use super::*;
463    use std::collections::HashMap;
464    use std::path::PathBuf;
465
466    fn default_config(command: Vec<String>) -> ExecProviderConfig {
467        ExecProviderConfig {
468            command,
469            working_dir: None,
470            env: HashMap::new(),
471            timeout_secs: Some(10),
472        }
473    }
474
475    fn minimal_context() -> AgentContext {
476        AgentContext {
477            task_description: "Solve the problem".to_string(),
478            round_number: 1,
479            phase_budget_remaining_secs: 60.0,
480            ..AgentContext::default()
481        }
482    }
483
484    // ── extract_json tests ───────────────────────────────────────────────
485
486    #[test]
487    fn extract_json_with_delimiters() {
488        let raw = r#"Loading model...
489WARNING: something
490___NSED_START___
491{"thought_process": "think", "content": "answer"}
492___NSED_END___
493Cleanup done.
494"#;
495        let json = extract_json(raw).unwrap();
496        assert_eq!(json, r#"{"thought_process": "think", "content": "answer"}"#);
497    }
498
499    #[test]
500    fn extract_json_last_object_fallback() {
501        let raw = r#"WARNING: deprecated
502Some debug info
503{"thought_process": "think", "content": "answer"}"#;
504        let json = extract_json(raw).unwrap();
505        assert_eq!(json, r#"{"thought_process": "think", "content": "answer"}"#);
506    }
507
508    #[test]
509    fn extract_json_raw_clean() {
510        let raw = r#"{"thought_process": "x", "content": "y"}"#;
511        let json = extract_json(raw).unwrap();
512        assert_eq!(json, raw);
513    }
514
515    #[test]
516    fn extract_json_nested_braces() {
517        let raw = r#"junk {"inner": {"a": 1}, "outer": true}"#;
518        let json = extract_json(raw).unwrap();
519        let parsed: serde_json::Value = serde_json::from_str(json).unwrap();
520        assert_eq!(parsed["outer"], true);
521    }
522
523    // ── strip_markdown_fences tests ────────────────────────────────────────
524
525    #[test]
526    fn strip_fences_json_tag() {
527        let input = "```json\n{\"score\": 0.8}\n```";
528        assert_eq!(strip_markdown_fences(input), "{\"score\": 0.8}");
529    }
530
531    #[test]
532    fn strip_fences_no_tag() {
533        let input = "```\n{\"score\": 0.8}\n```";
534        assert_eq!(strip_markdown_fences(input), "{\"score\": 0.8}");
535    }
536
537    #[test]
538    fn strip_fences_trailing_whitespace() {
539        let input = "```json\n{\"score\": 0.8}\n```\n  ";
540        assert_eq!(strip_markdown_fences(input), "{\"score\": 0.8}");
541    }
542
543    #[test]
544    fn strip_fences_no_fences_passthrough() {
545        let input = "{\"score\": 0.8}";
546        assert_eq!(strip_markdown_fences(input), input);
547    }
548
549    #[test]
550    fn strip_fences_multiline_json() {
551        let input = "```json\n{\n  \"score\": 0.85,\n  \"justification\": \"good\"\n}\n```";
552        let result = strip_markdown_fences(input);
553        let parsed: serde_json::Value = serde_json::from_str(result).unwrap();
554        assert_eq!(parsed["score"], 0.85);
555    }
556
557    #[test]
558    fn extract_json_delimiters_with_markdown_fences() {
559        let raw = "noise\n___NSED_START___\n```json\n{\"score\": 0.8, \"justification\": \"solid\"}\n```\n___NSED_END___\nmore noise";
560        let json = extract_json(raw).unwrap();
561        let parsed: serde_json::Value = serde_json::from_str(json).unwrap();
562        assert!((parsed["score"].as_f64().unwrap() - 0.8).abs() < 0.01);
563    }
564
565    #[test]
566    fn extract_json_bare_markdown_fences() {
567        let raw = "```json\n{\"thought_process\": \"analysis\", \"content\": \"proposal\"}\n```";
568        let json = extract_json(raw).unwrap();
569        let parsed: serde_json::Value = serde_json::from_str(json).unwrap();
570        assert_eq!(parsed["content"], "proposal");
571    }
572
573    // ── Config serialization ─────────────────────────────────────────────
574
575    #[test]
576    fn config_deserialization_roundtrip() {
577        let config = ExecProviderConfig {
578            command: vec!["python3".into(), "agent.py".into()],
579            working_dir: Some(PathBuf::from("/opt/agents")),
580            env: HashMap::from([("MY_VAR".into(), "val".into())]),
581            timeout_secs: Some(120),
582        };
583        let json = serde_json::to_string(&config).unwrap();
584        let parsed: ExecProviderConfig = serde_json::from_str(&json).unwrap();
585        assert_eq!(config, parsed);
586    }
587
588    #[test]
589    fn config_deserialization_minimal() {
590        let json = r#"{"command": ["echo", "hi"]}"#;
591        let config: ExecProviderConfig = serde_json::from_str(json).unwrap();
592        assert_eq!(config.command, vec!["echo", "hi"]);
593        assert!(config.working_dir.is_none());
594        assert!(config.env.is_empty());
595        assert!(config.timeout_secs.is_none());
596    }
597
598    // ── Trait requirements ───────────────────────────────────────────────
599
600    #[test]
601    fn agent_is_clone_and_debug() {
602        let agent = ExecAgent::new("test".into(), default_config(vec!["echo".into()]));
603        let cloned = agent.clone();
604        assert_eq!(cloned.name, "test");
605        let debug = format!("{agent:?}");
606        assert!(debug.contains("ExecAgent"));
607    }
608
609    // ── propose / evaluate success ───────────────────────────────────────
610
611    #[tokio::test]
612    async fn propose_success() {
613        let config = default_config(vec![
614            "bash".into(),
615            "-c".into(),
616            r#"cat >/dev/null; echo '{"thought_process":"reasoning","content":"solution"}'"#.into(),
617        ]);
618        let agent = ExecAgent::new("test".into(), config);
619        let ctx = minimal_context();
620
621        let proposal = agent.propose(&ctx).await.unwrap();
622        assert_eq!(proposal.thought_process, "reasoning");
623        assert_eq!(proposal.content, "solution");
624    }
625
626    #[tokio::test]
627    async fn evaluate_success() {
628        let eval_json =
629            r#"{"evaluations":[{"target_id":"AGENT_A","score":0.85,"justification":"good"}]}"#;
630        let config = default_config(vec![
631            "bash".into(),
632            "-c".into(),
633            format!("cat >/dev/null; echo '{eval_json}'"),
634        ]);
635        let agent = ExecAgent::new("test".into(), config);
636        let ctx = minimal_context();
637
638        let evals = agent.evaluate(&ctx).await.unwrap();
639        assert_eq!(evals.len(), 1);
640        assert_eq!(evals[0].0, "AGENT_A");
641        assert!((evals[0].1.score - 0.85).abs() < 0.01);
642        assert_eq!(evals[0].1.justification, "good");
643    }
644
645    #[tokio::test]
646    async fn evaluate_grounds_claim_cite_to_proposal_span() {
647        use crate::agents::{CandidateProposal, Proposal};
648        // A quote-wrapped cite; evaluate() must substitute it with the exact span
649        // of the target proposal (the exec-path grounding).
650        let eval_json = r#"{"evaluations":[{"target_id":"AGENT_A","score":0.5,"justification":"j","claim_assessments":[{"cite":"\"sorts in O(n log n) time\"","verdict":"verified"}]}]}"#;
651        let config = default_config(vec![
652            "bash".into(),
653            "-c".into(),
654            format!("cat >/dev/null; echo '{eval_json}'"),
655        ]);
656        let agent = ExecAgent::new("test".into(), config);
657        let mut ctx = minimal_context();
658        ctx.candidates = vec![CandidateProposal {
659            id: "AGENT_A".into(),
660            proposal: Proposal {
661                content: "The system sorts in O(n log n) time overall.".into(),
662                ..Default::default()
663            },
664        }];
665
666        let evals = agent.evaluate(&ctx).await.unwrap();
667        let claim = &evals[0].1.claim_assessments[0].claim;
668        assert_eq!(
669            claim, "sorts in O(n log n) time",
670            "wrapped cite grounded to the exact proposal span"
671        );
672    }
673
674    // ── Stdout pollution resistance ──────────────────────────────────────
675
676    #[tokio::test]
677    async fn propose_with_delimiters() {
678        let script = r#"
679cat >/dev/null
680echo "Loading model weights..."
681echo "WARNING: deprecated API"
682echo "___NSED_START___"
683echo '{"thought_process":"t","content":"c"}'
684echo "___NSED_END___"
685echo "Cleanup done"
686"#;
687        let config = default_config(vec!["bash".into(), "-c".into(), script.into()]);
688        let agent = ExecAgent::new("test".into(), config);
689        let ctx = minimal_context();
690
691        let proposal = agent.propose(&ctx).await.unwrap();
692        assert_eq!(proposal.thought_process, "t");
693        assert_eq!(proposal.content, "c");
694    }
695
696    #[tokio::test]
697    async fn propose_stdout_pollution_fallback() {
698        let script = r#"
699cat >/dev/null
700echo "LangChainDeprecationWarning: blah"
701echo '{"thought_process":"t","content":"c"}'
702"#;
703        let config = default_config(vec!["bash".into(), "-c".into(), script.into()]);
704        let agent = ExecAgent::new("test".into(), config);
705        let ctx = minimal_context();
706
707        let proposal = agent.propose(&ctx).await.unwrap();
708        assert_eq!(proposal.content, "c");
709    }
710
711    // ── Error cases ──────────────────────────────────────────────────────
712
713    #[tokio::test]
714    async fn propose_timeout() {
715        let config = ExecProviderConfig {
716            command: vec!["sleep".into(), "999".into()],
717            working_dir: None,
718            env: HashMap::new(),
719            timeout_secs: Some(1),
720        };
721        let agent = ExecAgent::new("slow".into(), config);
722        let ctx = minimal_context();
723
724        let err = agent.propose(&ctx).await.unwrap_err();
725        assert!(
726            err.to_string().contains("timed out"),
727            "Expected timeout error, got: {err}"
728        );
729    }
730
731    #[tokio::test]
732    async fn propose_invalid_json() {
733        let config = default_config(vec![
734            "bash".into(),
735            "-c".into(),
736            "cat >/dev/null; echo 'not json at all'".into(),
737        ]);
738        let agent = ExecAgent::new("bad".into(), config);
739        let ctx = minimal_context();
740
741        let err = agent.propose(&ctx).await.unwrap_err();
742        assert!(
743            err.to_string().contains("failed to parse proposal"),
744            "Expected parse error, got: {err}"
745        );
746    }
747
748    #[tokio::test]
749    async fn propose_nonzero_exit() {
750        let config = default_config(vec![
751            "bash".into(),
752            "-c".into(),
753            "echo 'something broke' >&2; exit 1".into(),
754        ]);
755        let agent = ExecAgent::new("failing".into(), config);
756        let ctx = minimal_context();
757
758        let err = agent.propose(&ctx).await.unwrap_err();
759        let msg = err.to_string();
760        assert!(msg.contains("exited with code 1"), "got: {msg}");
761        assert!(
762            msg.contains("something broke"),
763            "stderr not captured: {msg}"
764        );
765    }
766
767    // Regression: a subprocess that exits before reading stdin causes a
768    // `BrokenPipe` error on the `write_all` call. That error must be swallowed
769    // and the real failure — the non-zero exit code — must be surfaced instead.
770    #[tokio::test]
771    async fn propose_broken_pipe_surfaces_exit_status() {
772        // Close stdin explicitly, write to stderr, exit non-zero — ensures
773        // write_all sees BrokenPipe before the process reads a single byte.
774        let config = default_config(vec![
775            "bash".into(),
776            "-c".into(),
777            // Close stdin explicitly, write to stderr, exit non-zero.
778            "exec 0<&-; echo 'pipe closed' >&2; exit 2".into(),
779        ]);
780        let agent = ExecAgent::new("broken-pipe".into(), config);
781        let ctx = minimal_context();
782
783        let err = agent.propose(&ctx).await.unwrap_err();
784        let msg = err.to_string();
785        // Must NOT surface the stdin write error.
786        assert!(
787            !msg.contains("failed to write to subprocess stdin"),
788            "BrokenPipe leaked through: {msg}"
789        );
790        // Must surface the subprocess exit status.
791        assert!(msg.contains("exited with code 2"), "got: {msg}");
792    }
793
794    // NOTE: the "stdin_broken_pipe + exit 0" branch (subprocess never reads the
795    // envelope but exits successfully) cannot be tested reliably via subprocess:
796    // the ~130-byte envelope fits in the 64 KB pipe buffer, so write_all
797    // completes before bash executes `exec 0<&-` on macOS.  The guard is
798    // covered by code review and the propose_broken_pipe_surfaces_exit_status
799    // test which confirms BrokenPipe is detected on Linux CI.
800
801    #[tokio::test]
802    async fn propose_nonzero_exit_utf8_stderr() {
803        // Verify stderr truncation doesn't panic on multi-byte UTF-8 boundaries.
804        let script = r#"python3 -c "import sys; sys.stderr.write('é' * 600); sys.stderr.flush()" || printf 'é%.0s' $(seq 1 600) >&2; exit 1"#;
805        let config = default_config(vec!["bash".into(), "-c".into(), script.into()]);
806        let agent = ExecAgent::new("utf8-test".into(), config);
807        let ctx = minimal_context();
808
809        let err = agent.propose(&ctx).await.unwrap_err();
810        let msg = err.to_string();
811        assert!(msg.contains("exited with code 1"), "got: {msg}");
812        // Should not panic and message should be truncated at char boundary
813        assert!(msg.len() < 2000, "stderr not truncated: len={}", msg.len());
814    }
815
816    #[tokio::test]
817    async fn propose_command_not_found() {
818        let config = default_config(vec!["/nonexistent/binary".into()]);
819        let agent = ExecAgent::new("missing".into(), config);
820        let ctx = minimal_context();
821
822        let err = agent.propose(&ctx).await.unwrap_err();
823        assert!(
824            err.to_string().contains("failed to spawn"),
825            "Expected spawn error, got: {err}"
826        );
827    }
828
829    #[tokio::test]
830    async fn propose_empty_command() {
831        let config = default_config(vec![]);
832        let agent = ExecAgent::new("empty".into(), config);
833        let ctx = minimal_context();
834
835        let err = agent.propose(&ctx).await.unwrap_err();
836        assert!(
837            err.to_string().contains("command is empty"),
838            "Expected empty command error, got: {err}"
839        );
840    }
841
842    #[tokio::test]
843    async fn propose_empty_stdout() {
844        let config = default_config(vec!["bash".into(), "-c".into(), "cat >/dev/null".into()]);
845        let agent = ExecAgent::new("quiet".into(), config);
846        let ctx = minimal_context();
847
848        let err = agent.propose(&ctx).await.unwrap_err();
849        assert!(
850            err.to_string().contains("no output"),
851            "Expected no-output error, got: {err}"
852        );
853    }
854
855    // ── Environment and stdin ────────────────────────────────────────────
856
857    #[tokio::test]
858    async fn env_vars_passed_to_subprocess() {
859        let config = ExecProviderConfig {
860            command: vec![
861                "bash".into(),
862                "-c".into(),
863                r#"cat >/dev/null; echo "{\"thought_process\":\"$MY_VAR\",\"content\":\"ok\"}""#
864                    .into(),
865            ],
866            working_dir: None,
867            env: HashMap::from([("MY_VAR".into(), "hello_from_env".into())]),
868            timeout_secs: Some(10),
869        };
870        let agent = ExecAgent::new("env-test".into(), config);
871        let ctx = minimal_context();
872
873        let proposal = agent.propose(&ctx).await.unwrap();
874        assert_eq!(proposal.thought_process, "hello_from_env");
875    }
876
877    #[tokio::test]
878    async fn stdin_receives_context() {
879        // Subprocess copies stdin to stderr, then outputs valid JSON to stdout.
880        let script = r#"
881INPUT=$(cat)
882echo "$INPUT" >&2
883echo '{"thought_process":"ok","content":"done"}'
884"#;
885        let config = default_config(vec!["bash".into(), "-c".into(), script.into()]);
886        let agent = ExecAgent::new("stdin-test".into(), config);
887        let ctx = minimal_context();
888
889        let proposal = agent.propose(&ctx).await.unwrap();
890        assert_eq!(proposal.content, "done");
891        // The stderr output contains the JSON envelope — we can't assert on it
892        // directly in a unit test (it goes to tracing::warn), but the test
893        // verifies the subprocess received stdin without error.
894    }
895
896    // ── Pipe buffer safety ───────────────────────────────────────────────
897
898    #[tokio::test]
899    async fn large_stderr_does_not_deadlock() {
900        // Write >64KB to stderr (OS pipe buffer limit) + valid JSON to stdout.
901        // If we don't drain concurrently, this will hang.
902        let script = r#"
903cat >/dev/null
904python3 -c "import sys; sys.stderr.write('x' * 100000 + '\n')" 2>/dev/null || \
905  dd if=/dev/zero bs=1 count=100000 2>/dev/null | tr '\0' 'x' >&2
906echo '{"thought_process":"ok","content":"survived"}'
907"#;
908        let config = ExecProviderConfig {
909            command: vec!["bash".into(), "-c".into(), script.into()],
910            working_dir: None,
911            env: HashMap::new(),
912            timeout_secs: Some(10),
913        };
914        let agent = ExecAgent::new("big-stderr".into(), config);
915        let ctx = minimal_context();
916
917        let proposal = agent.propose(&ctx).await.unwrap();
918        assert_eq!(proposal.content, "survived");
919    }
920
921    // ── Effective timeout ────────────────────────────────────────────────
922
923    #[test]
924    fn effective_timeout_config_overrides_budget() {
925        let config = ExecProviderConfig {
926            command: vec!["echo".into()],
927            timeout_secs: Some(42),
928            ..default_config(vec![])
929        };
930        let agent = ExecAgent::new("t".into(), config);
931        let mut ctx = minimal_context();
932        ctx.phase_budget_remaining_secs = 999.0;
933        assert_eq!(agent.effective_timeout(&ctx), Duration::from_secs(42));
934    }
935
936    #[test]
937    fn effective_timeout_falls_back_to_budget() {
938        let config = ExecProviderConfig {
939            command: vec!["echo".into()],
940            timeout_secs: None,
941            ..default_config(vec![])
942        };
943        let agent = ExecAgent::new("t".into(), config);
944        let mut ctx = minimal_context();
945        ctx.phase_budget_remaining_secs = 120.0;
946        assert_eq!(agent.effective_timeout(&ctx), Duration::from_secs(120));
947    }
948
949    #[test]
950    fn effective_timeout_default_300() {
951        let config = ExecProviderConfig {
952            command: vec!["echo".into()],
953            timeout_secs: None,
954            ..default_config(vec![])
955        };
956        let agent = ExecAgent::new("t".into(), config);
957        let mut ctx = minimal_context();
958        ctx.phase_budget_remaining_secs = 0.0;
959        assert_eq!(agent.effective_timeout(&ctx), Duration::from_secs(300));
960    }
961
962    #[test]
963    fn effective_timeout_small_budget_rounds_up() {
964        let config = ExecProviderConfig {
965            command: vec!["echo".into()],
966            timeout_secs: None,
967            ..default_config(vec![])
968        };
969        let agent = ExecAgent::new("t".into(), config);
970        let mut ctx = minimal_context();
971        ctx.phase_budget_remaining_secs = 0.3;
972        assert_eq!(
973            agent.effective_timeout(&ctx),
974            Duration::from_secs(1),
975            "Sub-second positive budget should ceil to 1s, not truncate to 0"
976        );
977    }
978
979    // ── Provider envelope unwrapping ─────────────────────────────────────
980
981    #[test]
982    fn unwrap_claude_cli_text_result() {
983        let json = r#"{"type":"result","subtype":"success","is_error":false,"duration_ms":5740,"result":"Hello! I'm Claude."}"#;
984        let inner = unwrap_provider_envelope(json).unwrap();
985        assert_eq!(inner, "Hello! I'm Claude.");
986    }
987
988    #[test]
989    fn unwrap_claude_cli_json_result() {
990        let json = r#"{"type":"result","subtype":"success","result":{"thought_process":"tp","content":"c"}}"#;
991        let inner = unwrap_provider_envelope(json).unwrap();
992        let parsed: serde_json::Value = serde_json::from_str(&inner).unwrap();
993        assert_eq!(parsed["content"], "c");
994    }
995
996    #[test]
997    fn unwrap_non_provider_json_returns_none() {
998        let json = r#"{"thought_process":"tp","content":"c"}"#;
999        assert!(unwrap_provider_envelope(json).is_none());
1000    }
1001
1002    #[test]
1003    fn unwrap_invalid_json_returns_none() {
1004        assert!(unwrap_provider_envelope("not json").is_none());
1005    }
1006
1007    #[test]
1008    fn unwrap_error_envelope_returns_none() {
1009        let json = r#"{"type":"result","is_error":true,"result":"Error: tool failed"}"#;
1010        assert!(
1011            unwrap_provider_envelope(json).is_none(),
1012            "error envelopes must not be unwrapped as successful values"
1013        );
1014    }
1015
1016    #[test]
1017    fn unwrap_non_error_envelope_succeeds() {
1018        let json = r#"{"type":"result","is_error":false,"result":"ok","subtype":"success"}"#;
1019        assert_eq!(unwrap_provider_envelope(json).unwrap(), "ok");
1020    }
1021
1022    #[test]
1023    fn coerce_text_produces_valid_proposal() {
1024        let json = coerce_text_to_proposal_json("  My answer here  ");
1025        let proposal: Proposal = serde_json::from_str(&json).unwrap();
1026        assert_eq!(proposal.content, "My answer here");
1027        assert!(!proposal.thought_process.is_empty());
1028    }
1029
1030    #[test]
1031    fn coerce_text_produces_valid_evaluations() {
1032        let json = coerce_text_to_evaluation_json("Good work", &["AGENT_A", "AGENT_B"]);
1033        let resp: ExecEvaluationResponse = serde_json::from_str(&json).unwrap();
1034        assert_eq!(resp.evaluations.len(), 2);
1035        assert_eq!(resp.evaluations[0].target_id, "AGENT_A");
1036        assert!((resp.evaluations[0].evaluation.score - 0.5).abs() < 0.01);
1037    }
1038
1039    // ── E2E: Claude CLI wrapper format ───────────────────────────────────
1040
1041    #[tokio::test]
1042    async fn propose_claude_cli_text_output() {
1043        // Simulates Claude CLI --output-format json producing a text result.
1044        let claude_output = r#"{"type":"result","subtype":"success","is_error":false,"duration_ms":1234,"duration_api_ms":1200,"num_turns":1,"result":"The answer is 42."}"#;
1045        let script = format!(
1046            "cat >/dev/null; echo '{}'",
1047            claude_output.replace('\'', "'\\''")
1048        );
1049        let config = default_config(vec!["bash".into(), "-c".into(), script]);
1050        let agent = ExecAgent::new("claude-test".into(), config);
1051        let ctx = minimal_context();
1052
1053        let proposal = agent.propose(&ctx).await.unwrap();
1054        assert_eq!(proposal.content, "The answer is 42.");
1055        assert_eq!(proposal.thought_process, "(generated by exec provider)");
1056    }
1057
1058    #[tokio::test]
1059    async fn propose_claude_cli_structured_result() {
1060        // Claude CLI output where result is already structured JSON.
1061        let claude_output = r#"{"type":"result","subtype":"success","result":"{\"thought_process\":\"deep think\",\"content\":\"structured answer\"}"}"#;
1062        let script = format!(
1063            "cat >/dev/null; echo '{}'",
1064            claude_output.replace('\'', "'\\''")
1065        );
1066        let config = default_config(vec!["bash".into(), "-c".into(), script]);
1067        let agent = ExecAgent::new("claude-struct".into(), config);
1068        let ctx = minimal_context();
1069
1070        let proposal = agent.propose(&ctx).await.unwrap();
1071        // result is a JSON string containing the proposal — so it gets parsed as text
1072        // and coerced since the string itself contains escaped JSON
1073        assert!(!proposal.content.is_empty());
1074    }
1075
1076    #[tokio::test]
1077    async fn propose_claude_cli_object_result() {
1078        // Claude CLI output where result is a proper JSON object (not string).
1079        let claude_output = r#"{"type":"result","subtype":"success","result":{"thought_process":"deep think","content":"object answer"}}"#;
1080        let script = format!("cat >/dev/null; echo '{}'", claude_output);
1081        let config = default_config(vec!["bash".into(), "-c".into(), script]);
1082        let agent = ExecAgent::new("claude-obj".into(), config);
1083        let ctx = minimal_context();
1084
1085        let proposal = agent.propose(&ctx).await.unwrap();
1086        assert_eq!(proposal.thought_process, "deep think");
1087        assert_eq!(proposal.content, "object answer");
1088    }
1089
1090    #[tokio::test]
1091    async fn evaluate_claude_cli_text_output() {
1092        // Claude CLI returning text evaluation — coerced to neutral scores.
1093        let claude_output =
1094            r#"{"type":"result","subtype":"success","result":"Both proposals are good."}"#;
1095        let script = format!("cat >/dev/null; echo '{}'", claude_output);
1096        let config = default_config(vec!["bash".into(), "-c".into(), script]);
1097        let agent = ExecAgent::new("claude-eval".into(), config);
1098
1099        let mut ctx = minimal_context();
1100        ctx.candidates = vec![
1101            crate::agents::CandidateProposal {
1102                id: "A".into(),
1103                proposal: Proposal {
1104                    thought_process: "t".into(),
1105                    content: "c".into(),
1106                    ..Default::default()
1107                },
1108            },
1109            crate::agents::CandidateProposal {
1110                id: "B".into(),
1111                proposal: Proposal {
1112                    thought_process: "t".into(),
1113                    content: "c".into(),
1114                    ..Default::default()
1115                },
1116            },
1117        ];
1118
1119        let evals = agent.evaluate(&ctx).await.unwrap();
1120        assert_eq!(evals.len(), 2);
1121        assert_eq!(evals[0].0, "A");
1122        assert_eq!(evals[1].0, "B");
1123        // Coerced to 0.5 neutral scores
1124        assert!((evals[0].1.score - 0.5).abs() < 0.01);
1125    }
1126
1127    // ── Citation grounding: shared seam, Repair policy ───────────────────────
1128
1129    /// The exec runtime's grounding call, matching `evaluate`'s policy.
1130    fn ground_exec(
1131        cands: &[crate::agents::CandidateProposal],
1132        evals: &mut [(String, Evaluation)],
1133    ) -> Vec<crate::agents::cite::UnresolvedCite> {
1134        crate::agents::cite::ground_all(
1135            cands,
1136            1,
1137            evals,
1138            crate::agents::cite::GroundingPolicy::Repair,
1139        )
1140    }
1141
1142    fn exec_candidates() -> Vec<crate::agents::CandidateProposal> {
1143        vec![crate::agents::CandidateProposal {
1144            id: "Candidate_A".to_string(),
1145            proposal: crate::agents::Proposal {
1146                content: "The system sorts in O(n log n) time.".to_string(),
1147                thought_process: "I weighed a hash join first.".to_string(),
1148                ..Default::default()
1149            },
1150        }]
1151    }
1152
1153    fn exec_eval(cites: &[&str]) -> Vec<(String, Evaluation)> {
1154        vec![(
1155            "Candidate_A".to_string(),
1156            Evaluation {
1157                claim_assessments: cites
1158                    .iter()
1159                    .map(|c| crate::agents::ClaimAssessment {
1160                        claim: (*c).to_string(),
1161                        verdict: crate::agents::ClaimVerdict::Verified,
1162                        ..Default::default()
1163                    })
1164                    .collect(),
1165                ..Default::default()
1166            },
1167        )]
1168    }
1169
1170    #[test]
1171    fn exec_grounds_a_decorated_cite_to_the_exact_span() {
1172        let cands = exec_candidates();
1173        let mut evals = exec_eval(&["> \"sorts in O(n log n) time\""]);
1174
1175        let unresolved = ground_exec(&cands, &mut evals);
1176
1177        assert!(unresolved.is_empty());
1178        assert_eq!(
1179            evals[0].1.claim_assessments[0].claim,
1180            "sorts in O(n log n) time"
1181        );
1182    }
1183
1184    #[test]
1185    fn exec_reports_but_keeps_an_unresolvable_cite() {
1186        let cands = exec_candidates();
1187        let mut evals = exec_eval(&["sorts in O(n log n) time", "runs in constant time"]);
1188
1189        let unresolved = ground_exec(&cands, &mut evals);
1190
1191        assert_eq!(unresolved.len(), 1);
1192        assert_eq!(unresolved[0].cite, "runs in constant time");
1193        assert_eq!(
1194            evals[0].1.claim_assessments.len(),
1195            2,
1196            "Repair never drops a claim"
1197        );
1198    }
1199}
quorum_rs/agents/exec_agent.rs

quorum_rs/agents/
exec_agent.rs
