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

kcode_agent_runtime/
lib.rs
