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

kcode_agent_runtime/
lib.rs
