kcode_agent_runtime/
lib.rs

1//! Provider-neutral agent loops over `kcode-intelligence-router`.
2
3#![deny(missing_docs)]
4#![forbid(unsafe_code)]
5
6use std::{future::Future, pin::Pin, time::Duration};
7
8use anyhow::{Context, ensure};
9use kcode_codex_runtime_v2::{
10    AgentEvent, AgentRequest, DynamicTool, DynamicToolCall, ReasoningEffort, TokenUsage, ToolResult,
11};
12use kcode_intelligence_router::{AgentProvider, Intelligence, ResolvedAgentModel, UsageReceipt};
13use serde_json::{Value, json};
14use sha2::{Digest, Sha256};
15use uuid::Uuid;
16
17/// Default maximum provider calls for primary and delegated agent loops.
18pub const DEFAULT_ROUND_LIMIT: u64 = 250;
19const PROTOCOL_TOKEN_RESERVE: u64 = 4_096;
20const SUPERSEDED_TOOL_OUTPUT: &str = "[Tool output was displayed here, but has since been updated and now appears elsewhere in the context]";
21const REMOVED_TOOL_OUTPUT: &str = "[Tool output was displayed here, but is no longer current]";
22
23/// A boxed asynchronous host operation.
24pub type HostFuture<'a, T> = Pin<Box<dyn Future<Output = anyhow::Result<T>> + Send + 'a>>;
25
26/// One application tool call requested by a subagent.
27#[derive(Clone, Debug, PartialEq)]
28pub struct ToolCall {
29    /// Exact application tool name.
30    pub name: String,
31    /// Tool arguments.
32    pub arguments: Value,
33}
34
35/// One complete, typed audit fact selected by the subagent runtime.
36#[derive(Clone, Debug, PartialEq)]
37pub enum AuditEvent {
38    /// Immutable starting context and resolved provider capacity.
39    Started {
40        /// Parent operation that causally owns the subagent.
41        parent_operation_id: Uuid,
42        /// Caller-selected model identifier.
43        model: String,
44        /// Exact provider model.
45        provider_model: String,
46        /// Resolved provider transport.
47        provider: AgentProvider,
48        /// Total provider context window.
49        context_window_tokens: u64,
50        /// Maximum permitted input.
51        max_input_tokens: u64,
52        /// Ordered immutable starting context.
53        context: Vec<String>,
54        /// Focused subagent task.
55        task: String,
56        /// Opaque application metadata supplied at launch.
57        host: Value,
58    },
59    /// Exact input identity submitted for one provider round.
60    InferenceSubmitted {
61        /// Parent operation that causally owns the subagent.
62        parent_operation_id: Uuid,
63        /// One-based subagent round.
64        round: u64,
65        /// SHA-256 hash of the exact provider input.
66        manifest_hash: String,
67        /// Runtime input estimate including protocol reserve.
68        estimated_input_tokens: u64,
69    },
70    /// Application tool invocation requested by the provider.
71    ToolCall {
72        /// Parent operation that causally owns the subagent.
73        parent_operation_id: Uuid,
74        /// Exact application tool name.
75        name: String,
76        /// Exact tool arguments.
77        arguments: Value,
78    },
79    /// Complete application tool result retained for audit.
80    ToolResult {
81        /// Parent operation that causally owns the subagent.
82        parent_operation_id: Uuid,
83        /// Exact application tool name.
84        name: String,
85        /// Whether execution itself succeeded.
86        ok: bool,
87        /// Whether the resulting context projection fit.
88        projection_accepted: bool,
89        /// Exact application result.
90        result: String,
91    },
92    /// Canonical accounting for one completed or interrupted provider round.
93    ProviderReceipt {
94        /// Parent operation that causally owns the subagent.
95        parent_operation_id: Uuid,
96        /// One-based subagent round.
97        round: u64,
98        /// SHA-256 hash of the exact provider input.
99        manifest_hash: String,
100        /// Provider-native cumulative usage retained for exact audit detail.
101        usage: Option<kcode_codex_runtime_v2::TokenUsage>,
102        /// Canonical durable receipt written by the intelligence router.
103        receipt: Box<UsageReceipt>,
104    },
105    /// Final non-empty subagent answer.
106    Completed {
107        /// Parent operation that causally owns the subagent.
108        parent_operation_id: Uuid,
109        /// Caller-selected model identifier.
110        model: String,
111        /// Terminal response returned to the application.
112        response: String,
113    },
114}
115
116/// One replaceable state section rendered into every later context slice.
117#[derive(Clone, Debug, Eq, PartialEq)]
118pub struct StateUpdate {
119    /// Stable state identity. A later update with this key replaces the prior text.
120    pub key: String,
121    /// Current rendered state, or `None` to remove it.
122    pub text: Option<String>,
123}
124
125/// Result returned by the application after one tool or capture operation.
126#[derive(Clone, Debug, PartialEq)]
127pub struct ToolOutcome {
128    /// Exact result retained for audit.
129    pub text: String,
130    /// Whether the operation succeeded.
131    pub ok: bool,
132    /// Replaceable state made current by the operation.
133    pub state_updates: Vec<StateUpdate>,
134    /// State identities whose complete values this result displays.
135    ///
136    /// A later update to one of these identities replaces this retained result
137    /// with a concise supersession marker while preserving the exact audit
138    /// result.
139    pub displayed_state_keys: Vec<String>,
140    /// Opaque application token requesting a tool-free freeform output capture.
141    pub capture: Option<Value>,
142}
143
144impl ToolOutcome {
145    /// Constructs a simple successful result.
146    pub fn success(text: impl Into<String>) -> Self {
147        Self {
148            text: text.into(),
149            ok: true,
150            state_updates: Vec::new(),
151            displayed_state_keys: Vec::new(),
152            capture: None,
153        }
154    }
155
156    /// Constructs a simple failed result.
157    pub fn failure(text: impl Into<String>) -> Self {
158        Self {
159            text: text.into(),
160            ok: false,
161            state_updates: Vec::new(),
162            displayed_state_keys: Vec::new(),
163            capture: None,
164        }
165    }
166}
167
168/// Read-only capacity view supplied while a host evaluates a tool.
169#[derive(Clone)]
170pub struct ContextBudget {
171    projection: Projection,
172    max_input_tokens: u64,
173}
174
175impl ContextBudget {
176    /// Current estimated input tokens, including protocol reserve.
177    pub fn estimated_tokens(&self) -> u64 {
178        self.projection.estimated_tokens()
179    }
180
181    /// Maximum permitted input tokens.
182    pub fn max_input_tokens(&self) -> u64 {
183        self.max_input_tokens
184    }
185
186    /// Returns whether replacing one projected state would fit.
187    pub fn fits_state(&self, key: impl Into<String>, text: impl Into<String>) -> bool {
188        let mut projection = self.projection.clone();
189        projection.apply_updates(&[StateUpdate {
190            key: key.into(),
191            text: Some(text.into()),
192        }]);
193        projection.estimated_tokens() <= self.max_input_tokens
194    }
195}
196
197/// Application-owned behavior invoked by the generic subagent loop.
198pub trait Host: Send {
199    /// Renders the retained invocation text before execution.
200    fn render_tool_call(&mut self, call: &ToolCall) -> anyhow::Result<String>;
201
202    /// Executes one application tool.
203    fn execute_tool<'a>(
204        &'a mut self,
205        call: ToolCall,
206        operation_id: Uuid,
207        budget: ContextBudget,
208    ) -> HostFuture<'a, ToolOutcome>;
209
210    /// Completes an opaque freeform capture requested by a prior tool result.
211    fn complete_capture<'a>(
212        &'a mut self,
213        capture: Value,
214        contents: String,
215        budget: ContextBudget,
216    ) -> HostFuture<'a, ToolOutcome>;
217
218    /// Records one durable audit event selected by the runtime.
219    fn record(&mut self, event: AuditEvent) -> anyhow::Result<()>;
220}
221
222/// Inputs shared by every provider call in one primary session run.
223#[derive(Clone, Debug, Eq, PartialEq)]
224pub struct SessionRunRequest {
225    /// Stable user identifier used for router accounting.
226    pub user_id: String,
227    /// Top-level operation whose cancellation propagates to every provider call.
228    pub operation_id: Uuid,
229    /// Number of provider rounds already durably used by a restored session.
230    pub rounds_used: u64,
231    /// Maximum cumulative provider rounds permitted for the logical turn.
232    pub round_limit: u64,
233}
234
235/// Exact provider input prepared by the application for one primary round.
236#[derive(Clone, Debug, Eq, PartialEq)]
237pub struct PreparedRound {
238    /// Complete provider-visible context.
239    pub input: String,
240    /// Exact requested model identifier.
241    pub model: String,
242    /// Provider-neutral reasoning effort.
243    pub reasoning_effort: String,
244    /// Session-specific explanation attached to the single application-tool bridge.
245    pub tool_description: String,
246    /// Optional complete provider-turn timeout.
247    pub timeout: Option<Duration>,
248}
249
250/// Result of preparing the next primary round.
251#[derive(Clone, Debug, Eq, PartialEq)]
252pub enum RoundPreparation {
253    /// Run the prepared provider round.
254    Run(PreparedRound),
255    /// Finish without starting another provider round.
256    Complete(Option<String>),
257}
258
259/// One durable provider-protocol fact emitted by the primary runtime.
260#[derive(Clone, Debug, PartialEq)]
261pub enum SessionEvent {
262    /// Exact input identity submitted for one provider round.
263    InferenceSubmitted {
264        /// Cumulative one-based provider round.
265        round: u64,
266        /// SHA-256 hash of the exact provider input.
267        manifest_hash: String,
268        /// Exact requested model identifier.
269        model: String,
270    },
271    /// Exact provider transport input retained for diagnostics.
272    ProviderInput {
273        /// Cumulative one-based provider round.
274        round: u64,
275        /// Exact provider input record.
276        input: String,
277    },
278    /// Cumulative live usage reported during the provider round.
279    UsageUpdated {
280        /// Cumulative one-based provider round.
281        round: u64,
282        /// Provider-native cumulative usage.
283        usage: TokenUsage,
284    },
285    /// Canonical receipt and final usage for a completed or interrupted call.
286    ProviderReceipt {
287        /// Cumulative one-based provider round.
288        round: u64,
289        /// Final provider-native usage when available.
290        usage: Option<TokenUsage>,
291        /// Canonical durable receipt written by the intelligence router.
292        receipt: Box<UsageReceipt>,
293    },
294}
295
296/// Complete application handling of one primary-session tool call.
297#[derive(Clone, Debug, PartialEq)]
298pub struct SessionToolOutcome {
299    /// Exact result returned through the provider's native tool protocol.
300    pub text: String,
301    /// Whether tool execution succeeded.
302    pub ok: bool,
303    /// Opaque token requesting tool-free freeform output capture.
304    pub capture: Option<Value>,
305    /// Whether the logical session must stop after this tool response.
306    pub stop: bool,
307    /// Whether the logical session should finish after this provider round.
308    pub finish_after_round: bool,
309    /// Whether this tool already emitted the session's externally visible response.
310    pub emitted_response: bool,
311}
312
313impl SessionToolOutcome {
314    /// Constructs a simple successful tool result.
315    pub fn success(text: impl Into<String>) -> Self {
316        Self {
317            text: text.into(),
318            ok: true,
319            capture: None,
320            stop: false,
321            finish_after_round: false,
322            emitted_response: false,
323        }
324    }
325
326    /// Constructs a simple failed tool result.
327    pub fn failure(text: impl Into<String>) -> Self {
328        Self {
329            text: text.into(),
330            ok: false,
331            capture: None,
332            stop: false,
333            finish_after_round: false,
334            emitted_response: false,
335        }
336    }
337}
338
339/// Provider completion summarized for the application-owned session policy.
340#[derive(Clone, Debug, Eq, PartialEq)]
341pub struct RoundCompletion {
342    /// Terminal assistant text, which may be empty after tool use.
343    pub answer: String,
344    /// Whether the provider called at least one application tool.
345    pub used_tool: bool,
346    /// Whether a successful tool requested session completion after this round.
347    pub finish_requested: bool,
348    /// Whether a tool already emitted the externally visible response.
349    pub emitted_response: bool,
350}
351
352/// Application decision after completing a semantic primary-session transition.
353#[derive(Clone, Debug, Eq, PartialEq)]
354pub enum SessionControl {
355    /// Prepare another fresh provider round.
356    Continue,
357    /// Finish the logical turn with the optional externally visible answer.
358    Complete(Option<String>),
359}
360
361/// Application-owned semantics invoked by the primary agent runtime.
362pub trait SessionHost: Send {
363    /// Prepares the complete context and runtime selection for one provider round.
364    fn prepare_round<'a>(&'a mut self, round: u64) -> HostFuture<'a, RoundPreparation>;
365
366    /// Records one provider-protocol fact and durably checkpoints its effects.
367    fn record<'a>(&'a mut self, event: SessionEvent) -> HostFuture<'a, ()>;
368
369    /// Executes and durably projects one parsed or invalid application tool call.
370    fn execute_tool<'a>(
371        &'a mut self,
372        call: anyhow::Result<ToolCall>,
373        provider_operation_id: Uuid,
374    ) -> HostFuture<'a, SessionToolOutcome>;
375
376    /// Completes an opaque freeform capture requested by a prior tool result.
377    fn complete_capture<'a>(
378        &'a mut self,
379        capture: Value,
380        contents: String,
381    ) -> HostFuture<'a, SessionControl>;
382
383    /// Applies the terminal provider answer and decides whether another round is needed.
384    fn complete_round<'a>(
385        &'a mut self,
386        completion: RoundCompletion,
387    ) -> HostFuture<'a, SessionControl>;
388}
389
390/// Error returned when a restored primary turn exhausts its cumulative round budget.
391#[derive(Debug)]
392pub struct SessionRoundLimitError {
393    limit: u64,
394}
395
396impl std::fmt::Display for SessionRoundLimitError {
397    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
398        write!(
399            formatter,
400            "agent exceeded the {}-round tool-loop safety limit",
401            self.limit
402        )
403    }
404}
405
406impl std::error::Error for SessionRoundLimitError {}
407
408/// Returns whether an error is the primary runtime's round-limit signal.
409pub fn is_session_round_limit(error: &anyhow::Error) -> bool {
410    error.downcast_ref::<SessionRoundLimitError>().is_some()
411}
412
413/// Inputs for one complete subagent run.
414#[derive(Clone, Debug, PartialEq)]
415pub struct RunRequest {
416    /// Stable user identifier used for router accounting.
417    pub user_id: String,
418    /// Running parent operation whose cancellation propagates to each turn.
419    pub parent_operation_id: Uuid,
420    /// Exact requested model selector.
421    pub model: String,
422    /// Provider-neutral reasoning effort.
423    pub reasoning_effort: String,
424    /// Ordered immutable context sections.
425    pub context: Vec<String>,
426    /// Exact task presented after the context.
427    pub task: String,
428    /// Optional per-turn timeout.
429    pub timeout: Option<Duration>,
430    /// Additional application metadata included in the start audit event.
431    pub start_metadata: Value,
432}
433
434/// Completed subagent output.
435#[derive(Clone, Debug, Eq, PartialEq)]
436pub struct RunResult {
437    /// Final non-empty assistant answer.
438    pub answer: String,
439    /// Model used for every turn.
440    pub model: ResolvedAgentModel,
441}
442
443/// Cloneable provider-neutral subagent runtime.
444#[derive(Clone)]
445pub struct AgentRuntime {
446    intelligence: Intelligence,
447    round_limit: u64,
448}
449
450impl AgentRuntime {
451    /// Constructs a runtime over the sole direct-model boundary.
452    pub fn new(intelligence: Intelligence) -> Self {
453        Self {
454            intelligence,
455            round_limit: DEFAULT_ROUND_LIMIT,
456        }
457    }
458
459    /// Resolves a model without running an agent.
460    pub async fn resolve_model(&self, requested: &str) -> anyhow::Result<ResolvedAgentModel> {
461        self.intelligence
462            .resolve_agent_model(requested)
463            .await
464            .map_err(anyhow::Error::new)
465    }
466
467    /// Runs one primary logical turn across fresh provider rounds until the host completes it.
468    pub async fn run_session<H: SessionHost>(
469        &self,
470        request: SessionRunRequest,
471        host: &mut H,
472    ) -> anyhow::Result<Option<String>> {
473        ensure!(
474            request.round_limit > 0,
475            "session round limit must be positive"
476        );
477        ensure!(
478            request.rounds_used <= request.round_limit,
479            "restored session round count exceeds its safety limit"
480        );
481        let user = self
482            .intelligence
483            .for_user(request.user_id)
484            .map_err(anyhow::Error::new)?;
485
486        for round_index in request.rounds_used..request.round_limit {
487            let round = round_index + 1;
488            let prepared = match host.prepare_round(round).await? {
489                RoundPreparation::Run(prepared) => prepared,
490                RoundPreparation::Complete(answer) => return Ok(answer),
491            };
492            let manifest_hash = hex::encode(Sha256::digest(prepared.input.as_bytes()));
493            host.record(SessionEvent::InferenceSubmitted {
494                round,
495                manifest_hash: manifest_hash.clone(),
496                model: prepared.model.clone(),
497            })
498            .await?;
499
500            let mut provider_request = AgentRequest::new(prepared.input, prepared.model);
501            provider_request.reasoning_effort = reasoning_effort(&prepared.reasoning_effort)?;
502            provider_request.previous_thread_id = None;
503            provider_request.tools = vec![ktool_definition(&prepared.tool_description)];
504            if let Some(timeout) = prepared.timeout {
505                provider_request.timeout = timeout;
506            }
507            let mut turn = match user
508                .start_agent_turn(request.operation_id, None, provider_request)
509                .await
510            {
511                Ok(turn) => turn,
512                Err(error) => {
513                    if let Some(receipt) = error.receipt().cloned() {
514                        host.record(SessionEvent::ProviderReceipt {
515                            round,
516                            usage: None,
517                            receipt: Box::new(receipt),
518                        })
519                        .await?;
520                    }
521                    return Err(anyhow::Error::new(error));
522                }
523            };
524            let mut used_tool = false;
525            let mut finish_requested = false;
526            let mut emitted_response = false;
527            let mut pending_capture: Option<Value> = None;
528            let completed = loop {
529                let event = match turn.next_event().await {
530                    Ok(Some(event)) => event,
531                    Ok(None) => {
532                        let receipt = turn.finish_unavailable()?.clone();
533                        host.record(SessionEvent::ProviderReceipt {
534                            round,
535                            usage: None,
536                            receipt: Box::new(receipt),
537                        })
538                        .await?;
539                        anyhow::bail!("provider ended without a terminal turn event");
540                    }
541                    Err(error) => {
542                        if let Some(receipt) = error.receipt().cloned() {
543                            host.record(SessionEvent::ProviderReceipt {
544                                round,
545                                usage: None,
546                                receipt: Box::new(receipt),
547                            })
548                            .await?;
549                        }
550                        return Err(anyhow::Error::new(error));
551                    }
552                };
553                match event {
554                    AgentEvent::ProviderInput(input) => {
555                        host.record(SessionEvent::ProviderInput { round, input })
556                            .await?;
557                    }
558                    AgentEvent::UsageUpdated(usage) => {
559                        host.record(SessionEvent::UsageUpdated { round, usage })
560                            .await?;
561                    }
562                    AgentEvent::ToolCall(native) => {
563                        used_tool = true;
564                        let call = parse_ktool_call(&native)
565                            .map_err(|error| anyhow::anyhow!("Invalid Ktool call: {error}"));
566                        let mut outcome = host.execute_tool(call, request.operation_id).await?;
567                        finish_requested |= outcome.ok && outcome.finish_after_round;
568                        emitted_response |= outcome.ok && outcome.emitted_response;
569                        pending_capture = outcome.capture.take();
570                        let stop = outcome.stop;
571                        respond_session_or_record(
572                            &mut turn,
573                            host,
574                            round,
575                            &native.call_id,
576                            if outcome.ok {
577                                ToolResult::success(outcome.text)
578                            } else {
579                                ToolResult::failure(outcome.text)
580                            },
581                        )
582                        .await?;
583                        if stop {
584                            return Ok(None);
585                        }
586                    }
587                    AgentEvent::Completed(completed) => break completed,
588                }
589            };
590            let receipt = turn
591                .receipt()
592                .context("provider completed without a usage receipt")?
593                .clone();
594            host.record(SessionEvent::ProviderReceipt {
595                round,
596                usage: completed.usage.clone(),
597                receipt: Box::new(receipt),
598            })
599            .await?;
600
601            let control = if let Some(capture) = pending_capture {
602                host.complete_capture(capture, completed.answer).await?
603            } else {
604                host.complete_round(RoundCompletion {
605                    answer: completed.answer,
606                    used_tool,
607                    finish_requested,
608                    emitted_response,
609                })
610                .await?
611            };
612            match control {
613                SessionControl::Continue => {}
614                SessionControl::Complete(answer) => return Ok(answer),
615            }
616        }
617        Err(SessionRoundLimitError {
618            limit: request.round_limit,
619        }
620        .into())
621    }
622
623    /// Runs one fresh-context subagent to a final non-empty answer.
624    pub async fn run<H: Host>(
625        &self,
626        request: RunRequest,
627        host: &mut H,
628    ) -> anyhow::Result<RunResult> {
629        let selected = self.resolve_model(&request.model).await?;
630        let reasoning_effort = reasoning_effort(&request.reasoning_effort)?;
631        let mut projection = Projection::new(request.context, request.task);
632        ensure_capacity(&projection, selected.max_input_tokens)?;
633        host.record(AuditEvent::Started {
634            parent_operation_id: request.parent_operation_id,
635            model: request.model.clone(),
636            provider_model: selected.provider_model.clone(),
637            provider: selected.provider,
638            context_window_tokens: selected.context_window_tokens,
639            max_input_tokens: selected.max_input_tokens,
640            context: projection.context.clone(),
641            task: projection.task.clone(),
642            host: request.start_metadata.clone(),
643        })?;
644        let user = self
645            .intelligence
646            .for_user(request.user_id)
647            .map_err(anyhow::Error::new)?;
648        let mut deferred_capture: Option<Value> = None;
649
650        for round in 0..self.round_limit {
651            let capturing = deferred_capture.is_some();
652            ensure_capacity(&projection, selected.max_input_tokens)?;
653            let input = projection.render();
654            let manifest_hash = hex::encode(Sha256::digest(input.as_bytes()));
655            host.record(AuditEvent::InferenceSubmitted {
656                parent_operation_id: request.parent_operation_id,
657                round: round + 1,
658                manifest_hash: manifest_hash.clone(),
659                estimated_input_tokens: projection.estimated_tokens(),
660            })?;
661            let mut provider_request = AgentRequest::new(input, selected.requested_model.clone());
662            provider_request.reasoning_effort = reasoning_effort;
663            provider_request.ephemeral = true;
664            provider_request.tools = if capturing {
665                Vec::new()
666            } else {
667                vec![ktool_definition(
668                    "Call one available Ktool by its exact name.",
669                )]
670            };
671            if let Some(timeout) = request.timeout {
672                provider_request.timeout = timeout;
673            }
674            let child_operation_id = Uuid::new_v4();
675            let mut turn = match user
676                .start_agent_turn(
677                    child_operation_id,
678                    Some(request.parent_operation_id),
679                    provider_request,
680                )
681                .await
682            {
683                Ok(turn) => turn,
684                Err(error) => {
685                    if let Some(receipt) = error.receipt().cloned() {
686                        host.record(AuditEvent::ProviderReceipt {
687                            parent_operation_id: request.parent_operation_id,
688                            round: round + 1,
689                            manifest_hash,
690                            usage: None,
691                            receipt: Box::new(receipt),
692                        })?;
693                    }
694                    return Err(anyhow::Error::new(error));
695                }
696            };
697            let mut used_tool = false;
698            let mut pending_capture: Option<Value> = None;
699            let mut requires_rerender = false;
700            let completed = loop {
701                let event = match turn.next_event().await {
702                    Ok(Some(event)) => event,
703                    Ok(None) => {
704                        let receipt = turn.finish_unavailable()?.clone();
705                        host.record(AuditEvent::ProviderReceipt {
706                            parent_operation_id: request.parent_operation_id,
707                            round: round + 1,
708                            manifest_hash: manifest_hash.clone(),
709                            usage: None,
710                            receipt: Box::new(receipt),
711                        })?;
712                        anyhow::bail!("subagent provider ended without a terminal turn event");
713                    }
714                    Err(error) => {
715                        if let Some(receipt) = error.receipt().cloned() {
716                            host.record(AuditEvent::ProviderReceipt {
717                                parent_operation_id: request.parent_operation_id,
718                                round: round + 1,
719                                manifest_hash: manifest_hash.clone(),
720                                usage: None,
721                                receipt: Box::new(receipt),
722                            })?;
723                        }
724                        return Err(anyhow::Error::new(error));
725                    }
726                };
727                match event {
728                    AgentEvent::ProviderInput(_) => {}
729                    AgentEvent::UsageUpdated(_) => {}
730                    AgentEvent::ToolCall(native) => {
731                        used_tool = true;
732                        if capturing {
733                            respond_or_record(
734                                &mut turn,
735                                host,
736                                request.parent_operation_id,
737                                round + 1,
738                                &manifest_hash,
739                                &native.call_id,
740                                ToolResult::failure(
741                                    "No application tool is available while complete freeform output is being captured.",
742                                ),
743                            )
744                            .await?;
745                            continue;
746                        }
747                        if pending_capture.is_some() {
748                            respond_or_record(
749                                &mut turn,
750                                host,
751                                request.parent_operation_id,
752                                round + 1,
753                                &manifest_hash,
754                                &native.call_id,
755                                ToolResult::failure(
756                                    "A freeform output capture is pending; no other tool can run first.",
757                                ),
758                            )
759                            .await?;
760                            continue;
761                        }
762                        if requires_rerender {
763                            respond_or_record(
764                                &mut turn,
765                                host,
766                                request.parent_operation_id,
767                                round + 1,
768                                &manifest_hash,
769                                &native.call_id,
770                                ToolResult::failure(
771                                    "A state update is waiting to be re-rendered. End this slice before calling another tool.",
772                                ),
773                            )
774                            .await?;
775                            continue;
776                        }
777                        let call = match parse_ktool_call(&native) {
778                            Ok(call) => call,
779                            Err(error) => {
780                                let text = format!("Invalid application tool call: {error}");
781                                projection.push_history(format!("Ktool result:\n{text}"));
782                                respond_or_record(
783                                    &mut turn,
784                                    host,
785                                    request.parent_operation_id,
786                                    round + 1,
787                                    &manifest_hash,
788                                    &native.call_id,
789                                    ToolResult::failure(text),
790                                )
791                                .await?;
792                                continue;
793                            }
794                        };
795                        host.record(AuditEvent::ToolCall {
796                            parent_operation_id: request.parent_operation_id,
797                            name: call.name.clone(),
798                            arguments: call.arguments.clone(),
799                        })?;
800                        projection.push_history(format!(
801                            "Ktool call:\n{}",
802                            host.render_tool_call(&call)?
803                        ));
804                        let budget = ContextBudget {
805                            projection: projection.clone(),
806                            max_input_tokens: selected.max_input_tokens,
807                        };
808                        let mut outcome = host
809                            .execute_tool(call.clone(), child_operation_id, budget)
810                            .await
811                            .unwrap_or_else(|error| {
812                                ToolOutcome::failure(format!("{} failed: {error}", call.name))
813                            });
814                        let exact_result = outcome.text.clone();
815                        let initially_ok = outcome.ok;
816                        let has_state_updates = initially_ok && !outcome.state_updates.is_empty();
817                        let mut provider_result = outcome.text.clone();
818                        let mut candidate = projection.clone();
819                        candidate.apply_updates(successful_state_updates(&outcome));
820                        candidate.push_history_displaying(
821                            format!("Ktool result:\n{provider_result}"),
822                            if initially_ok {
823                                outcome.displayed_state_keys.clone()
824                            } else {
825                                Default::default()
826                            },
827                        );
828                        let accepted = candidate.estimated_tokens() <= selected.max_input_tokens;
829                        if accepted {
830                            projection = candidate;
831                            requires_rerender = has_state_updates;
832                        } else {
833                            outcome.ok = false;
834                            outcome.capture = None;
835                            provider_result = "The tool ran, but its result or updated state could not fit in the subagent context. Do not retry it; report the capacity failure to Kennedy.".into();
836                            projection.push_history(format!("Ktool result:\n{provider_result}"));
837                        }
838                        host.record(AuditEvent::ToolResult {
839                            parent_operation_id: request.parent_operation_id,
840                            name: call.name.clone(),
841                            ok: initially_ok,
842                            projection_accepted: accepted,
843                            result: exact_result,
844                        })?;
845                        pending_capture = outcome.capture.take();
846                        respond_or_record(
847                            &mut turn,
848                            host,
849                            request.parent_operation_id,
850                            round + 1,
851                            &manifest_hash,
852                            &native.call_id,
853                            if outcome.ok {
854                                ToolResult::success(provider_result)
855                            } else {
856                                ToolResult::failure(provider_result)
857                            },
858                        )
859                        .await?;
860                    }
861                    AgentEvent::Completed(completed) => break completed,
862                }
863            };
864            let receipt = turn
865                .receipt()
866                .context("subagent provider completed without a usage receipt")?
867                .clone();
868            host.record(AuditEvent::ProviderReceipt {
869                parent_operation_id: request.parent_operation_id,
870                round: round + 1,
871                manifest_hash: manifest_hash.clone(),
872                usage: completed.usage.clone(),
873                receipt: Box::new(receipt),
874            })?;
875
876            let capture = deferred_capture.take().or(pending_capture);
877            if let Some(capture) = capture {
878                if !capturing && completed.answer.is_empty() {
879                    deferred_capture = Some(capture);
880                    continue;
881                }
882                let budget = ContextBudget {
883                    projection: projection.clone(),
884                    max_input_tokens: selected.max_input_tokens,
885                };
886                let outcome = host
887                    .complete_capture(capture, completed.answer, budget)
888                    .await?;
889                let mut candidate = projection.clone();
890                candidate.apply_updates(successful_state_updates(&outcome));
891                candidate.push_history(format!("Ktool result:\n{}", outcome.text));
892                ensure_capacity(&candidate, selected.max_input_tokens)?;
893                projection = candidate;
894                continue;
895            }
896            if requires_rerender {
897                let draft = completed.answer.trim();
898                if !draft.is_empty() {
899                    projection.push_history(format!(
900                        "Assistant draft produced before the state refresh:\n{draft}"
901                    ));
902                }
903                continue;
904            }
905            let answer = completed.answer.trim().to_owned();
906            if !answer.is_empty() {
907                host.record(AuditEvent::Completed {
908                    parent_operation_id: request.parent_operation_id,
909                    model: request.model.clone(),
910                    response: answer.clone(),
911                })?;
912                return Ok(RunResult {
913                    answer,
914                    model: selected,
915                });
916            }
917            ensure!(
918                used_tool,
919                "subagent provider completed without a response or tool call"
920            );
921        }
922        anyhow::bail!(
923            "subagent exceeded the {}-round tool-loop safety limit",
924            self.round_limit
925        )
926    }
927}
928
929async fn respond_or_record<H: Host>(
930    turn: &mut kcode_intelligence_router::AgentTurn,
931    host: &mut H,
932    parent_operation_id: Uuid,
933    round: u64,
934    manifest_hash: &str,
935    call_id: &str,
936    result: ToolResult,
937) -> anyhow::Result<()> {
938    if let Err(error) = turn.respond(call_id, result).await {
939        let receipt = turn.finish_unavailable()?.clone();
940        host.record(AuditEvent::ProviderReceipt {
941            parent_operation_id,
942            round,
943            manifest_hash: manifest_hash.into(),
944            usage: None,
945            receipt: Box::new(receipt),
946        })?;
947        return Err(anyhow::Error::new(error));
948    }
949    Ok(())
950}
951
952async fn respond_session_or_record<H: SessionHost>(
953    turn: &mut kcode_intelligence_router::AgentTurn,
954    host: &mut H,
955    round: u64,
956    call_id: &str,
957    result: ToolResult,
958) -> anyhow::Result<()> {
959    if let Err(error) = turn.respond(call_id, result).await {
960        let receipt = turn.finish_unavailable()?.clone();
961        host.record(SessionEvent::ProviderReceipt {
962            round,
963            usage: None,
964            receipt: Box::new(receipt),
965        })
966        .await?;
967        return Err(anyhow::Error::new(error));
968    }
969    Ok(())
970}
971
972#[derive(Clone)]
973struct Projection {
974    context: Vec<String>,
975    task: String,
976    history: Vec<ProjectedHistory>,
977    states: Vec<ProjectedState>,
978}
979
980#[derive(Clone)]
981struct ProjectedHistory {
982    text: String,
983    displayed_state_keys: Vec<String>,
984}
985
986#[derive(Clone)]
987struct ProjectedState {
988    key: String,
989    text: String,
990}
991
992impl Projection {
993    fn new(context: Vec<String>, task: String) -> Self {
994        Self {
995            context,
996            task,
997            history: Vec::new(),
998            states: Vec::new(),
999        }
1000    }
1001
1002    fn render(&self) -> String {
1003        self.context
1004            .iter()
1005            .map(String::as_str)
1006            .chain(std::iter::once(self.task.as_str()))
1007            .chain(self.history.iter().map(|entry| entry.text.as_str()))
1008            .chain(
1009                self.states
1010                    .iter()
1011                    .filter(|state| {
1012                        !self.history.iter().any(|entry| {
1013                            entry
1014                                .displayed_state_keys
1015                                .iter()
1016                                .any(|key| key == &state.key)
1017                        })
1018                    })
1019                    .map(|state| state.text.as_str()),
1020            )
1021            .filter(|section| !section.is_empty())
1022            .collect::<Vec<_>>()
1023            .join("\n\n")
1024    }
1025
1026    fn push_history(&mut self, text: impl Into<String>) {
1027        self.push_history_displaying(text, Vec::new());
1028    }
1029
1030    fn push_history_displaying(
1031        &mut self,
1032        text: impl Into<String>,
1033        displayed_state_keys: Vec<String>,
1034    ) {
1035        self.history.push(ProjectedHistory {
1036            text: text.into(),
1037            displayed_state_keys,
1038        });
1039    }
1040
1041    fn update_state(&mut self, key: String, text: Option<String>) {
1042        self.states.retain(|state| state.key != key);
1043        if let Some(text) = text {
1044            self.states.push(ProjectedState { key, text });
1045        }
1046    }
1047
1048    fn apply_updates(&mut self, updates: &[StateUpdate]) {
1049        for update in updates {
1050            let marker = if update.text.is_some() {
1051                SUPERSEDED_TOOL_OUTPUT
1052            } else {
1053                REMOVED_TOOL_OUTPUT
1054            };
1055            for entry in &mut self.history {
1056                if entry
1057                    .displayed_state_keys
1058                    .iter()
1059                    .any(|key| key == &update.key)
1060                {
1061                    entry.text = marker.into();
1062                    entry.displayed_state_keys.clear();
1063                }
1064            }
1065            self.update_state(update.key.clone(), update.text.clone());
1066        }
1067    }
1068
1069    fn estimated_tokens(&self) -> u64 {
1070        (self.render().chars().count() as u64)
1071            .div_ceil(4)
1072            .saturating_add(PROTOCOL_TOKEN_RESERVE)
1073    }
1074}
1075
1076fn successful_state_updates(outcome: &ToolOutcome) -> &[StateUpdate] {
1077    if outcome.ok {
1078        &outcome.state_updates
1079    } else {
1080        &[]
1081    }
1082}
1083
1084fn ensure_capacity(projection: &Projection, max_input_tokens: u64) -> anyhow::Result<()> {
1085    let estimated = projection.estimated_tokens();
1086    ensure!(
1087        estimated <= max_input_tokens,
1088        "subagent context requires approximately {estimated} input tokens, over the selected model's {max_input_tokens}-token input limit"
1089    );
1090    Ok(())
1091}
1092
1093fn ktool_definition(description: &str) -> DynamicTool {
1094    DynamicTool::new(
1095        "call_ktool",
1096        description,
1097        json!({
1098            "type": "object",
1099            "additionalProperties": false,
1100            "required": ["name", "arguments"],
1101            "properties": {
1102                "name": {"type": "string"},
1103                "arguments": {"type": "object"}
1104            }
1105        }),
1106    )
1107}
1108
1109fn parse_ktool_call(call: &DynamicToolCall) -> anyhow::Result<ToolCall> {
1110    ensure!(call.tool == "call_ktool", "unknown provider tool");
1111    let arguments = call
1112        .arguments
1113        .as_object()
1114        .context("call_ktool arguments must be an object")?;
1115    ensure!(
1116        arguments
1117            .keys()
1118            .all(|key| matches!(key.as_str(), "name" | "arguments")),
1119        "call_ktool contains unknown arguments"
1120    );
1121    let name = arguments
1122        .get("name")
1123        .and_then(Value::as_str)
1124        .map(str::trim)
1125        .filter(|name| !name.is_empty() && name.chars().count() <= 100)
1126        .context("call_ktool.name must be a non-empty bounded string")?
1127        .to_owned();
1128    let arguments = arguments
1129        .get("arguments")
1130        .filter(|value| value.is_object())
1131        .context("call_ktool.arguments must be an object")?
1132        .clone();
1133    Ok(ToolCall { name, arguments })
1134}
1135
1136fn reasoning_effort(value: &str) -> anyhow::Result<ReasoningEffort> {
1137    Ok(match value {
1138        "none" => ReasoningEffort::None,
1139        "minimal" => ReasoningEffort::Minimal,
1140        "low" => ReasoningEffort::Low,
1141        "medium" => ReasoningEffort::Medium,
1142        "high" => ReasoningEffort::High,
1143        "xhigh" => ReasoningEffort::XHigh,
1144        "max" => ReasoningEffort::Max,
1145        _ => anyhow::bail!("unsupported reasoning effort {value:?}"),
1146    })
1147}
1148
1149#[cfg(test)]
1150mod tests {
1151    use super::*;
1152
1153    #[test]
1154    fn projection_replaces_state_and_budget_accounts_for_reserve() {
1155        let mut projection = Projection::new(vec!["context".into()], "task".into());
1156        projection.update_state("file".into(), Some("old".into()));
1157        projection.update_state("file".into(), Some("new".into()));
1158        assert_eq!(projection.states.len(), 1);
1159        assert!(projection.render().contains("new"));
1160        assert!(!projection.render().contains("old"));
1161        assert!(projection.estimated_tokens() >= PROTOCOL_TOKEN_RESERVE);
1162    }
1163
1164    #[test]
1165    fn state_update_supersedes_earlier_display_and_moves_current_value_after_history() {
1166        let mut projection = Projection::new(vec!["context".into()], "task".into());
1167        projection.push_history("Ktool call: open");
1168        projection.push_history_displaying(
1169            "Ktool result:\ncomplete old source",
1170            vec!["tool-state:7".into()],
1171        );
1172        projection.push_history("Ktool call: write");
1173        projection.apply_updates(&[StateUpdate {
1174            key: "tool-state:7".into(),
1175            text: Some("complete new source".into()),
1176        }]);
1177        projection.push_history("Ktool result: write completed");
1178
1179        let rendered = projection.render();
1180        assert!(!rendered.contains("complete old source"));
1181        assert_eq!(rendered.matches(SUPERSEDED_TOOL_OUTPUT).count(), 1);
1182        assert_eq!(rendered.matches("complete new source").count(), 1);
1183        assert!(
1184            rendered.find(SUPERSEDED_TOOL_OUTPUT).unwrap()
1185                < rendered.find("Ktool call: write").unwrap()
1186        );
1187        assert!(
1188            rendered.find("Ktool result: write completed").unwrap()
1189                < rendered.find("complete new source").unwrap()
1190        );
1191    }
1192
1193    #[test]
1194    fn current_state_is_not_duplicated_when_the_latest_result_displays_it() {
1195        let mut projection = Projection::new(vec!["context".into()], "task".into());
1196        projection.apply_updates(&[StateUpdate {
1197            key: "tool-state:7".into(),
1198            text: Some("complete source".into()),
1199        }]);
1200        projection.push_history_displaying(
1201            "Ktool result:\ncomplete source",
1202            vec!["tool-state:7".into()],
1203        );
1204
1205        assert_eq!(projection.render().matches("complete source").count(), 1);
1206    }
1207
1208    #[test]
1209    fn state_update_only_supersedes_displays_with_the_same_identity() {
1210        let mut projection = Projection::new(vec!["context".into()], "task".into());
1211        projection.push_history_displaying("first output", vec!["first".into()]);
1212        projection.push_history_displaying("second output", vec!["second".into()]);
1213        projection.apply_updates(&[StateUpdate {
1214            key: "first".into(),
1215            text: Some("current first output".into()),
1216        }]);
1217
1218        let rendered = projection.render();
1219        assert!(
1220            !rendered
1221                .split("\n\n")
1222                .any(|section| section == "first output")
1223        );
1224        assert!(rendered.contains(SUPERSEDED_TOOL_OUTPUT));
1225        assert!(rendered.contains("second output"));
1226        assert!(rendered.contains("current first output"));
1227    }
1228
1229    #[test]
1230    fn removed_state_uses_a_truthful_supersession_marker() {
1231        let mut projection = Projection::new(vec!["context".into()], "task".into());
1232        projection.push_history_displaying("retired output", vec!["state".into()]);
1233        projection.apply_updates(&[StateUpdate {
1234            key: "state".into(),
1235            text: None,
1236        }]);
1237
1238        let rendered = projection.render();
1239        assert!(!rendered.contains("retired output"));
1240        assert!(rendered.contains(REMOVED_TOOL_OUTPUT));
1241    }
1242
1243    #[test]
1244    fn failed_outcomes_cannot_update_or_supersede_state() {
1245        let mut projection = Projection::new(vec!["context".into()], "task".into());
1246        projection.push_history_displaying("current output", vec!["state".into()]);
1247        let failed = ToolOutcome {
1248            text: "write failed".into(),
1249            ok: false,
1250            state_updates: vec![StateUpdate {
1251                key: "state".into(),
1252                text: Some("invalid update".into()),
1253            }],
1254            displayed_state_keys: Vec::new(),
1255            capture: None,
1256        };
1257
1258        projection.apply_updates(successful_state_updates(&failed));
1259
1260        let rendered = projection.render();
1261        assert!(rendered.contains("current output"));
1262        assert!(!rendered.contains("invalid update"));
1263        assert!(!rendered.contains(SUPERSEDED_TOOL_OUTPUT));
1264    }
1265
1266    #[test]
1267    fn large_stateful_result_is_rendered_exactly_once() {
1268        let loaded_node = format!(
1269            "Node body:\n{}\n\nFixed connections:\nfixed-node\n\nRecent connections:\nrecent-node",
1270            "x".repeat(2_000)
1271        );
1272        let outcome = ToolOutcome {
1273            text: loaded_node.clone(),
1274            ok: true,
1275            state_updates: vec![StateUpdate {
1276                key: "loaded-node".into(),
1277                text: Some(loaded_node.clone()),
1278            }],
1279            displayed_state_keys: vec!["loaded-node".into()],
1280            capture: None,
1281        };
1282        let mut projection =
1283            Projection::new(vec!["initial node description".into()], "task".into());
1284        let provider_result = outcome.text.clone();
1285        projection.apply_updates(successful_state_updates(&outcome));
1286        projection.push_history_displaying(
1287            format!("Ktool result:\n{provider_result}"),
1288            outcome.displayed_state_keys,
1289        );
1290
1291        let rendered = projection.render();
1292        assert_eq!(rendered.matches(loaded_node.as_str()).count(), 1);
1293        assert!(rendered.contains("Fixed connections:\nfixed-node"));
1294        assert!(rendered.contains("Recent connections:\nrecent-node"));
1295    }
1296
1297    #[test]
1298    fn native_tool_wrapper_is_strict() {
1299        let call = parse_ktool_call(&DynamicToolCall {
1300            call_id: "1".into(),
1301            tool: "call_ktool".into(),
1302            arguments: json!({"name": "Read", "arguments": {"id": 1}}),
1303        })
1304        .unwrap();
1305        assert_eq!(call.name, "Read");
1306        assert_eq!(call.arguments["id"], 1);
1307    }
1308}
kcode_agent_runtime/lib.rs

kcode_agent_runtime/
lib.rs
