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    pub max_output_tokens: Option<u32>,
157}
158
159#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
160#[serde(rename_all = "snake_case")]
161pub enum ModelFinish {
162    Stop,
163    ToolCalls,
164    Length,
165    ContentFilter,
166    Other,
167}
168
169impl ModelFinish {
170    pub fn is_complete(self) -> bool {
171        matches!(self, Self::Stop | Self::ToolCalls)
172    }
173}
174
175#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
176#[serde(rename_all = "camelCase")]
177pub struct TokenUsage {
178    /// `None` means the provider did not report this value. It must never be
179    /// collapsed to zero because zero is a real, materially different usage.
180    pub input_tokens: Option<u64>,
181    pub output_tokens: Option<u64>,
182}
183
184impl TokenUsage {
185    /// Additive identity used only when beginning a known aggregate.
186    pub const fn zero() -> Self {
187        Self {
188            input_tokens: Some(0),
189            output_tokens: Some(0),
190        }
191    }
192
193    /// Aggregate without inventing missing provider measurements. Once any
194    /// turn is unknown, the corresponding execution total stays unknown.
195    pub fn add_assign(&mut self, value: &Self) {
196        self.input_tokens = add_known(self.input_tokens, value.input_tokens);
197        self.output_tokens = add_known(self.output_tokens, value.output_tokens);
198    }
199}
200
201fn add_known(left: Option<u64>, right: Option<u64>) -> Option<u64> {
202    left.zip(right)
203        .map(|(left, right)| left.saturating_add(right))
204}
205
206#[cfg(test)]
207mod token_usage_tests {
208    use super::TokenUsage;
209
210    #[test]
211    fn aggregate_preserves_unknown_and_saturates_known_values() {
212        let mut usage = TokenUsage::zero();
213        usage.add_assign(&TokenUsage {
214            input_tokens: Some(u64::MAX),
215            output_tokens: Some(3),
216        });
217        usage.add_assign(&TokenUsage {
218            input_tokens: Some(1),
219            output_tokens: None,
220        });
221        assert_eq!(usage.input_tokens, Some(u64::MAX));
222        assert_eq!(usage.output_tokens, None);
223    }
224}
225
226#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
227#[serde(rename_all = "camelCase")]
228pub struct ModelResponse {
229    pub output: Vec<ModelContent>,
230    pub finish: ModelFinish,
231    pub usage: TokenUsage,
232}
233
234#[async_trait]
235pub trait ModelSession: Send + Sync {
236    async fn invoke(
237        &self,
238        operation: &OperationContext,
239        request: ModelRequest,
240    ) -> Result<ModelResponse, PortFailure>;
241}
242
243#[derive(Debug, Clone, PartialEq, Eq)]
244pub struct ConversationMessage {
245    pub role: ModelRole,
246    pub text: String,
247}
248
249#[async_trait]
250pub trait ConversationSession: Send + Sync {
251    async fn load_recent(
252        &self,
253        operation: &OperationContext,
254        limit: usize,
255    ) -> Result<Vec<ConversationMessage>, PortFailure>;
256    async fn append(
257        &self,
258        operation: &OperationContext,
259        messages: &[ConversationMessage],
260    ) -> Result<(), PortFailure>;
261}
262
263#[derive(Debug, Clone, Copy, PartialEq, Eq)]
264pub enum ContextAuditStage {
265    Original,
266    Assembled,
267    Compressed,
268}
269
270#[derive(Debug, Clone)]
271pub struct ContextAuditSnapshot {
272    pub id: String,
273    pub stage: ContextAuditStage,
274    pub purpose: String,
275    pub messages: Vec<ModelMessage>,
276    pub tools: Vec<ModelToolDefinition>,
277    pub protocol_overhead_tokens: u64,
278    pub reserved_output_tokens: u64,
279    pub estimated_tokens: u64,
280    /// Digest of the exact messages, Tool projection, and output reservation
281    /// used for the corresponding model request.
282    pub request_digest: String,
283}
284
285#[derive(Debug, Clone, Copy, PartialEq, Eq)]
286pub enum ContextDerivationKind {
287    Assemble,
288    Compress,
289    Trim,
290    Reorder,
291}
292
293#[derive(Debug, Clone)]
294pub struct ContextAuditDerivation {
295    pub id: String,
296    pub source_snapshot_id: String,
297    pub target_snapshot_id: String,
298    pub kind: ContextDerivationKind,
299    pub input_tokens: u64,
300    pub output_tokens: u64,
301    pub input_items: usize,
302    pub output_items: usize,
303}
304
305#[derive(Debug, Clone, Copy, PartialEq, Eq)]
306pub enum ModelInvocationAuditState {
307    Prepared,
308    Succeeded,
309    Failed,
310}
311
312#[derive(Debug, Clone)]
313pub struct ModelInvocationAudit {
314    pub id: String,
315    pub request_snapshot_id: String,
316    pub model: String,
317    pub state: ModelInvocationAuditState,
318    pub usage: TokenUsage,
319    pub error_code: Option<String>,
320}
321
322#[async_trait]
323pub trait ContextAuditSession: Send + Sync {
324    async fn snapshot(
325        &self,
326        operation: &OperationContext,
327        snapshot: ContextAuditSnapshot,
328    ) -> Result<(), PortFailure>;
329    async fn derivation(
330        &self,
331        operation: &OperationContext,
332        derivation: ContextAuditDerivation,
333    ) -> Result<(), PortFailure>;
334    async fn model_invocation(
335        &self,
336        operation: &OperationContext,
337        invocation: ModelInvocationAudit,
338    ) -> Result<(), PortFailure>;
339}
340
341#[derive(Debug, Clone, PartialEq, Eq)]
342pub struct WorkspaceEntry {
343    pub path: String,
344    pub is_dir: bool,
345    pub size: Option<u64>,
346}
347
348#[derive(Debug, Clone, PartialEq, Eq)]
349pub struct CommandRequest {
350    pub command: String,
351    pub cwd: Option<String>,
352}
353
354#[derive(Debug, Clone, PartialEq, Eq)]
355pub struct CommandOutput {
356    pub exit_code: i32,
357    pub stdout: String,
358    pub stderr: String,
359    pub truncated: bool,
360}
361
362#[async_trait]
363pub trait WorkspaceSession: Send + Sync {
364    async fn describe(&self, operation: &OperationContext) -> Result<String, PortFailure>;
365    async fn read(&self, operation: &OperationContext, path: &str) -> Result<Vec<u8>, PortFailure>;
366    async fn write(
367        &self,
368        operation: &OperationContext,
369        path: &str,
370        content: &[u8],
371    ) -> Result<(), PortFailure>;
372    async fn list(
373        &self,
374        operation: &OperationContext,
375        path: &str,
376    ) -> Result<Vec<WorkspaceEntry>, PortFailure>;
377    async fn search(
378        &self,
379        operation: &OperationContext,
380        path: &str,
381        query: &str,
382        limit: usize,
383    ) -> Result<Vec<String>, PortFailure>;
384    async fn execute(
385        &self,
386        operation: &OperationContext,
387        request: CommandRequest,
388    ) -> Result<CommandOutput, PortFailure>;
389}
390
391#[derive(Debug, Clone)]
392pub struct ResolvedExecutionContext {
393    pub runtime_instance_id: RuntimeInstanceId,
394    pub runtime_type: String,
395    pub model: String,
396    pub workspace_id: WorkspaceId,
397    pub package_id: String,
398    pub package_version: String,
399    pub package_digest: Option<String>,
400    pub metadata: BTreeMap<String, String>,
401}
402
403#[async_trait]
404pub trait RuntimeInstanceResolver: Send + Sync {
405    async fn resolve(
406        &self,
407        operation: &OperationContext,
408        caller: &CallerScope,
409        id: &RuntimeInstanceId,
410    ) -> Result<ResolvedExecutionContext, PortFailure>;
411}
412
413#[async_trait]
414pub trait TraceSession: Send + Sync {
415    async fn append(
416        &self,
417        operation: &OperationContext,
418        event_type: &str,
419        payload: Value,
420    ) -> Result<(), PortFailure>;
421}
422
423#[derive(Debug, Clone, PartialEq, Eq)]
424pub struct InteractionRequest {
425    pub request_id: String,
426    pub prompt: String,
427}
428
429#[derive(Debug, Clone, PartialEq, Eq)]
430pub struct InteractionResponse {
431    pub text: String,
432}
433
434#[async_trait]
435pub trait InteractionSession: Send + Sync {
436    /// Register a pending request before it becomes externally visible.
437    async fn prepare(
438        &self,
439        operation: &OperationContext,
440        request: InteractionRequest,
441    ) -> Result<(), PortFailure>;
442
443    /// Wait for a response to a request registered by [`Self::prepare`].
444    async fn wait(
445        &self,
446        operation: &OperationContext,
447        request_id: &str,
448    ) -> Result<InteractionResponse, PortFailure>;
449
450    async fn request(
451        &self,
452        operation: &OperationContext,
453        request: InteractionRequest,
454    ) -> Result<InteractionResponse, PortFailure> {
455        let request_id = request.request_id.clone();
456        self.prepare(operation, request).await?;
457        self.wait(operation, &request_id).await
458    }
459}
460
461#[derive(Debug, Clone)]
462pub struct SubagentRequest {
463    pub request_id: String,
464    pub prompt: String,
465    pub max_model_turns: usize,
466}
467
468#[derive(Debug, Clone)]
469pub struct SubagentOutcome {
470    pub answer: String,
471}
472
473#[async_trait]
474pub trait SubagentSession: Send + Sync {
475    async fn execute(
476        &self,
477        operation: &OperationContext,
478        request: SubagentRequest,
479    ) -> Result<SubagentOutcome, PortFailure>;
480}
481
482#[derive(Debug, Clone)]
483pub struct ResourceSnapshot {
484    pub instructions: Vec<String>,
485    pub skills: BTreeMap<String, String>,
486    pub digest: String,
487}
488
489#[async_trait]
490pub trait ResourceSession: Send + Sync {
491    async fn load(&self, operation: &OperationContext) -> Result<ResourceSnapshot, PortFailure>;
492}
493
494#[derive(Debug, Clone)]
495pub struct SessionScope {
496    pub execution_id: runtime_types::ExecutionId,
497    pub conversation_id: Option<ConversationId>,
498    pub resolved: ResolvedExecutionContext,
499}
500
501#[derive(Debug, Clone, PartialEq, Eq)]
502pub struct ExecutionDelegationLease {
503    pub lease_ref: DelegationLeaseRef,
504    pub expires_at_seconds: u64,
505    pub revision: u64,
506}
507
508pub struct ExecutionSessions {
509    pub model: std::sync::Arc<dyn ModelSession>,
510    pub conversation: std::sync::Arc<dyn ConversationSession>,
511    pub context_audit: std::sync::Arc<dyn ContextAuditSession>,
512    pub workspace: std::sync::Arc<dyn WorkspaceSession>,
513    pub trace: std::sync::Arc<dyn TraceSession>,
514    pub interaction: std::sync::Arc<dyn InteractionSession>,
515    pub subagent: std::sync::Arc<dyn SubagentSession>,
516    pub resources: std::sync::Arc<dyn ResourceSession>,
517}
518
519#[async_trait]
520pub trait ExecutionSessionFactory: Send + Sync {
521    async fn establish_delegation(
522        &self,
523        operation: &OperationContext,
524        authority: &RequestAuthority,
525        scope: &SessionScope,
526    ) -> Result<ExecutionDelegationLease, PortFailure>;
527
528    async fn create(
529        &self,
530        operation: &OperationContext,
531        caller: &CallerScope,
532        delegation: &ExecutionDelegationLease,
533        scope: &SessionScope,
534    ) -> Result<ExecutionSessions, PortFailure>;
535
536    async fn renew_delegation(
537        &self,
538        operation: &OperationContext,
539        caller: &CallerScope,
540        delegation: &ExecutionDelegationLease,
541    ) -> Result<ExecutionDelegationLease, PortFailure>;
542
543    async fn revoke_delegation(
544        &self,
545        operation: &OperationContext,
546        caller: &CallerScope,
547        delegation: &ExecutionDelegationLease,
548    ) -> Result<(), PortFailure>;
549
550    /// Deliver input to a currently waiting execution. Implementations must
551    /// deduplicate `operation_id` and reject input when no request is pending.
552    async fn submit_input(
553        &self,
554        operation: &OperationContext,
555        caller: &CallerScope,
556        execution_id: &runtime_types::ExecutionId,
557        operation_id: &OperationId,
558        request_id: &str,
559        response: InteractionResponse,
560    ) -> Result<(), PortFailure>;
561}
agent_runtime_ports/lib.rs

agent_runtime_ports/
lib.rs
