monoloop_contracts/
transaction.rs

1//! Transaction request, events, terminal, sinks, and runtime port.
2
3use crate::canonical::CanonicalUnitEvent;
4use crate::config::{InvocationConfig, SessionConfig};
5use crate::id::ToolId;
6use crate::id::{ChannelId, SessionId, SessionKey, TransactionId};
7use crate::input::CanonicalInput;
8use crate::safe::SafeDiagnostic;
9use crate::tool::ToolLifecycleEvent;
10use serde::{Deserialize, Serialize};
11use std::future::Future;
12use std::pin::Pin;
13use std::sync::Arc;
14use std::time::Duration;
15use thiserror::Error;
16
17/// Future returned by event delivery (no async_trait required).
18pub type EventDelivery =
19    Pin<Box<dyn Future<Output = Result<(), EventDeliveryError>> + Send + 'static>>;
20
21/// Caller event sink (push-based).
22pub trait TransactionEventSink: Send + Sync + 'static {
23    /// Deliver one ordered event. Must return promptly with a future.
24    fn deliver(&self, event: TransactionEvent) -> EventDelivery;
25}
26
27/// Future returned by completion callback.
28pub type CompletionDelivery =
29    Pin<Box<dyn Future<Output = Result<(), CompletionDeliveryError>> + Send + 'static>>;
30
31/// One-shot completion callback.
32pub trait CompletionCallback: Send + 'static {
33    /// Invoke exactly once with the terminal result.
34    fn call(self: Box<Self>, end: TransactionEnd) -> CompletionDelivery;
35}
36
37/// Closure adapter for [`TransactionEventSink`].
38pub struct FnEventSink<F>(pub F);
39
40impl<F> TransactionEventSink for FnEventSink<F>
41where
42    F: Fn(TransactionEvent) -> EventDelivery + Send + Sync + 'static,
43{
44    fn deliver(&self, event: TransactionEvent) -> EventDelivery {
45        (self.0)(event)
46    }
47}
48
49/// Closure adapter for [`CompletionCallback`].
50pub struct FnCompletionCallback<F>(pub F);
51
52impl<F> CompletionCallback for FnCompletionCallback<F>
53where
54    F: FnOnce(TransactionEnd) -> CompletionDelivery + Send + 'static,
55{
56    fn call(self: Box<Self>, end: TransactionEnd) -> CompletionDelivery {
57        (self.0)(end)
58    }
59}
60
61/// Transaction submission request (synchronous admission; async progress).
62pub struct TransactionRequest {
63    /// Explicit Channel selection.
64    pub channel_id: ChannelId,
65    /// Existing session when known; `None` for new external create or direct-LLM generate.
66    pub session_id: Option<SessionId>,
67    /// Canonical input messages.
68    pub input: CanonicalInput,
69    /// Optional external-agent session configuration.
70    pub session_config: Option<SessionConfig>,
71    /// Invocation configuration.
72    pub invocation_config: InvocationConfig,
73    /// Selected host tool ids (deduplicated at admission).
74    pub tools: Vec<ToolId>,
75    /// Required event sink.
76    pub events: Arc<dyn TransactionEventSink>,
77    /// Required completion callback.
78    pub completion: Box<dyn CompletionCallback>,
79}
80
81/// Immediate admission receipt (no network performed).
82#[derive(Clone, Debug, PartialEq, Eq)]
83pub struct AdmissionReceipt {
84    /// Generated transaction id.
85    pub transaction_id: TransactionId,
86    /// Session id when already known (direct LLM or existing external).
87    pub session_id: Option<SessionId>,
88}
89
90/// How to address an in-flight transaction for control.
91#[derive(Clone, Debug, PartialEq, Eq, Hash)]
92pub enum TransactionSelector {
93    /// By transaction id (valid during external session creation).
94    Transaction(TransactionId),
95    /// By established session key.
96    Session(SessionKey),
97}
98
99/// Termination mode.
100#[derive(Clone, Debug, PartialEq, Eq)]
101pub enum TerminationMode {
102    /// Cooperative cancellation.
103    Cancel {
104        /// Reason.
105        reason: CancellationReason,
106    },
107    /// Forced terminate.
108    ForceTerminate {
109        /// Reason.
110        reason: TerminationReason,
111    },
112}
113
114/// Cancellation reason.
115#[derive(Clone, Debug, PartialEq, Eq)]
116pub struct CancellationReason {
117    /// Closed code.
118    pub code: CancellationReasonCode,
119    /// Optional safe detail.
120    pub detail: Option<SafeDiagnostic>,
121}
122
123/// Cancellation reason codes.
124#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
125pub enum CancellationReasonCode {
126    /// Caller requested cancel.
127    CallerRequested,
128    /// Runtime is shutting down.
129    RuntimeShutdown,
130}
131
132/// Force-termination reason.
133#[derive(Clone, Debug, PartialEq, Eq)]
134pub struct TerminationReason {
135    /// Closed code.
136    pub code: TerminationReasonCode,
137    /// Optional safe detail.
138    pub detail: Option<SafeDiagnostic>,
139}
140
141/// Termination reason codes.
142#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
143pub enum TerminationReasonCode {
144    /// Caller requested force.
145    CallerRequested,
146    /// Cancel grace expired.
147    CancellationGraceExpired,
148    /// Runtime is shutting down.
149    RuntimeShutdown,
150}
151
152/// Immediate disposition of a terminate request.
153#[derive(Clone, Copy, Debug, PartialEq, Eq)]
154pub enum TerminationDisposition {
155    /// Request accepted.
156    Accepted,
157    /// Already terminal or already requested.
158    AlreadyRequested,
159    /// Transaction already terminal.
160    AlreadyTerminal,
161    /// Unknown selector.
162    NotFound,
163}
164
165/// Shutdown future type.
166pub type Shutdown = Pin<Box<dyn Future<Output = ShutdownDisposition> + Send + 'static>>;
167
168/// Shutdown summary counts.
169#[derive(Clone, Debug, PartialEq, Eq, Default)]
170pub struct ShutdownDisposition {
171    /// Actors finalized normally.
172    pub normally_finalized: u64,
173    /// Supervisor claimed finalization after abort.
174    pub supervisor_finalized: u64,
175    /// Callback future failed.
176    pub callback_failed: u64,
177    /// Callback future aborted at deadline.
178    pub callback_aborted: u64,
179    /// Invariant failures during shutdown.
180    pub invariant_failed: u64,
181}
182
183/// Public transaction runtime port (implementation in monoloop-loop).
184pub trait TransactionRuntime: Send + Sync {
185    /// Synchronously admit a transaction or return a typed error.
186    fn submit(&self, request: TransactionRequest) -> Result<AdmissionReceipt, AdmissionError>;
187
188    /// Request cancellation or forced termination.
189    fn terminate(
190        &self,
191        selector: TransactionSelector,
192        mode: TerminationMode,
193    ) -> TerminationDisposition;
194
195    /// Drain and stop the runtime.
196    fn shutdown(&self, deadline: Duration) -> Shutdown;
197}
198
199/// Ordered transaction event.
200#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
201pub struct TransactionEvent {
202    /// Transaction id.
203    pub transaction_id: TransactionId,
204    /// Channel id.
205    pub channel_id: ChannelId,
206    /// Session id (established by this point for ordinary events).
207    pub session_id: SessionId,
208    /// Contiguous sequence starting at 1, including `Ended`.
209    pub sequence: u64,
210    /// Payload.
211    pub payload: TransactionEventPayload,
212}
213
214/// Event payload variants.
215#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
216pub enum TransactionEventPayload {
217    /// External session identity established.
218    SessionEstablished {
219        /// Authoritative external id.
220        external_session_id: crate::id::ExternalSessionId,
221    },
222    /// Complete canonical unit from Interpreter composition.
223    CanonicalUnit(CanonicalUnitEvent),
224    /// Host tool lifecycle.
225    ToolLifecycle(ToolLifecycleEvent),
226    /// Safe diagnostic.
227    Diagnostic(TransactionDiagnostic),
228    /// Terminal event (exactly once).
229    Ended(TransactionEnd),
230}
231
232/// Bounded safe transaction diagnostic.
233#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
234pub struct TransactionDiagnostic {
235    /// Safe diagnostic.
236    pub diagnostic: SafeDiagnostic,
237}
238
239/// Terminal transaction result.
240#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
241pub struct TransactionEnd {
242    /// Transaction id.
243    pub transaction_id: TransactionId,
244    /// Session when established.
245    pub session_id: Option<SessionId>,
246    /// Channel.
247    pub channel_id: ChannelId,
248    /// Terminal kind.
249    pub kind: TransactionEndKind,
250    /// Prior cause when terminal selection raced (optional).
251    pub prior_terminal_cause: Option<TransactionEndKind>,
252    /// Whether the terminal event was accepted by the sink.
253    pub event_delivery: EventDeliveryOutcome,
254    /// Number of events emitted including `Ended`.
255    pub emitted_events: u64,
256    /// Bounded usage facts.
257    pub usage: TransactionUsage,
258    /// Safe diagnostics.
259    pub diagnostics: Vec<TransactionDiagnostic>,
260}
261
262/// Closed terminal kinds.
263#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
264pub enum TransactionEndKind {
265    /// Successful completion.
266    Completed,
267    /// Caller must continue (caller-controlled policy).
268    ContinuationRequired,
269    /// Cancelled.
270    Cancelled,
271    /// Force-terminated.
272    Terminated,
273    /// Runtime shutdown.
274    RuntimeShutdown,
275    /// Deadline exceeded.
276    DeadlineExceeded,
277    /// Channel open/attach failed.
278    ChannelOpenFailed,
279    /// Outbound encoding failed.
280    EncodingFailed,
281    /// Connector failed.
282    ConnectorFailed,
283    /// Interpretation failed.
284    InterpretationFailed,
285    /// Tool exchange failed.
286    ToolExchangeFailed,
287    /// Event delivery failed.
288    EventDeliveryFailed,
289    /// Resource limit exceeded.
290    LimitExceeded,
291    /// Internal invariant failed.
292    InvariantFailed,
293}
294
295/// Terminal event delivery outcome.
296#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
297pub enum EventDeliveryOutcome {
298    /// Sink accepted.
299    Accepted,
300    /// Sink failed or timed out.
301    Failed,
302}
303
304/// Bounded usage facts (unavailable is not zero).
305#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Default)]
306pub struct TransactionUsage {
307    /// Provider input tokens when known.
308    pub provider_input_tokens: Option<u64>,
309    /// Provider output tokens when known.
310    pub provider_output_tokens: Option<u64>,
311    /// Number of provider exchanges.
312    pub provider_exchanges: u32,
313    /// Number of tool executions started.
314    pub tools_started: u32,
315    /// Number of tool executions completed (success or domain failure).
316    pub tools_completed: u32,
317}
318
319/// Event delivery error (safe).
320#[derive(Clone, Debug, Error, PartialEq, Eq)]
321pub enum EventDeliveryError {
322    /// Sink rejected or failed.
323    #[error("event delivery failed")]
324    Failed,
325    /// Delivery deadline exceeded.
326    #[error("event delivery deadline exceeded")]
327    DeadlineExceeded,
328}
329
330/// Completion callback delivery error (safe).
331#[derive(Clone, Debug, Error, PartialEq, Eq)]
332pub enum CompletionDeliveryError {
333    /// Callback failed.
334    #[error("completion callback failed")]
335    Failed,
336    /// Callback deadline exceeded.
337    #[error("completion callback deadline exceeded")]
338    DeadlineExceeded,
339}
340
341/// Synchronous admission error.
342#[derive(Clone, Debug, Error, PartialEq, Eq)]
343#[error("{kind:?}: {message}")]
344pub struct AdmissionError {
345    /// Closed kind.
346    pub kind: AdmissionErrorKind,
347    /// Safe bounded message.
348    pub message: String,
349}
350
351impl AdmissionError {
352    /// Construct an admission error.
353    pub fn new(kind: AdmissionErrorKind, message: impl Into<String>) -> Self {
354        Self {
355            kind,
356            message: message.into(),
357        }
358    }
359}
360
361/// Admission error kinds.
362#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
363pub enum AdmissionErrorKind {
364    /// Runtime not accepting.
365    RuntimeShuttingDown,
366    /// Unknown Channel id.
367    UnknownChannel,
368    /// Session already has an active transaction.
369    SessionAlreadyActive,
370    /// Unknown tool id.
371    UnknownTool,
372    /// Duplicate tool id in request.
373    DuplicateTool,
374    /// Invalid canonical input.
375    InvalidInput,
376    /// Invalid configuration merge.
377    InvalidConfiguration,
378    /// Capability mismatch for Channel/tools/session.
379    CapabilityMismatch,
380    /// Capacity exceeded.
381    CapacityExceeded,
382    /// Actor spawn failed.
383    SpawnFailed,
384}
385
386#[cfg(test)]
387mod tests {
388    use super::*;
389    use crate::input::user_text_input;
390
391    #[test]
392    fn end_kind_round_trip() {
393        let kind = TransactionEndKind::Completed;
394        let json = serde_json::to_string(&kind).unwrap();
395        let back: TransactionEndKind = serde_json::from_str(&json).unwrap();
396        assert_eq!(kind, back);
397    }
398
399    #[tokio::test]
400    async fn sink_adapters_return_futures() {
401        let sink = FnEventSink(|_e| Box::pin(async { Ok(()) }) as EventDelivery);
402        let events: Arc<dyn TransactionEventSink> = Arc::new(sink);
403        let end = TransactionEnd {
404            transaction_id: TransactionId::generate(),
405            session_id: None,
406            channel_id: ChannelId::try_new("ch").unwrap(),
407            kind: TransactionEndKind::Completed,
408            prior_terminal_cause: None,
409            event_delivery: EventDeliveryOutcome::Accepted,
410            emitted_events: 1,
411            usage: TransactionUsage::default(),
412            diagnostics: vec![],
413        };
414        let ev = TransactionEvent {
415            transaction_id: end.transaction_id,
416            channel_id: end.channel_id.clone(),
417            session_id: SessionId::try_new("s").unwrap(),
418            sequence: 1,
419            payload: TransactionEventPayload::Ended(end.clone()),
420        };
421        events.deliver(ev).await.unwrap();
422
423        let cb: Box<dyn CompletionCallback> = Box::new(FnCompletionCallback(|_e| {
424            Box::pin(async { Ok(()) }) as CompletionDelivery
425        }));
426        cb.call(end).await.unwrap();
427
428        let _input = user_text_input("hello").unwrap();
429    }
430}
monoloop_contracts/transaction.rs

monoloop_contracts/
transaction.rs
