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///
216/// Live assistant text arrives only as [`Self::CanonicalUnit`] (complete units).
217/// There is **no** token / delta stream on this port.
218#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
219pub enum TransactionEventPayload {
220    /// External session identity established.
221    SessionEstablished {
222        /// Authoritative external id.
223        external_session_id: crate::id::ExternalSessionId,
224    },
225    /// Complete canonical unit from Interpreter composition (not a token delta).
226    CanonicalUnit(CanonicalUnitEvent),
227    /// Host tool lifecycle.
228    ToolLifecycle(ToolLifecycleEvent),
229    /// Safe diagnostic.
230    Diagnostic(TransactionDiagnostic),
231    /// Terminal event (exactly once).
232    Ended(TransactionEnd),
233}
234
235/// Bounded safe transaction diagnostic.
236#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
237pub struct TransactionDiagnostic {
238    /// Safe diagnostic.
239    pub diagnostic: SafeDiagnostic,
240}
241
242/// Terminal transaction result.
243#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
244pub struct TransactionEnd {
245    /// Transaction id.
246    pub transaction_id: TransactionId,
247    /// Session when established.
248    pub session_id: Option<SessionId>,
249    /// Channel.
250    pub channel_id: ChannelId,
251    /// Terminal kind.
252    pub kind: TransactionEndKind,
253    /// Prior cause when terminal selection raced (optional).
254    pub prior_terminal_cause: Option<TransactionEndKind>,
255    /// Whether the terminal event was accepted by the sink.
256    pub event_delivery: EventDeliveryOutcome,
257    /// Number of events emitted including `Ended`.
258    pub emitted_events: u64,
259    /// Bounded usage facts.
260    pub usage: TransactionUsage,
261    /// Safe diagnostics.
262    pub diagnostics: Vec<TransactionDiagnostic>,
263}
264
265/// Closed terminal kinds.
266#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
267pub enum TransactionEndKind {
268    /// Successful completion.
269    Completed,
270    /// Caller must continue (caller-controlled policy).
271    ContinuationRequired,
272    /// Cancelled.
273    Cancelled,
274    /// Force-terminated.
275    Terminated,
276    /// Runtime shutdown.
277    RuntimeShutdown,
278    /// Deadline exceeded.
279    DeadlineExceeded,
280    /// Channel open/attach failed.
281    ChannelOpenFailed,
282    /// Outbound encoding failed.
283    EncodingFailed,
284    /// Connector failed.
285    ConnectorFailed,
286    /// Interpretation failed.
287    InterpretationFailed,
288    /// Tool exchange failed.
289    ToolExchangeFailed,
290    /// Event delivery failed.
291    EventDeliveryFailed,
292    /// Resource limit exceeded.
293    LimitExceeded,
294    /// Internal invariant failed.
295    InvariantFailed,
296}
297
298/// Terminal event delivery outcome.
299#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
300pub enum EventDeliveryOutcome {
301    /// Sink accepted.
302    Accepted,
303    /// Sink failed or timed out.
304    Failed,
305}
306
307/// Bounded usage facts (unavailable is not zero).
308#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Default)]
309pub struct TransactionUsage {
310    /// Provider input tokens when known.
311    pub provider_input_tokens: Option<u64>,
312    /// Provider output tokens when known.
313    pub provider_output_tokens: Option<u64>,
314    /// Number of provider exchanges.
315    pub provider_exchanges: u32,
316    /// Number of tool executions started.
317    pub tools_started: u32,
318    /// Number of tool executions completed (success or domain failure).
319    pub tools_completed: u32,
320}
321
322/// Event delivery error (safe).
323#[derive(Clone, Debug, Error, PartialEq, Eq)]
324pub enum EventDeliveryError {
325    /// Sink rejected or failed.
326    #[error("event delivery failed")]
327    Failed,
328    /// Delivery deadline exceeded.
329    #[error("event delivery deadline exceeded")]
330    DeadlineExceeded,
331}
332
333/// Completion callback delivery error (safe).
334#[derive(Clone, Debug, Error, PartialEq, Eq)]
335pub enum CompletionDeliveryError {
336    /// Callback failed.
337    #[error("completion callback failed")]
338    Failed,
339    /// Callback deadline exceeded.
340    #[error("completion callback deadline exceeded")]
341    DeadlineExceeded,
342}
343
344/// Synchronous admission error.
345#[derive(Clone, Debug, Error, PartialEq, Eq)]
346#[error("{kind:?}: {message}")]
347pub struct AdmissionError {
348    /// Closed kind.
349    pub kind: AdmissionErrorKind,
350    /// Safe bounded message.
351    pub message: String,
352}
353
354impl AdmissionError {
355    /// Construct an admission error.
356    pub fn new(kind: AdmissionErrorKind, message: impl Into<String>) -> Self {
357        Self {
358            kind,
359            message: message.into(),
360        }
361    }
362}
363
364/// Admission error kinds.
365#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
366pub enum AdmissionErrorKind {
367    /// Runtime not accepting.
368    RuntimeShuttingDown,
369    /// Unknown Channel id.
370    UnknownChannel,
371    /// Session already has an active transaction.
372    SessionAlreadyActive,
373    /// Unknown tool id.
374    UnknownTool,
375    /// Duplicate tool id in request.
376    DuplicateTool,
377    /// Invalid canonical input.
378    InvalidInput,
379    /// Invalid configuration merge.
380    InvalidConfiguration,
381    /// Capability mismatch for Channel/tools/session.
382    CapabilityMismatch,
383    /// Capacity exceeded.
384    CapacityExceeded,
385    /// Actor spawn failed.
386    SpawnFailed,
387}
388
389#[cfg(test)]
390mod tests {
391    use super::*;
392    use crate::input::user_text_input;
393
394    #[test]
395    fn end_kind_round_trip() {
396        let kind = TransactionEndKind::Completed;
397        let json = serde_json::to_string(&kind).unwrap();
398        let back: TransactionEndKind = serde_json::from_str(&json).unwrap();
399        assert_eq!(kind, back);
400    }
401
402    #[tokio::test]
403    async fn sink_adapters_return_futures() {
404        let sink = FnEventSink(|_e| Box::pin(async { Ok(()) }) as EventDelivery);
405        let events: Arc<dyn TransactionEventSink> = Arc::new(sink);
406        let end = TransactionEnd {
407            transaction_id: TransactionId::generate(),
408            session_id: None,
409            channel_id: ChannelId::try_new("ch").unwrap(),
410            kind: TransactionEndKind::Completed,
411            prior_terminal_cause: None,
412            event_delivery: EventDeliveryOutcome::Accepted,
413            emitted_events: 1,
414            usage: TransactionUsage::default(),
415            diagnostics: vec![],
416        };
417        let ev = TransactionEvent {
418            transaction_id: end.transaction_id,
419            channel_id: end.channel_id.clone(),
420            session_id: SessionId::try_new("s").unwrap(),
421            sequence: 1,
422            payload: TransactionEventPayload::Ended(end.clone()),
423        };
424        events.deliver(ev).await.unwrap();
425
426        let cb: Box<dyn CompletionCallback> = Box::new(FnCompletionCallback(|_e| {
427            Box::pin(async { Ok(()) }) as CompletionDelivery
428        }));
429        cb.call(end).await.unwrap();
430
431        let _input = user_text_input("hello").unwrap();
432    }
433}
monoloop_contracts/transaction.rs

monoloop_contracts/
transaction.rs
