agentd/runtime/
worker.rs

1// SPDX-License-Identifier: AGPL-3.0-only
2//! The **turn worker**: a child process (`Role::Turn`) that
3//! runs ONE turn over the context slice the supervisor handed it — a root /
4//! conversation turn, a bounded `agent` step, or a structured `think` — and
5//! reports the transcript delta, the usage and the outcome (`TurnDone`).
6//!
7//! It calls the model and MCP tools itself; **internal** tools (memory, plan,
8//! subagents, sleep, finish, …) are **round-tripped** to the supervisor
9//! (`ToolRequest` → `ToolResult`) so state changes are made by the state
10//! owner. Before each model call it may ask for budget admission
11//! (`BudgetRequest` → `BudgetGrant`).
12//!
13//! The turn keeps tool results **structured** (`structuredContent` first,
14//! text-JSON second, text last), validates a schema'd final answer and
15//! re-asks on a miss, keeps a serializable transcript, estimates tokens for
16//! admission, and detects call loops. The loop body is [`run_turn`], generic
17//! over a [`Bridge`] so it is unit-testable in-process; [`run_turn_child`] is
18//! the process entry.
19
20use crate::context::{Msg, tokens};
21use crate::intel::client::IntelClient;
22use crate::jsonschema;
23use crate::mcp::client::McpClient;
24use crate::obs::log::Logger;
25use crate::subagent::control::{Up, send_up};
26use crate::subagent::protocol::{AgentMsg, SpawnPayload, TurnKind, TurnResult, TurnSpec};
27use crate::subagent::replies::{Replies, Reply};
28use crate::wire::intel::{Message, Request, ToolCall, Usage};
29use serde_json::{Value, json};
30use std::sync::Arc;
31use std::sync::atomic::{AtomicBool, Ordering};
32use std::time::{Duration, Instant};
33
34/// Default per-response completion cap.
35pub const DEFAULT_MAX_TOKENS_PER_CALL: u32 = 4096;
36/// Re-asks when a schema'd answer misses.
37pub const SCHEMA_REASKS: u32 = 2;
38/// The same tool call (name + args) this many times in one turn is a loop.
39pub const LOOP_REPEATS: usize = 4;
40/// Longest a single MCP tool call may take inside a turn.
41pub const TOOL_CALL_CAP: Duration = Duration::from_secs(600);
42
43/// A budget grant as the worker sees it.
44#[derive(Debug, Clone, PartialEq)]
45pub struct BudgetReply {
46    pub ok: bool,
47    pub wait_ms: u64,
48    pub model: Option<String>,
49    pub reason: Option<String>,
50}
51
52/// The supervisor link: internal-tool and budget round-trips + cancellation.
53pub trait Bridge {
54    /// Execute an internal tool via the supervisor. `None` = no answer
55    /// (cancelled / channel gone / deadline).
56    fn tool_request(
57        &mut self,
58        name: &str,
59        args: &Value,
60        deadline: Instant,
61    ) -> Option<(Value, bool)>;
62    /// Ask for budget admission. `None` = no answer.
63    fn budget_request(&mut self, estimate: u64, deadline: Instant) -> Option<BudgetReply>;
64    fn cancelled(&self) -> bool;
65    /// A progress event (liveness).
66    fn progress(&mut self, _event: &str, _fields: Value) {}
67}
68
69/// The process bridge: frames over the control channel.
70pub struct ChildBridge<'a> {
71    pub up: &'a Up,
72    pub replies: &'a Arc<Replies>,
73    pub cancel: &'a Arc<AtomicBool>,
74}
75
76impl Bridge for ChildBridge<'_> {
77    fn tool_request(
78        &mut self,
79        name: &str,
80        args: &Value,
81        deadline: Instant,
82    ) -> Option<(Value, bool)> {
83        let id = self.replies.next_id();
84        send_up(
85            self.up,
86            &AgentMsg::ToolRequest {
87                id,
88                name: name.to_string(),
89                args: args.clone(),
90            },
91        );
92        match self.replies.wait(id, deadline, self.cancel)? {
93            Reply::Tool { result, is_error } => Some((result, is_error)),
94            Reply::Budget { .. } => None,
95        }
96    }
97    fn budget_request(&mut self, estimate: u64, deadline: Instant) -> Option<BudgetReply> {
98        let id = self.replies.next_id();
99        send_up(self.up, &AgentMsg::BudgetRequest { id, estimate });
100        match self.replies.wait(id, deadline, self.cancel)? {
101            Reply::Budget {
102                ok,
103                wait_ms,
104                model,
105                reason,
106            } => Some(BudgetReply {
107                ok,
108                wait_ms,
109                model,
110                reason,
111            }),
112            Reply::Tool { .. } => None,
113        }
114    }
115    fn cancelled(&self) -> bool {
116        self.cancel.load(Ordering::Relaxed)
117    }
118    fn progress(&mut self, event: &str, fields: Value) {
119        send_up(
120            self.up,
121            &AgentMsg::Event {
122                event: event.to_string(),
123                fields,
124            },
125        );
126    }
127}
128
129/// How MCP tools are reached from the turn.
130pub trait McpCaller {
131    /// Call `tool` on `server` with `args` and extra `_meta`; `(content, is_error)`
132    /// where content is the structured result when the server gave one.
133    fn call(
134        &self,
135        server: &str,
136        tool: &str,
137        args: Value,
138        meta: Value,
139        timeout: Duration,
140    ) -> Result<(Value, bool), String>;
141}
142
143/// The connected MCP clients as an [`McpCaller`].
144pub struct McpClients<'a>(pub &'a [McpClient]);
145
146impl McpCaller for McpClients<'_> {
147    fn call(
148        &self,
149        server: &str,
150        tool: &str,
151        args: Value,
152        meta: Value,
153        timeout: Duration,
154    ) -> Result<(Value, bool), String> {
155        // Pace the call against the server's rate bucket before dialing out.
156        // The registry is per-process, seeded when THIS process connected its
157        // clients, so a worker only paces the calls it makes itself.
158        crate::mcp::pace::take(server)?;
159        let c = self
160            .0
161            .iter()
162            .find(|c| c.name() == server)
163            .ok_or_else(|| format!("mcp server {server:?} is not connected"))?;
164        let res = c
165            .call_tool_with_meta_within(tool, Some(args), meta, timeout)
166            .map_err(|e| e.to_string())?;
167        Ok((tool_result_value(&res), res.is_error()))
168    }
169}
170
171/// The value of an MCP tool result: `structuredContent`, else the text parsed
172/// as JSON, else the text.
173pub fn tool_result_value(res: &::mcp::wire::CallToolResult) -> Value {
174    if let Some(sc) = &res.structured_content {
175        return sc.clone();
176    }
177    let text = res.text();
178    serde_json::from_str::<Value>(&text).unwrap_or(Value::String(text))
179}
180
181/// The knobs of one turn (from the payload).
182#[derive(Debug, Clone)]
183pub struct TurnLimits {
184    pub max_rounds: u32,
185    pub max_tokens: u64,
186    pub deadline: Instant,
187    pub model: String,
188}
189
190/// Run ONE turn. Never panics; every failure is a `TurnResult` status.
191pub fn run_turn(
192    spec: &TurnSpec,
193    limits: &TurnLimits,
194    intel: &IntelClient,
195    mcp: &dyn McpCaller,
196    bridge: &mut dyn Bridge,
197    log: &Logger,
198) -> TurnResult {
199    let start = Instant::now();
200    let mut result = TurnResult {
201        status: "completed".into(),
202        ..Default::default()
203    };
204    let mut messages: Vec<Message> = Vec::with_capacity(spec.messages.len() + 2);
205    if !spec.system.trim().is_empty() {
206        messages.push(Message::System(spec.system.clone()));
207    }
208    messages.extend(spec.messages.iter().map(Msg::to_wire));
209    let tools = if spec.kind == TurnKind::Think {
210        Vec::new()
211    } else {
212        spec.tools.clone()
213    };
214    let mut model = limits.model.clone();
215    let max_rounds = if spec.max_rounds > 0 {
216        spec.max_rounds
217    } else if limits.max_rounds > 0 {
218        limits.max_rounds
219    } else {
220        u32::MAX
221    };
222    let max_tokens_per_call = if spec.max_tokens_per_call > 0 {
223        spec.max_tokens_per_call
224    } else {
225        DEFAULT_MAX_TOKENS_PER_CALL
226    };
227    let mut reasks = 0u32;
228    let mut recent_calls: Vec<String> = Vec::new();
229    let mut usage_total = Usage::default();
230    let mut delta: Vec<Msg> = Vec::new();
231    let mut last_text: Option<String> = None;
232    // The OTEL `invoke_agent` span for this turn — a no-op handle
233    // without the `otel` feature / an endpoint. `Option` so `finish` (which
234    // consumes the span) can be `take`n from the return-macro without a move
235    // across the loop.
236    let mut run_span = Some(crate::obs::otel::run_begin(
237        log.ctx().trace_id.as_deref(),
238        crate::obs::otel::now_unix_nanos(),
239    ));
240
241    log.info("turn.start", json!({"turn": spec.turn_id, "kind": spec.kind, "messages": messages.len(), "tools": tools.len()}));
242
243    macro_rules! finish_with {
244        ($status:expr) => {{
245            result.status = $status.to_string();
246            result.messages = delta;
247            result.usage = usage_total;
248            result.text = last_text.clone();
249            if let Some(rs) = run_span.take() {
250                rs.finish(&model, usage_total.input_tokens, usage_total.output_tokens, $status == "completed");
251            }
252            log.info("turn.done", json!({"turn": spec.turn_id, "status": result.status, "rounds": result.rounds, "tool_calls": result.tool_calls, "tokens_in": usage_total.input_tokens, "tokens_out": usage_total.output_tokens}));
253            return result;
254        }};
255    }
256
257    loop {
258        if bridge.cancelled() {
259            finish_with!("cancelled");
260        }
261        if Instant::now() >= limits.deadline {
262            finish_with!("deadline");
263        }
264        if result.rounds >= max_rounds {
265            finish_with!("exhausted_steps");
266        }
267        if limits.max_tokens > 0 && usage_total.total() >= limits.max_tokens {
268            finish_with!("exhausted_tokens");
269        }
270        // Budget admission: the supervisor owns the token budget, so estimate
271        // what this call will cost (prompt + tool schemas + the completion cap)
272        // and ask before spending it.
273        if spec.budget_admission {
274            let estimate: u64 = messages
275                .iter()
276                .map(|m| tokens::estimate(&render_len(m)) + tokens::MESSAGE_OVERHEAD)
277                .sum::<u64>()
278                + tools
279                    .iter()
280                    .map(|t| tokens::estimate(&t.name) + tokens::estimate_value(&t.input_schema))
281                    .sum::<u64>()
282                + max_tokens_per_call as u64;
283            loop {
284                match bridge.budget_request(estimate, limits.deadline) {
285                    None => finish_with!("cancelled"),
286                    Some(BudgetReply {
287                        ok: true, model: m, ..
288                    }) => {
289                        if let Some(m) = m
290                            && m != model
291                        {
292                            log.info(
293                                "turn.model_degraded",
294                                json!({"turn": spec.turn_id, "from": model, "to": m}),
295                            );
296                            model = m;
297                        }
298                        break;
299                    }
300                    Some(BudgetReply {
301                        ok: false,
302                        wait_ms,
303                        reason,
304                        ..
305                    }) => {
306                        if let Some(r) = reason {
307                            result.error = Some(format!("budget: {r}"));
308                            finish_with!(if r.contains("refus") {
309                                "refused"
310                            } else {
311                                "exhausted_tokens"
312                            });
313                        }
314                        let remaining = limits.deadline.saturating_duration_since(Instant::now());
315                        let wait = Duration::from_millis(wait_ms.max(50)).min(remaining);
316                        if wait.is_zero() {
317                            finish_with!("deadline");
318                        }
319                        log.info(
320                            "turn.budget_wait",
321                            json!({"turn": spec.turn_id, "wait_ms": wait.as_millis() as u64}),
322                        );
323                        std::thread::sleep(wait);
324                        if bridge.cancelled() {
325                            finish_with!("cancelled");
326                        }
327                    }
328                }
329            }
330        }
331        // The model call. Announce it first: the supervisor turns this into the
332        // display clients' live activity — `thinking` from here until the
333        // response lands.
334        bridge.progress(
335            "turn.think",
336            json!({"turn": spec.turn_id, "round": result.rounds + 1}),
337        );
338        let req = Request {
339            model: model.clone(),
340            messages: messages.clone(),
341            tools: tools.clone(),
342            max_tokens: max_tokens_per_call,
343            temperature: spec.temperature.or(Some(0.0)),
344        };
345        let chat_start = crate::obs::otel::now_unix_nanos();
346        let resp = match intel.complete(&req) {
347            Ok(r) => r,
348            Err(e) => {
349                if let Some(rs) = run_span.as_mut() {
350                    rs.record_chat(&model, 0, 0, false, chat_start);
351                }
352                result.error = Some(format!("intel: {e}"));
353                finish_with!("failed");
354            }
355        };
356        result.rounds += 1;
357        usage_total.input_tokens += resp.usage.input_tokens;
358        usage_total.output_tokens += resp.usage.output_tokens;
359        if let Some(rs) = run_span.as_mut() {
360            rs.record_chat(
361                &model,
362                resp.usage.input_tokens,
363                resp.usage.output_tokens,
364                true,
365                chat_start,
366            );
367        }
368        // Carry the round's token usage upward too — the supervisor attributes
369        // it to this turn's live activity (the instance counters are settled
370        // separately from the terminal usage, so this never double-counts).
371        bridge.progress("turn.round", json!({"turn": spec.turn_id, "round": result.rounds, "tool_calls": resp.tool_calls.len(), "tokens_in": resp.usage.input_tokens, "tokens_out": resp.usage.output_tokens}));
372        log.debug("turn.round", json!({"turn": spec.turn_id, "round": result.rounds, "tokens_in": resp.usage.input_tokens, "tokens_out": resp.usage.output_tokens, "tool_calls": resp.tool_calls.len()}));
373
374        if resp.wants_tools() {
375            if let Some(t) = &resp.text
376                && !t.is_empty()
377            {
378                last_text = Some(t.clone());
379            }
380            let assistant = Msg::assistant(resp.text.clone(), resp.tool_calls.clone());
381            messages.push(assistant.to_wire());
382            delta.push(assistant);
383            let mut finish_seen = None;
384            for (i, tc) in resp.tool_calls.iter().enumerate() {
385                if bridge.cancelled() {
386                    seal_unanswered(&mut delta, &resp.tool_calls[i..], "cancelled");
387                    finish_with!("cancelled");
388                }
389                result.tool_calls += 1;
390                // Loop detection: the same call repeated.
391                let sig = format!("{}:{}", tc.name, tc.arguments);
392                recent_calls.push(sig.clone());
393                if recent_calls.iter().filter(|s| **s == sig).count() >= LOOP_REPEATS {
394                    result.error = Some(format!(
395                        "the model repeated {} with identical arguments {LOOP_REPEATS} times",
396                        tc.name
397                    ));
398                    seal_unanswered(&mut delta, &resp.tool_calls[i..], "loop_detected");
399                    finish_with!("loop_detected");
400                }
401                let call_start = Instant::now();
402                let tool_span_start = crate::obs::otel::now_unix_nanos();
403                // The one signal that says WHAT it is doing right now — for MCP
404                // tools the supervisor never sees the call otherwise (the child
405                // holds its own MCP connections).
406                bridge.progress(
407                    "turn.tool",
408                    json!({"turn": spec.turn_id, "tool": tc.name, "i": i + 1, "of": resp.tool_calls.len()}),
409                );
410                let (content, is_error) = execute_call(spec, limits, tc, i, mcp, bridge, log);
411                if let Some(rs) = run_span.as_mut() {
412                    rs.record_tool(&tc.name, !is_error, tool_span_start);
413                }
414                log.info("tool.result", json!({"turn": spec.turn_id, "tool": tc.name, "is_error": is_error, "ms": call_start.elapsed().as_millis() as u64}));
415                let msg = Msg::tool(tc.id.clone(), tc.name.clone(), content, is_error);
416                messages.push(msg.to_wire());
417                delta.push(msg);
418                if tc.name == "finish" && !is_error {
419                    finish_seen = Some(tc.arguments.clone());
420                }
421            }
422            if let Some(f) = finish_seen {
423                result.finish = Some(f);
424                last_text = last_text.or_else(|| {
425                    result
426                        .finish
427                        .as_ref()
428                        .and_then(|f| f.get("output"))
429                        .map(|o| match o {
430                            Value::String(s) => s.clone(),
431                            other => other.to_string(),
432                        })
433                });
434                finish_with!("completed");
435            }
436            continue;
437        }
438
439        // A final answer.
440        let text = resp
441            .text
442            .clone()
443            .or_else(|| last_text.clone())
444            .unwrap_or_default();
445        let assistant = Msg::assistant(Some(text.clone()), Vec::new());
446        // Structured answers: parse + validate, re-ask on a miss.
447        let wants_object = spec.kind == TurnKind::Think || spec.output_schema.is_some();
448        if wants_object {
449            match parse_json_answer(&text) {
450                Ok(v) => {
451                    let check = match &spec.output_schema {
452                        Some(schema) => {
453                            jsonschema::validate(schema, &v).map_err(|e| jsonschema::explain(&e))
454                        }
455                        None => Ok(()),
456                    };
457                    match check {
458                        Ok(()) => {
459                            messages.push(assistant.to_wire());
460                            delta.push(assistant);
461                            last_text = Some(text);
462                            result.value = Some(v);
463                            finish_with!("completed");
464                        }
465                        Err(e) => {
466                            if reasks < SCHEMA_REASKS {
467                                reasks += 1;
468                                log.info("turn.reask", json!({"turn": spec.turn_id, "reason": e}));
469                                messages.push(assistant.to_wire());
470                                delta.push(assistant);
471                                let ask = Msg::user(
472                                    format!(
473                                        "Your answer did not match the required schema: {e}. Reply again with ONLY one JSON object that matches the schema."
474                                    ),
475                                    None,
476                                );
477                                messages.push(ask.to_wire());
478                                delta.push(ask);
479                                continue;
480                            }
481                            result.error =
482                                Some(format!("answer does not match the output schema: {e}"));
483                            messages.push(assistant.to_wire());
484                            delta.push(assistant);
485                            last_text = Some(text);
486                            finish_with!("failed");
487                        }
488                    }
489                }
490                Err(e) => {
491                    if reasks < SCHEMA_REASKS {
492                        reasks += 1;
493                        log.info("turn.reask", json!({"turn": spec.turn_id, "reason": e}));
494                        messages.push(assistant.to_wire());
495                        delta.push(assistant);
496                        let ask = Msg::user(format!("{e}. Reply with ONLY one JSON object."), None);
497                        messages.push(ask.to_wire());
498                        delta.push(ask);
499                        continue;
500                    }
501                    result.error = Some(e);
502                    delta.push(assistant);
503                    last_text = Some(text);
504                    finish_with!("failed");
505                }
506            }
507        }
508        delta.push(assistant);
509        last_text = Some(text);
510        let _ = start;
511        finish_with!("completed");
512    }
513}
514
515fn render_len(m: &Message) -> String {
516    match m {
517        Message::System(s) | Message::User(s) => s.clone(),
518        Message::Assistant { text, tool_calls } => format!(
519            "{}{}",
520            text.as_deref().unwrap_or(""),
521            tool_calls
522                .iter()
523                .map(|c| c.arguments.to_string())
524                .collect::<String>()
525        ),
526        Message::ToolResult { content, .. } => content.clone(),
527    }
528}
529
530/// Answer the tool calls the turn never got to run, so the transcript delta it
531/// reports is self-consistent.
532///
533/// Every provider dialect requires one tool result per `tool_calls` id on the
534/// preceding assistant message, and the tool loop has exits that fire BETWEEN
535/// pushing that assistant message and pushing its results (cancellation between
536/// calls, loop detection on the offending call). The delta is appended verbatim
537/// to the DURABLE context, so an unanswered id is not one lost result: every
538/// later turn and every restart replays the same malformed context and the
539/// provider rejects it with a fatal 400 forever — reported as a retryable
540/// `intel:` failure (exit `INTEL_UNAVAILABLE`), so an external scheduler keeps
541/// retrying a request agentd itself malformed. The synthetic result is an error
542/// result, the same shape [`execute_call`] uses when the supervisor never
543/// answers, so the model sees the call did not happen rather than a made-up
544/// success.
545fn seal_unanswered(delta: &mut Vec<Msg>, unanswered: &[ToolCall], status: &str) {
546    for tc in unanswered {
547        delta.push(Msg::tool(
548            tc.id.clone(),
549            tc.name.clone(),
550            Value::String(format!(
551                "{}: not executed — the turn ended ({status}) before this call ran",
552                tc.name
553            )),
554            true,
555        ));
556    }
557}
558
559/// Dispatch one tool call: internal (round-trip) → MCP (own call) → code.
560fn execute_call(
561    spec: &TurnSpec,
562    limits: &TurnLimits,
563    tc: &ToolCall,
564    index: usize,
565    mcp: &dyn McpCaller,
566    bridge: &mut dyn Bridge,
567    log: &Logger,
568) -> (Value, bool) {
569    let name = tc.name.as_str();
570    log.info(
571        "tool.call",
572        json!({"turn": spec.turn_id, "tool": name, "id": tc.id}),
573    );
574    if spec.internal.iter().any(|n| n == name) {
575        return match bridge.tool_request(name, &tc.arguments, limits.deadline) {
576            Some(r) => r,
577            None => (
578                Value::String(format!(
579                    "{name}: no answer from the supervisor (cancelled or deadline)"
580                )),
581                true,
582            ),
583        };
584    }
585    if let Some((server, tool)) = spec.mcp_routes.get(name) {
586        let remaining = limits.deadline.saturating_duration_since(Instant::now());
587        let timeout = remaining.min(TOOL_CALL_CAP).max(Duration::from_millis(100));
588        let mut meta = spec.tool_meta.clone().unwrap_or_else(|| json!({}));
589        if !spec.idempotency_prefix.is_empty() {
590            meta["agent/idempotency_key"] = json!(format!(
591                "{}#{}.{}",
592                spec.idempotency_prefix, spec.turn_id, index
593            ));
594        }
595        return match mcp.call(server, tool, tc.arguments.clone(), meta, timeout) {
596            Ok(r) => r,
597            Err(e) => (Value::String(format!("tool transport error: {e}")), true),
598        };
599    }
600    if let Some(r) = crate::tools::call(name, &tc.arguments) {
601        return match r {
602            Ok(v) => (v, false),
603            Err(e) => (Value::String(e), true),
604        };
605    }
606    (Value::String(format!("error: no such tool '{name}'")), true)
607}
608
609/// Parse a model's JSON answer tolerantly (fences, prose around the object).
610pub fn parse_json_answer(answer: &str) -> Result<Value, String> {
611    let t = answer.trim();
612    if let Ok(v) = serde_json::from_str::<Value>(t) {
613        return Ok(v);
614    }
615    let t2 = t
616        .trim_start_matches("```json")
617        .trim_start_matches("```")
618        .trim_end_matches("```")
619        .trim();
620    if let Ok(v) = serde_json::from_str::<Value>(t2) {
621        return Ok(v);
622    }
623    if let (Some(a), Some(b)) = (t.find('{'), t.rfind('}'))
624        && b > a
625        && let Ok(v) = serde_json::from_str::<Value>(&t[a..=b])
626    {
627        return Ok(v);
628    }
629    if let (Some(a), Some(b)) = (t.find('['), t.rfind(']'))
630        && b > a
631        && let Ok(v) = serde_json::from_str::<Value>(&t[a..=b])
632    {
633        return Ok(v);
634    }
635    Err(format!(
636        "the answer was not valid JSON: {}",
637        t.chars().take(120).collect::<String>()
638    ))
639}
640
641/// The process entry for a `Role::Turn` child: run the turn, report
642/// `Usage` + `TurnDone`, return the exit code.
643pub fn run_turn_child(
644    payload: &SpawnPayload,
645    intel: &IntelClient,
646    servers: &[McpClient],
647    up: &Up,
648    cancel: &Arc<AtomicBool>,
649    replies: &Arc<Replies>,
650    log: &Logger,
651) -> i32 {
652    let Some(spec) = payload.turn.as_deref() else {
653        send_up(
654            up,
655            &AgentMsg::Failed {
656                error: "role is turn but the payload carries no turn spec".into(),
657            },
658        );
659        return crate::exit::USAGE;
660    };
661    let limits = TurnLimits {
662        max_rounds: if spec.max_rounds > 0 {
663            spec.max_rounds
664        } else {
665            payload.limits.max_steps
666        },
667        max_tokens: payload.limits.max_tokens,
668        deadline: Instant::now() + Duration::from_millis(payload.limits.deadline_ms.max(1)),
669        model: payload.intelligence.model.clone().unwrap_or_default(),
670    };
671    let mut bridge = ChildBridge {
672        up,
673        replies,
674        cancel,
675    };
676    let result = run_turn(spec, &limits, intel, &McpClients(servers), &mut bridge, log);
677    let code = match result.status.as_str() {
678        "completed" => crate::exit::SUCCESS,
679        "failed"
680            if result
681                .error
682                .as_deref()
683                .is_some_and(|e| e.starts_with("intel:")) =>
684        {
685            crate::exit::INTEL_UNAVAILABLE
686        }
687        "failed" | "cancelled" => crate::exit::GENERIC,
688        "refused" => crate::exit::REFUSED,
689        "exhausted_steps" | "exhausted_tokens" => crate::exit::BUDGET,
690        "deadline" => crate::exit::DEADLINE,
691        _ => crate::exit::PARTIAL,
692    };
693    send_up(up, &AgentMsg::Usage(result.usage));
694    send_up(
695        up,
696        &AgentMsg::TurnDone {
697            turn: Box::new(result),
698        },
699    );
700    code
701}
702
703#[cfg(test)]
704mod tests {
705    use super::*;
706    use crate::obs::log::{Comp, Level, LogCtx};
707    use crate::wire::intel::ToolDef;
708    use std::collections::BTreeMap;
709
710    fn log() -> Logger {
711        Logger::new(
712            LogCtx {
713                run_id: "t".into(),
714                agent_id: "0".into(),
715                agent_path: "0".into(),
716                comp: Comp::Agent,
717                pid: 0,
718                trace_id: None,
719            },
720            Level::Warn,
721        )
722    }
723
724    /// Start the in-process mock LLM with a playbook file; returns the client.
725    fn mock_intel(playbook: &Value) -> IntelClient {
726        let dir = std::env::temp_dir();
727        // pid + a process-wide counter: tests run in parallel threads, and a
728        // clock-derived suffix COLLIDED under load — two tests sharing one
729        // addr file each read whichever mock announced last, and one turn
730        // got the other's playbook.
731        static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
732        let n = std::process::id() as u64 * 100_000
733            + SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
734        let pb = dir.join(format!("agentd-worker-pb-{n}.json"));
735        std::fs::write(&pb, playbook.to_string()).unwrap();
736        let addr_file = dir.join(format!("agentd-worker-mock-{n}.addr"));
737        let _ = std::fs::remove_file(&addr_file);
738        let af = addr_file.to_string_lossy().to_string();
739        let script = format!("file:{}", pb.to_string_lossy());
740        std::thread::spawn(move || crate::intel::mock::run(&af, &script));
741        let deadline = Instant::now() + Duration::from_secs(5);
742        while !addr_file.exists() {
743            assert!(Instant::now() < deadline, "mock never announced");
744            std::thread::sleep(Duration::from_millis(5));
745        }
746        let addr = std::fs::read_to_string(&addr_file).unwrap();
747        IntelClient::from_parts(&format!("http://{}", addr.trim()), None).unwrap()
748    }
749
750    struct FakeBridge {
751        calls: Vec<(String, Value)>,
752        answers: BTreeMap<String, (Value, bool)>,
753        budget: Vec<BudgetReply>,
754        cancel: bool,
755    }
756    impl Bridge for FakeBridge {
757        fn tool_request(&mut self, name: &str, args: &Value, _d: Instant) -> Option<(Value, bool)> {
758            self.calls.push((name.to_string(), args.clone()));
759            Some(
760                self.answers
761                    .get(name)
762                    .cloned()
763                    .unwrap_or((json!({"ok": true}), false)),
764            )
765        }
766        fn budget_request(&mut self, _e: u64, _d: Instant) -> Option<BudgetReply> {
767            if self.budget.is_empty() {
768                Some(BudgetReply {
769                    ok: true,
770                    wait_ms: 0,
771                    model: None,
772                    reason: None,
773                })
774            } else {
775                Some(self.budget.remove(0))
776            }
777        }
778        fn cancelled(&self) -> bool {
779            self.cancel
780        }
781    }
782    struct FakeMcp;
783    impl McpCaller for FakeMcp {
784        fn call(
785            &self,
786            server: &str,
787            tool: &str,
788            args: Value,
789            meta: Value,
790            _t: Duration,
791        ) -> Result<(Value, bool), String> {
792            if server == "down" {
793                return Err("connection refused".into());
794            }
795            Ok((
796                json!({"server": server, "tool": tool, "args": args, "idem": meta["agent/idempotency_key"]}),
797                tool == "boom",
798            ))
799        }
800    }
801
802    fn limits() -> TurnLimits {
803        TurnLimits {
804            max_rounds: 8,
805            max_tokens: 0,
806            deadline: Instant::now() + Duration::from_secs(20),
807            model: "mock".into(),
808        }
809    }
810
811    fn spec(kind: TurnKind) -> TurnSpec {
812        TurnSpec {
813            kind,
814            system: "You are a test agent.".into(),
815            messages: vec![Msg::user("do the thing", None)],
816            tools: vec![
817                ToolDef {
818                    name: "memory.set".into(),
819                    description: "".into(),
820                    input_schema: json!({"type": "object"}),
821                },
822                ToolDef {
823                    name: "fs.read".into(),
824                    description: "".into(),
825                    input_schema: json!({"type": "object"}),
826                },
827                ToolDef {
828                    name: "finish".into(),
829                    description: "".into(),
830                    input_schema: json!({"type": "object"}),
831                },
832            ],
833            internal: vec!["memory.set".into(), "finish".into()],
834            mcp_routes: [
835                (
836                    "fs.read".to_string(),
837                    ("fs".to_string(), "read".to_string()),
838                ),
839                ("boom".to_string(), ("fs".to_string(), "boom".to_string())),
840            ]
841            .into_iter()
842            .collect(),
843            idempotency_prefix: "ctx/1".into(),
844            turn_id: "t1".into(),
845            ..Default::default()
846        }
847    }
848
849    #[test]
850    fn a_turn_round_trips_internal_tools_calls_mcp_itself_and_finishes() {
851        // The playbook indexes turns by the number of tool results in the
852        // transcript: one round with three calls, then the final answer.
853        let intel = mock_intel(&json!({"turns": [
854            {"tool_calls": [{"name": "memory.set", "arguments": {"key": "k", "value": 1}}, {"name": "fs.read", "arguments": {"path": "/x"}}, {"name": "boom", "arguments": {}}]},
855            {"content": "unused"},
856            {"content": "unused"},
857            {"content": "all done", "usage": {"prompt_tokens": 100, "completion_tokens": 20}}
858        ]}));
859        let mut bridge = FakeBridge {
860            calls: vec![],
861            answers: BTreeMap::new(),
862            budget: vec![],
863            cancel: false,
864        };
865        let r = run_turn(
866            &spec(TurnKind::Turn),
867            &limits(),
868            &intel,
869            &FakeMcp,
870            &mut bridge,
871            &log(),
872        );
873        assert_eq!(r.status, "completed", "{:?}", r.error);
874        assert_eq!(r.text.as_deref(), Some("all done"));
875        assert_eq!(r.rounds, 2);
876        assert_eq!(r.tool_calls, 3);
877        assert_eq!(
878            bridge.calls,
879            vec![("memory.set".to_string(), json!({"key": "k", "value": 1}))],
880            "only the internal tool round-tripped"
881        );
882        // Delta: assistant, tool, tool, tool, assistant.
883        assert_eq!(r.messages.len(), 5);
884        match &r.messages[2] {
885            Msg::Tool {
886                name,
887                content,
888                is_error,
889                ..
890            } => {
891                assert_eq!(name, "fs.read");
892                assert!(!is_error);
893                assert_eq!(content["server"], json!("fs"));
894                assert_eq!(
895                    content["idem"],
896                    json!("ctx/1#t1.1"),
897                    "idempotency key stamped per call"
898                );
899            }
900            other => panic!("{other:?}"),
901        }
902        assert!(matches!(&r.messages[3], Msg::Tool { is_error: true, .. }));
903        assert_eq!(r.usage.input_tokens, 11 + 100);
904        assert_eq!(r.usage.output_tokens, 7 + 20);
905    }
906
907    #[test]
908    fn finish_ends_the_turn_and_a_think_returns_an_object_with_reasks() {
909        let intel = mock_intel(&json!({"turns": [
910            {"tool_calls": [{"name": "finish", "arguments": {"status": "completed", "output": {"n": 3}}}]},
911            {"content": "never reached"}
912        ]}));
913        let mut bridge = FakeBridge {
914            calls: vec![],
915            answers: BTreeMap::new(),
916            budget: vec![],
917            cancel: false,
918        };
919        let r = run_turn(
920            &spec(TurnKind::Agent),
921            &limits(),
922            &intel,
923            &FakeMcp,
924            &mut bridge,
925            &log(),
926        );
927        assert_eq!(r.status, "completed");
928        assert_eq!(r.finish.as_ref().unwrap()["output"]["n"], json!(3));
929        assert_eq!(r.rounds, 1);
930        // A think: the first answer misses the schema, the re-ask fixes it.
931        let intel = mock_intel(&json!({"turns": [
932            {"content": "Sure: {\"intent\": \"maybe\"}"},
933            {"content": "```json\n{\"intent\": \"task\", \"needs_plan\": true}\n```"}
934        ]}));
935        let mut s = spec(TurnKind::Think);
936        s.output_schema = Some(
937            json!({"type": "object", "properties": {"intent": {"enum": ["chat", "task"]}, "needs_plan": {"type": "boolean"}}, "required": ["intent"]}),
938        );
939        // The mock indexes turns by tool results; a think has none, so it always
940        // answers turn 0 unless a match rule catches the re-ask.
941        let intel2 = mock_intel(
942            &json!({"turns": [{"content": "Sure: {\"intent\": \"maybe\"}"}], "match": [{"when_contains": "did not match the required schema", "content": {"intent": "task", "needs_plan": true}}]}),
943        );
944        drop(intel);
945        let r = run_turn(&s, &limits(), &intel2, &FakeMcp, &mut bridge, &log());
946        assert_eq!(r.status, "completed", "{:?}", r.error);
947        assert_eq!(r.value.as_ref().unwrap()["intent"], json!("task"));
948        assert_eq!(r.rounds, 2);
949        // Never valid ⇒ failed after the re-asks.
950        let intel3 = mock_intel(&json!({"turns": [{"content": "not json at all"}]}));
951        let r = run_turn(&s, &limits(), &intel3, &FakeMcp, &mut bridge, &log());
952        assert_eq!(r.status, "failed");
953        assert_eq!(r.rounds, 1 + SCHEMA_REASKS);
954    }
955
956    #[test]
957    fn loops_budget_waits_and_cancel_are_bounded() {
958        // The model repeats the same call forever ⇒ loop_detected.
959        let intel = mock_intel(
960            &json!({"turns": [{"tool_calls": [{"name": "fs.read", "arguments": {"path": "/same"}}]}]}),
961        );
962        let mut bridge = FakeBridge {
963            calls: vec![],
964            answers: BTreeMap::new(),
965            budget: vec![],
966            cancel: false,
967        };
968        let r = run_turn(
969            &spec(TurnKind::Turn),
970            &limits(),
971            &intel,
972            &FakeMcp,
973            &mut bridge,
974            &log(),
975        );
976        assert_eq!(r.status, "loop_detected");
977        assert_eq!(r.tool_calls, LOOP_REPEATS as u32);
978        // Round cap.
979        let mut l = limits();
980        l.max_rounds = 2;
981        let mut s = spec(TurnKind::Turn);
982        s.mcp_routes
983            .insert("fs.read".into(), ("down".into(), "read".into()));
984        let intel = mock_intel(&json!({"turns": [
985            {"tool_calls": [{"name": "fs.read", "arguments": {"path": "/a"}}]},
986            {"tool_calls": [{"name": "fs.read", "arguments": {"path": "/b"}}]},
987            {"tool_calls": [{"name": "fs.read", "arguments": {"path": "/c"}}]}
988        ]}));
989        let r = run_turn(&s, &l, &intel, &FakeMcp, &mut bridge, &log());
990        assert_eq!(r.status, "exhausted_steps");
991        assert!(
992            matches!(&r.messages[1], Msg::Tool { is_error: true, content, .. } if content.as_str().unwrap().contains("transport error"))
993        );
994        // Budget: one wait then ok with a degraded model; then a refusal.
995        let intel = mock_intel(&json!({"turns": [{"content": "ok"}]}));
996        let mut s = spec(TurnKind::Turn);
997        s.budget_admission = true;
998        let mut b = FakeBridge {
999            calls: vec![],
1000            answers: BTreeMap::new(),
1001            budget: vec![
1002                BudgetReply {
1003                    ok: false,
1004                    wait_ms: 10,
1005                    model: None,
1006                    reason: None,
1007                },
1008                BudgetReply {
1009                    ok: true,
1010                    wait_ms: 0,
1011                    model: Some("cheap".into()),
1012                    reason: None,
1013                },
1014            ],
1015            cancel: false,
1016        };
1017        let r = run_turn(&s, &limits(), &intel, &FakeMcp, &mut b, &log());
1018        assert_eq!(r.status, "completed");
1019        let mut b = FakeBridge {
1020            calls: vec![],
1021            answers: BTreeMap::new(),
1022            budget: vec![BudgetReply {
1023                ok: false,
1024                wait_ms: 0,
1025                model: None,
1026                reason: Some("budget refused: window exhausted".into()),
1027            }],
1028            cancel: false,
1029        };
1030        let r = run_turn(&s, &limits(), &intel, &FakeMcp, &mut b, &log());
1031        assert_eq!(r.status, "refused");
1032        // Cancel before the first call.
1033        let mut b = FakeBridge {
1034            calls: vec![],
1035            answers: BTreeMap::new(),
1036            budget: vec![],
1037            cancel: true,
1038        };
1039        let r = run_turn(
1040            &spec(TurnKind::Turn),
1041            &limits(),
1042            &intel,
1043            &FakeMcp,
1044            &mut b,
1045            &log(),
1046        );
1047        assert_eq!(r.status, "cancelled");
1048        assert_eq!(r.rounds, 0);
1049        // A dead intelligence is `failed` with an intel error.
1050        let dead = IntelClient::from_parts("http://127.0.0.1:9", None).unwrap();
1051        let r = run_turn(
1052            &spec(TurnKind::Turn),
1053            &limits(),
1054            &dead,
1055            &FakeMcp,
1056            &mut b,
1057            &log(),
1058        );
1059        assert_eq!(r.status, "cancelled", "cancel wins first");
1060        b.cancel = false;
1061        let r = run_turn(
1062            &spec(TurnKind::Turn),
1063            &limits(),
1064            &dead,
1065            &FakeMcp,
1066            &mut b,
1067            &log(),
1068        );
1069        assert_eq!(r.status, "failed");
1070        assert!(r.error.as_deref().unwrap().starts_with("intel:"));
1071    }
1072
1073    #[test]
1074    fn an_exit_inside_the_tool_loop_answers_every_call_it_persisted() {
1075        // Cancellation landing BETWEEN two calls of one assistant message (the
1076        // other early exit, loop detection, is covered end to end in
1077        // `wedged_context_e2e`). The delta is appended verbatim to the DURABLE
1078        // context, so leaving an id unanswered would malform that context for
1079        // every later turn and every restart — not just lose one result.
1080        struct CancelAfterOneCall(usize);
1081        impl Bridge for CancelAfterOneCall {
1082            fn tool_request(&mut self, _n: &str, _a: &Value, _d: Instant) -> Option<(Value, bool)> {
1083                self.0 += 1;
1084                Some((json!({"ok": true}), false))
1085            }
1086            fn budget_request(&mut self, _e: u64, _d: Instant) -> Option<BudgetReply> {
1087                None
1088            }
1089            fn cancelled(&self) -> bool {
1090                self.0 > 0
1091            }
1092        }
1093        // Three DISTINCT calls so loop detection does not fire first.
1094        let intel = mock_intel(&json!({"turns": [{"tool_calls": [
1095            {"name": "memory.set", "arguments": {"key": "a"}},
1096            {"name": "memory.set", "arguments": {"key": "b"}},
1097            {"name": "memory.set", "arguments": {"key": "c"}}
1098        ]}]}));
1099        let mut bridge = CancelAfterOneCall(0);
1100        let r = run_turn(
1101            &spec(TurnKind::Turn),
1102            &limits(),
1103            &intel,
1104            &FakeMcp,
1105            &mut bridge,
1106            &log(),
1107        );
1108        assert_eq!(r.status, "cancelled");
1109        // assistant + the one executed result + two sealed ones.
1110        assert_eq!(r.messages.len(), 4, "{:?}", r.messages);
1111        let answered: Vec<&str> = r
1112            .messages
1113            .iter()
1114            .filter_map(|m| match m {
1115                Msg::Tool { id, .. } => Some(id.as_str()),
1116                _ => None,
1117            })
1118            .collect();
1119        match &r.messages[0] {
1120            Msg::Assistant { tool_calls, .. } => {
1121                assert_eq!(tool_calls.len(), 3);
1122                for tc in tool_calls {
1123                    assert!(
1124                        answered.contains(&tc.id.as_str()),
1125                        "tool_call {} has no result: {:?}",
1126                        tc.id,
1127                        r.messages
1128                    );
1129                }
1130            }
1131            other => panic!("{other:?}"),
1132        }
1133        assert!(
1134            matches!(&r.messages[3], Msg::Tool { is_error: true, content, .. }
1135                if content.as_str().unwrap_or_default().contains("not executed")),
1136            "the sealed result says the call did not happen: {:?}",
1137            r.messages[3]
1138        );
1139    }
1140
1141    #[test]
1142    fn json_answers_are_parsed_tolerantly() {
1143        assert_eq!(parse_json_answer("{\"a\":1}").unwrap(), json!({"a": 1}));
1144        assert_eq!(
1145            parse_json_answer("```json\n{\"a\":1}\n```").unwrap(),
1146            json!({"a": 1})
1147        );
1148        assert_eq!(
1149            parse_json_answer("Here: {\"a\":1}. Done.").unwrap(),
1150            json!({"a": 1})
1151        );
1152        assert_eq!(parse_json_answer("[1,2]").unwrap(), json!([1, 2]));
1153        assert!(parse_json_answer("nope").is_err());
1154        let res = ::mcp::wire::CallToolResult {
1155            content: vec![json!({"type": "text", "text": "{\"x\": 2}"})],
1156            is_error: None,
1157            structured_content: None,
1158        };
1159        assert_eq!(tool_result_value(&res), json!({"x": 2}));
1160        let res = ::mcp::wire::CallToolResult {
1161            content: vec![json!({"type": "text", "text": "plain"})],
1162            is_error: None,
1163            structured_content: Some(json!({"s": 1})),
1164        };
1165        assert_eq!(tool_result_value(&res), json!({"s": 1}));
1166    }
1167}
agentd/runtime/worker.rs

agentd/runtime/
worker.rs
