agent_runtime_ports/
lib.rs

1//! Provider-neutral capabilities required by the Runtime Kernel.
2//!
3//! This crate contains no HTTP client, database driver, Axum type, Agent Infra
4//! DTO, or provider implementation.
5
6use std::{
7    collections::BTreeMap,
8    time::{Duration, Instant},
9};
10
11use async_trait::async_trait;
12use runtime_types::{
13    CallerScope, ConversationId, DelegationLeaseRef, ExecutionId, OperationId, RequestAuthority,
14    RuntimeInstanceId, WorkspaceId,
15};
16use serde::{Deserialize, Serialize};
17use serde_json::Value;
18use thiserror::Error;
19use tokio_util::sync::CancellationToken;
20
21#[derive(Clone, Debug)]
22pub struct OperationContext {
23    pub id: OperationId,
24    pub execution_id: ExecutionId,
25    pub deadline: Instant,
26    pub cancellation: CancellationToken,
27}
28
29impl OperationContext {
30    pub fn remaining(&self) -> Result<Duration, PortFailure> {
31        if self.cancellation.is_cancelled() {
32            return Err(PortFailure::canceled());
33        }
34        self.deadline
35            .checked_duration_since(Instant::now())
36            .filter(|remaining| !remaining.is_zero())
37            .ok_or_else(PortFailure::timeout)
38    }
39}
40
41#[derive(Debug, Clone, Copy, PartialEq, Eq)]
42pub enum CommitDisposition {
43    NotCommitted,
44    Committed,
45    Unknown,
46}
47
48#[derive(Debug, Clone, Copy, PartialEq, Eq)]
49pub enum PortFailureKind {
50    Invalid,
51    NotFound,
52    Forbidden,
53    Conflict,
54    RateLimited,
55    Timeout,
56    Canceled,
57    Unavailable,
58    Protocol,
59    Internal,
60}
61
62#[derive(Debug, Error, Clone)]
63#[error("{code}: {message}")]
64pub struct PortFailure {
65    pub kind: PortFailureKind,
66    pub code: String,
67    pub message: String,
68    pub retryable: bool,
69    pub commit: CommitDisposition,
70}
71
72impl PortFailure {
73    pub fn new(kind: PortFailureKind, code: impl Into<String>, message: impl Into<String>) -> Self {
74        Self {
75            kind,
76            code: code.into(),
77            message: message.into(),
78            retryable: false,
79            commit: CommitDisposition::NotCommitted,
80        }
81    }
82    pub fn timeout() -> Self {
83        Self::new(
84            PortFailureKind::Timeout,
85            "DEADLINE_EXCEEDED",
86            "operation deadline exceeded",
87        )
88    }
89    pub fn canceled() -> Self {
90        Self::new(PortFailureKind::Canceled, "CANCELED", "operation canceled")
91    }
92}
93
94#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
95#[serde(tag = "type", rename_all = "snake_case")]
96pub enum ModelContent {
97    Text {
98        text: String,
99    },
100    ToolUse {
101        id: String,
102        name: String,
103        arguments: Value,
104    },
105    ToolResult {
106        tool_use_id: String,
107        name: String,
108        content: String,
109        failed: bool,
110    },
111}
112
113#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
114#[serde(rename_all = "snake_case")]
115pub enum ModelRole {
116    System,
117    User,
118    Assistant,
119    Tool,
120}
121
122#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
123#[serde(rename_all = "camelCase")]
124pub struct ModelMessage {
125    pub role: ModelRole,
126    pub content: Vec<ModelContent>,
127}
128
129impl ModelMessage {
130    pub fn text(role: ModelRole, text: impl Into<String>) -> Self {
131        Self {
132            role,
133            content: vec![ModelContent::Text { text: text.into() }],
134        }
135    }
136}
137
138#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
139#[serde(rename_all = "camelCase")]
140pub struct ModelToolDefinition {
141    pub name: String,
142    pub description: String,
143    pub input_schema: Value,
144}
145
146#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
147#[serde(rename_all = "camelCase")]
148pub struct ModelRequest {
149    /// Stable across transport attempts and shared with Context invocation
150    /// audit. This is correlation identity, not permission to retry a model
151    /// side effect whose commit disposition is unknown.
152    pub invocation_id: String,
153    pub model: String,
154    pub messages: Vec<ModelMessage>,
155    pub tools: Vec<ModelToolDefinition>,
156    /// Provider-neutral generation controls. Keeping these controls on the
157    /// model port prevents the HTTP adapters from leaking OpenAI- or
158    /// Claude-specific DTOs into the Kernel while still preserving the
159    /// semantics that Agent Infra's canonical model contract supports.
160    pub generation: ModelGenerationOptions,
161}
162
163#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
164#[serde(rename_all = "camelCase")]
165pub struct ModelGenerationOptions {
166    pub max_output_tokens: Option<u32>,
167    pub temperature: Option<f32>,
168    #[serde(default)]
169    pub stop_sequences: Vec<String>,
170    pub response_format: Option<Value>,
171    pub tool_choice: ModelToolChoice,
172}
173
174#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
175#[serde(tag = "type", rename_all = "snake_case")]
176pub enum ModelToolChoice {
177    #[default]
178    Auto,
179    None,
180}
181
182#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
183#[serde(rename_all = "snake_case")]
184pub enum ModelFinish {
185    Stop,
186    ToolCalls,
187    Length,
188    ContentFilter,
189    Other,
190}
191
192impl ModelFinish {
193    pub fn is_complete(self) -> bool {
194        matches!(self, Self::Stop | Self::ToolCalls)
195    }
196}
197
198#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
199#[serde(rename_all = "camelCase")]
200pub struct TokenUsage {
201    /// `None` means the provider did not report this value. It must never be
202    /// collapsed to zero because zero is a real, materially different usage.
203    pub input_tokens: Option<u64>,
204    pub output_tokens: Option<u64>,
205}
206
207impl TokenUsage {
208    /// Additive identity used only when beginning a known aggregate.
209    pub const fn zero() -> Self {
210        Self {
211            input_tokens: Some(0),
212            output_tokens: Some(0),
213        }
214    }
215
216    /// Aggregate without inventing missing provider measurements. Once any
217    /// turn is unknown, the corresponding execution total stays unknown.
218    pub fn add_assign(&mut self, value: &Self) {
219        self.input_tokens = add_known(self.input_tokens, value.input_tokens);
220        self.output_tokens = add_known(self.output_tokens, value.output_tokens);
221    }
222}
223
224fn add_known(left: Option<u64>, right: Option<u64>) -> Option<u64> {
225    left.zip(right)
226        .map(|(left, right)| left.saturating_add(right))
227}
228
229#[cfg(test)]
230mod token_usage_tests {
231    use super::TokenUsage;
232
233    #[test]
234    fn aggregate_preserves_unknown_and_saturates_known_values() {
235        let mut usage = TokenUsage::zero();
236        usage.add_assign(&TokenUsage {
237            input_tokens: Some(u64::MAX),
238            output_tokens: Some(3),
239        });
240        usage.add_assign(&TokenUsage {
241            input_tokens: Some(1),
242            output_tokens: None,
243        });
244        assert_eq!(usage.input_tokens, Some(u64::MAX));
245        assert_eq!(usage.output_tokens, None);
246    }
247}
248
249#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
250#[serde(rename_all = "camelCase")]
251pub struct ModelResponse {
252    pub output: Vec<ModelContent>,
253    pub finish: ModelFinish,
254    pub usage: TokenUsage,
255}
256
257#[async_trait]
258pub trait ModelSession: Send + Sync {
259    async fn invoke(
260        &self,
261        operation: &OperationContext,
262        request: ModelRequest,
263    ) -> Result<ModelResponse, PortFailure>;
264}
265
266#[derive(Debug, Clone, PartialEq, Eq)]
267pub struct ConversationMessage {
268    pub role: ModelRole,
269    pub text: String,
270}
271
272#[async_trait]
273pub trait ConversationSession: Send + Sync {
274    async fn load_recent(
275        &self,
276        operation: &OperationContext,
277        limit: usize,
278    ) -> Result<Vec<ConversationMessage>, PortFailure>;
279    async fn append(
280        &self,
281        operation: &OperationContext,
282        messages: &[ConversationMessage],
283    ) -> Result<(), PortFailure>;
284}
285
286#[derive(Debug, Clone, Copy, PartialEq, Eq)]
287pub enum ContextAuditStage {
288    Original,
289    Assembled,
290    Compressed,
291}
292
293#[derive(Debug, Clone)]
294pub struct ContextAuditSnapshot {
295    pub id: String,
296    pub stage: ContextAuditStage,
297    pub purpose: String,
298    pub messages: Vec<ModelMessage>,
299    pub tools: Vec<ModelToolDefinition>,
300    pub protocol_overhead_tokens: u64,
301    pub reserved_output_tokens: u64,
302    pub estimated_tokens: u64,
303    /// Digest of the exact messages, Tool projection, and output reservation
304    /// used for the corresponding model request.
305    pub request_digest: String,
306}
307
308#[derive(Debug, Clone, Copy, PartialEq, Eq)]
309pub enum ContextDerivationKind {
310    Assemble,
311    Compress,
312    Trim,
313    Reorder,
314}
315
316#[derive(Debug, Clone)]
317pub struct ContextAuditDerivation {
318    pub id: String,
319    pub source_snapshot_id: String,
320    pub target_snapshot_id: String,
321    pub kind: ContextDerivationKind,
322    pub input_tokens: u64,
323    pub output_tokens: u64,
324    pub input_items: usize,
325    pub output_items: usize,
326}
327
328#[derive(Debug, Clone, Copy, PartialEq, Eq)]
329pub enum ModelInvocationAuditState {
330    Prepared,
331    Succeeded,
332    Failed,
333}
334
335#[derive(Debug, Clone)]
336pub struct ModelInvocationAudit {
337    pub id: String,
338    pub request_snapshot_id: String,
339    pub model: String,
340    pub state: ModelInvocationAuditState,
341    pub usage: TokenUsage,
342    pub error_code: Option<String>,
343}
344
345#[async_trait]
346pub trait ContextAuditSession: Send + Sync {
347    async fn snapshot(
348        &self,
349        operation: &OperationContext,
350        snapshot: ContextAuditSnapshot,
351    ) -> Result<(), PortFailure>;
352    async fn derivation(
353        &self,
354        operation: &OperationContext,
355        derivation: ContextAuditDerivation,
356    ) -> Result<(), PortFailure>;
357    async fn model_invocation(
358        &self,
359        operation: &OperationContext,
360        invocation: ModelInvocationAudit,
361    ) -> Result<(), PortFailure>;
362}
363
364#[derive(Debug, Clone, PartialEq, Eq)]
365pub struct WorkspaceEntry {
366    pub path: String,
367    pub is_dir: bool,
368    pub size: Option<u64>,
369}
370
371#[derive(Debug, Clone, PartialEq, Eq)]
372pub struct CommandRequest {
373    pub command: String,
374    pub cwd: Option<String>,
375}
376
377#[derive(Debug, Clone, PartialEq, Eq)]
378pub struct CommandOutput {
379    pub exit_code: i32,
380    pub stdout: String,
381    pub stderr: String,
382    pub truncated: bool,
383}
384
385#[async_trait]
386pub trait WorkspaceSession: Send + Sync {
387    async fn describe(&self, operation: &OperationContext) -> Result<String, PortFailure>;
388    async fn read(&self, operation: &OperationContext, path: &str) -> Result<Vec<u8>, PortFailure>;
389    async fn write(
390        &self,
391        operation: &OperationContext,
392        path: &str,
393        content: &[u8],
394    ) -> Result<(), PortFailure>;
395    async fn list(
396        &self,
397        operation: &OperationContext,
398        path: &str,
399    ) -> Result<Vec<WorkspaceEntry>, PortFailure>;
400    async fn search(
401        &self,
402        operation: &OperationContext,
403        path: &str,
404        query: &str,
405        limit: usize,
406    ) -> Result<Vec<String>, PortFailure>;
407    async fn execute(
408        &self,
409        operation: &OperationContext,
410        request: CommandRequest,
411    ) -> Result<CommandOutput, PortFailure>;
412}
413
414#[derive(Debug, Clone)]
415pub struct ResolvedExecutionContext {
416    pub runtime_instance_id: RuntimeInstanceId,
417    pub agent_id: String,
418    pub runtime_type: String,
419    pub model: String,
420    pub workspace_id: WorkspaceId,
421    pub definition_id: String,
422    pub definition_version: String,
423    pub definition_digest: Option<String>,
424    pub metadata: BTreeMap<String, String>,
425}
426
427#[async_trait]
428pub trait RuntimeInstanceResolver: Send + Sync {
429    async fn resolve(
430        &self,
431        operation: &OperationContext,
432        caller: &CallerScope,
433        id: &RuntimeInstanceId,
434    ) -> Result<ResolvedExecutionContext, PortFailure>;
435}
436
437#[async_trait]
438pub trait TraceSession: Send + Sync {
439    async fn append(
440        &self,
441        operation: &OperationContext,
442        event_type: &str,
443        payload: Value,
444    ) -> Result<(), PortFailure>;
445}
446
447#[derive(Debug, Clone, PartialEq, Eq)]
448pub struct InteractionRequest {
449    pub request_id: String,
450    pub prompt: String,
451}
452
453#[derive(Debug, Clone, PartialEq, Eq)]
454pub struct InteractionResponse {
455    pub text: String,
456}
457
458#[async_trait]
459pub trait InteractionSession: Send + Sync {
460    /// Register a pending request before it becomes externally visible.
461    async fn prepare(
462        &self,
463        operation: &OperationContext,
464        request: InteractionRequest,
465    ) -> Result<(), PortFailure>;
466
467    /// Wait for a response to a request registered by [`Self::prepare`].
468    async fn wait(
469        &self,
470        operation: &OperationContext,
471        request_id: &str,
472    ) -> Result<InteractionResponse, PortFailure>;
473
474    async fn request(
475        &self,
476        operation: &OperationContext,
477        request: InteractionRequest,
478    ) -> Result<InteractionResponse, PortFailure> {
479        let request_id = request.request_id.clone();
480        self.prepare(operation, request).await?;
481        self.wait(operation, &request_id).await
482    }
483}
484
485#[derive(Debug, Clone)]
486pub struct SubagentRequest {
487    pub request_id: String,
488    pub prompt: String,
489    pub max_model_turns: usize,
490}
491
492#[derive(Debug, Clone)]
493pub struct SubagentOutcome {
494    pub answer: String,
495}
496
497#[async_trait]
498pub trait SubagentSession: Send + Sync {
499    async fn execute(
500        &self,
501        operation: &OperationContext,
502        request: SubagentRequest,
503    ) -> Result<SubagentOutcome, PortFailure>;
504}
505
506#[derive(Debug, Clone)]
507pub struct ResourceSnapshot {
508    pub instructions: Vec<String>,
509    pub skills: BTreeMap<String, String>,
510    pub digest: String,
511}
512
513#[async_trait]
514pub trait ResourceSession: Send + Sync {
515    async fn load(&self, operation: &OperationContext) -> Result<ResourceSnapshot, PortFailure>;
516}
517
518#[derive(Debug, Clone)]
519pub struct SessionScope {
520    pub execution_id: runtime_types::ExecutionId,
521    pub conversation_id: Option<ConversationId>,
522    pub resolved: ResolvedExecutionContext,
523}
524
525#[derive(Debug, Clone, PartialEq, Eq)]
526pub struct ExecutionDelegationLease {
527    pub lease_ref: DelegationLeaseRef,
528    pub expires_at_seconds: u64,
529    pub revision: u64,
530}
531
532pub struct ExecutionSessions {
533    pub model: std::sync::Arc<dyn ModelSession>,
534    pub conversation: std::sync::Arc<dyn ConversationSession>,
535    pub context_audit: std::sync::Arc<dyn ContextAuditSession>,
536    pub workspace: std::sync::Arc<dyn WorkspaceSession>,
537    pub trace: std::sync::Arc<dyn TraceSession>,
538    pub interaction: std::sync::Arc<dyn InteractionSession>,
539    pub subagent: std::sync::Arc<dyn SubagentSession>,
540    pub resources: std::sync::Arc<dyn ResourceSession>,
541}
542
543#[async_trait]
544pub trait ExecutionSessionFactory: Send + Sync {
545    async fn establish_delegation(
546        &self,
547        operation: &OperationContext,
548        authority: &RequestAuthority,
549        scope: &SessionScope,
550    ) -> Result<ExecutionDelegationLease, PortFailure>;
551
552    async fn create(
553        &self,
554        operation: &OperationContext,
555        caller: &CallerScope,
556        delegation: &ExecutionDelegationLease,
557        scope: &SessionScope,
558    ) -> Result<ExecutionSessions, PortFailure>;
559
560    async fn renew_delegation(
561        &self,
562        operation: &OperationContext,
563        caller: &CallerScope,
564        delegation: &ExecutionDelegationLease,
565    ) -> Result<ExecutionDelegationLease, PortFailure>;
566
567    async fn revoke_delegation(
568        &self,
569        operation: &OperationContext,
570        caller: &CallerScope,
571        delegation: &ExecutionDelegationLease,
572    ) -> Result<(), PortFailure>;
573
574    /// Deliver input to a currently waiting execution. Implementations must
575    /// deduplicate `operation_id` and reject input when no request is pending.
576    async fn submit_input(
577        &self,
578        operation: &OperationContext,
579        caller: &CallerScope,
580        execution_id: &runtime_types::ExecutionId,
581        operation_id: &OperationId,
582        request_id: &str,
583        response: InteractionResponse,
584    ) -> Result<(), PortFailure>;
585}
agent_runtime_ports/lib.rs

agent_runtime_ports/
lib.rs
