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 agent_id: String,
395    pub runtime_type: String,
396    pub model: String,
397    pub workspace_id: WorkspaceId,
398    pub package_id: String,
399    pub package_version: String,
400    pub package_digest: Option<String>,
401    pub metadata: BTreeMap<String, String>,
402}
403
404#[async_trait]
405pub trait RuntimeInstanceResolver: Send + Sync {
406    async fn resolve(
407        &self,
408        operation: &OperationContext,
409        caller: &CallerScope,
410        id: &RuntimeInstanceId,
411    ) -> Result<ResolvedExecutionContext, PortFailure>;
412}
413
414#[async_trait]
415pub trait TraceSession: Send + Sync {
416    async fn append(
417        &self,
418        operation: &OperationContext,
419        event_type: &str,
420        payload: Value,
421    ) -> Result<(), PortFailure>;
422}
423
424#[derive(Debug, Clone, PartialEq, Eq)]
425pub struct InteractionRequest {
426    pub request_id: String,
427    pub prompt: String,
428}
429
430#[derive(Debug, Clone, PartialEq, Eq)]
431pub struct InteractionResponse {
432    pub text: String,
433}
434
435#[async_trait]
436pub trait InteractionSession: Send + Sync {
437    /// Register a pending request before it becomes externally visible.
438    async fn prepare(
439        &self,
440        operation: &OperationContext,
441        request: InteractionRequest,
442    ) -> Result<(), PortFailure>;
443
444    /// Wait for a response to a request registered by [`Self::prepare`].
445    async fn wait(
446        &self,
447        operation: &OperationContext,
448        request_id: &str,
449    ) -> Result<InteractionResponse, PortFailure>;
450
451    async fn request(
452        &self,
453        operation: &OperationContext,
454        request: InteractionRequest,
455    ) -> Result<InteractionResponse, PortFailure> {
456        let request_id = request.request_id.clone();
457        self.prepare(operation, request).await?;
458        self.wait(operation, &request_id).await
459    }
460}
461
462#[derive(Debug, Clone)]
463pub struct SubagentRequest {
464    pub request_id: String,
465    pub prompt: String,
466    pub max_model_turns: usize,
467}
468
469#[derive(Debug, Clone)]
470pub struct SubagentOutcome {
471    pub answer: String,
472}
473
474#[async_trait]
475pub trait SubagentSession: Send + Sync {
476    async fn execute(
477        &self,
478        operation: &OperationContext,
479        request: SubagentRequest,
480    ) -> Result<SubagentOutcome, PortFailure>;
481}
482
483#[derive(Debug, Clone)]
484pub struct ResourceSnapshot {
485    pub instructions: Vec<String>,
486    pub skills: BTreeMap<String, String>,
487    pub digest: String,
488}
489
490#[async_trait]
491pub trait ResourceSession: Send + Sync {
492    async fn load(&self, operation: &OperationContext) -> Result<ResourceSnapshot, PortFailure>;
493}
494
495#[derive(Debug, Clone)]
496pub struct SessionScope {
497    pub execution_id: runtime_types::ExecutionId,
498    pub conversation_id: Option<ConversationId>,
499    pub resolved: ResolvedExecutionContext,
500}
501
502#[derive(Debug, Clone, PartialEq, Eq)]
503pub struct ExecutionDelegationLease {
504    pub lease_ref: DelegationLeaseRef,
505    pub expires_at_seconds: u64,
506    pub revision: u64,
507}
508
509pub struct ExecutionSessions {
510    pub model: std::sync::Arc<dyn ModelSession>,
511    pub conversation: std::sync::Arc<dyn ConversationSession>,
512    pub context_audit: std::sync::Arc<dyn ContextAuditSession>,
513    pub workspace: std::sync::Arc<dyn WorkspaceSession>,
514    pub trace: std::sync::Arc<dyn TraceSession>,
515    pub interaction: std::sync::Arc<dyn InteractionSession>,
516    pub subagent: std::sync::Arc<dyn SubagentSession>,
517    pub resources: std::sync::Arc<dyn ResourceSession>,
518}
519
520#[async_trait]
521pub trait ExecutionSessionFactory: Send + Sync {
522    async fn establish_delegation(
523        &self,
524        operation: &OperationContext,
525        authority: &RequestAuthority,
526        scope: &SessionScope,
527    ) -> Result<ExecutionDelegationLease, PortFailure>;
528
529    async fn create(
530        &self,
531        operation: &OperationContext,
532        caller: &CallerScope,
533        delegation: &ExecutionDelegationLease,
534        scope: &SessionScope,
535    ) -> Result<ExecutionSessions, PortFailure>;
536
537    async fn renew_delegation(
538        &self,
539        operation: &OperationContext,
540        caller: &CallerScope,
541        delegation: &ExecutionDelegationLease,
542    ) -> Result<ExecutionDelegationLease, PortFailure>;
543
544    async fn revoke_delegation(
545        &self,
546        operation: &OperationContext,
547        caller: &CallerScope,
548        delegation: &ExecutionDelegationLease,
549    ) -> Result<(), PortFailure>;
550
551    /// Deliver input to a currently waiting execution. Implementations must
552    /// deduplicate `operation_id` and reject input when no request is pending.
553    async fn submit_input(
554        &self,
555        operation: &OperationContext,
556        caller: &CallerScope,
557        execution_id: &runtime_types::ExecutionId,
558        operation_id: &OperationId,
559        request_id: &str,
560        response: InteractionResponse,
561    ) -> Result<(), PortFailure>;
562}
agent_runtime_ports/lib.rs

agent_runtime_ports/
lib.rs
