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 thiserror::Error;
14
15/// Future returned by event delivery (no async_trait required).
16pub type EventDelivery =
17    Pin<Box<dyn Future<Output = Result<(), EventDeliveryError>> + Send + 'static>>;
18
19/// Caller event sink (push-based).
20pub trait TransactionEventSink: Send + Sync + 'static {
21    /// Deliver one ordered event. Must return promptly with a future.
22    fn deliver(&self, event: TransactionEvent) -> EventDelivery;
23}
24
25/// Future returned by completion callback.
26pub type CompletionDelivery =
27    Pin<Box<dyn Future<Output = Result<(), CompletionDeliveryError>> + Send + 'static>>;
28
29/// One-shot completion callback.
30pub trait CompletionCallback: Send + 'static {
31    /// Invoke exactly once with the terminal result.
32    fn call(self: Box<Self>, end: TransactionEnd) -> CompletionDelivery;
33}
34
35/// Closure adapter for [`TransactionEventSink`].
36pub struct FnEventSink<F>(pub F);
37
38impl<F> TransactionEventSink for FnEventSink<F>
39where
40    F: Fn(TransactionEvent) -> EventDelivery + Send + Sync + 'static,
41{
42    fn deliver(&self, event: TransactionEvent) -> EventDelivery {
43        (self.0)(event)
44    }
45}
46
47/// Closure adapter for [`CompletionCallback`].
48pub struct FnCompletionCallback<F>(pub F);
49
50impl<F> CompletionCallback for FnCompletionCallback<F>
51where
52    F: FnOnce(TransactionEnd) -> CompletionDelivery + Send + 'static,
53{
54    fn call(self: Box<Self>, end: TransactionEnd) -> CompletionDelivery {
55        (self.0)(end)
56    }
57}
58
59/// Runtime v2 submission request — concrete mailboxes, no host traits in-core.
60pub struct TransactionSubmitRequest {
61    /// Explicit Channel selection.
62    pub channel_id: ChannelId,
63    /// Existing session when known; `None` for new external create or direct-LLM generate.
64    pub session_id: Option<SessionId>,
65    /// Canonical input messages.
66    pub input: CanonicalInput,
67    /// Optional external-agent session configuration.
68    pub session_config: Option<SessionConfig>,
69    /// Invocation configuration.
70    pub invocation_config: InvocationConfig,
71    /// Selected host tool ids (deduplicated at admission).
72    pub tools: Vec<ToolId>,
73    /// Library-created delivery ports (caller holds the receiver half).
74    pub delivery: crate::delivery::TransactionDelivery,
75}
76
77/// Immediate admission receipt (no network performed).
78#[derive(Clone, Debug, PartialEq, Eq)]
79pub struct AdmissionReceipt {
80    /// Generated transaction id.
81    pub transaction_id: TransactionId,
82    /// Session id when already known (direct LLM or existing external).
83    pub session_id: Option<SessionId>,
84}
85
86/// How to address an in-flight transaction for control.
87#[derive(Clone, Debug, PartialEq, Eq, Hash)]
88pub enum TransactionSelector {
89    /// By transaction id (valid during external session creation).
90    Transaction(TransactionId),
91    /// By established session key.
92    Session(SessionKey),
93}
94
95/// Termination mode.
96#[derive(Clone, Debug, PartialEq, Eq)]
97pub enum TerminationMode {
98    /// Cooperative cancellation.
99    Cancel {
100        /// Reason.
101        reason: CancellationReason,
102    },
103    /// Forced terminate.
104    ForceTerminate {
105        /// Reason.
106        reason: TerminationReason,
107    },
108}
109
110/// Cancellation reason.
111#[derive(Clone, Debug, PartialEq, Eq)]
112pub struct CancellationReason {
113    /// Closed code.
114    pub code: CancellationReasonCode,
115    /// Optional safe detail.
116    pub detail: Option<SafeDiagnostic>,
117}
118
119/// Cancellation reason codes.
120#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
121pub enum CancellationReasonCode {
122    /// Caller requested cancel.
123    CallerRequested,
124    /// Runtime is shutting down.
125    RuntimeShutdown,
126}
127
128/// Force-termination reason.
129#[derive(Clone, Debug, PartialEq, Eq)]
130pub struct TerminationReason {
131    /// Closed code.
132    pub code: TerminationReasonCode,
133    /// Optional safe detail.
134    pub detail: Option<SafeDiagnostic>,
135}
136
137/// Termination reason codes.
138#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
139pub enum TerminationReasonCode {
140    /// Caller requested force.
141    CallerRequested,
142    /// Cancel grace expired.
143    CancellationGraceExpired,
144    /// Runtime is shutting down.
145    RuntimeShutdown,
146}
147
148/// Immediate disposition of a terminate request.
149#[derive(Clone, Copy, Debug, PartialEq, Eq)]
150pub enum TerminationDisposition {
151    /// Request accepted.
152    Accepted,
153    /// Already terminal or already requested.
154    AlreadyRequested,
155    /// Transaction already terminal.
156    AlreadyTerminal,
157    /// Unknown selector.
158    NotFound,
159    /// Control queue was full — request not enqueued (Law 22 fail-closed; D-039).
160    ///
161    /// This is **not** [`Self::AlreadyTerminal`]: the transaction may still be live.
162    ControlCapacityExceeded,
163    /// Control queue closed (runtime stopping / stopped) — request not enqueued.
164    RuntimeClosed,
165}
166
167/// Shutdown future type.
168pub type Shutdown = Pin<Box<dyn Future<Output = ShutdownDisposition> + Send + 'static>>;
169
170/// Shutdown summary counts.
171#[derive(Clone, Debug, PartialEq, Eq, Default)]
172pub struct ShutdownDisposition {
173    /// Actors finalized normally.
174    pub normally_finalized: u64,
175    /// Supervisor claimed finalization after abort.
176    pub supervisor_finalized: u64,
177    /// Callback future failed.
178    pub callback_failed: u64,
179    /// Callback future aborted at deadline.
180    pub callback_aborted: u64,
181    /// Invariant failures during shutdown.
182    pub invariant_failed: u64,
183}
184
185/// Ordered transaction event.
186#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
187pub struct TransactionEvent {
188    /// Transaction id.
189    pub transaction_id: TransactionId,
190    /// Channel id.
191    pub channel_id: ChannelId,
192    /// Session id (established by this point for ordinary events).
193    pub session_id: SessionId,
194    /// Contiguous sequence starting at 1, including `Ended`.
195    pub sequence: u64,
196    /// Payload.
197    pub payload: TransactionEventPayload,
198}
199
200/// Event payload variants.
201///
202/// Live assistant text arrives only as [`Self::CanonicalUnit`] (complete units).
203/// There is **no** token / delta stream on this port.
204#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
205pub enum TransactionEventPayload {
206    /// External session identity established.
207    SessionEstablished {
208        /// Authoritative external id.
209        external_session_id: crate::id::ExternalSessionId,
210    },
211    /// Complete canonical unit from Interpreter composition (not a token delta).
212    CanonicalUnit(CanonicalUnitEvent),
213    /// Host tool lifecycle.
214    ToolLifecycle(ToolLifecycleEvent),
215    /// Safe diagnostic.
216    Diagnostic(TransactionDiagnostic),
217    /// Terminal event (exactly once) — legacy v1 shape with embedded delivery.
218    Ended(TransactionEnd),
219    /// Terminal event body without self-referential delivery (Runtime v2).
220    EndedEvent(TransactionEndEvent),
221}
222
223/// Bounded safe transaction diagnostic.
224#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
225pub struct TransactionDiagnostic {
226    /// Safe diagnostic.
227    pub diagnostic: SafeDiagnostic,
228}
229
230/// Terminal transaction result.
231///
232/// **Legacy (v1):** embeds `event_delivery` inside the terminal event itself.
233/// Runtime v2 publishes [`TransactionEndEvent`] on the event stream and reports
234/// delivery/cleanup on [`TransactionCompletion`] instead.
235#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
236pub struct TransactionEnd {
237    /// Transaction id.
238    pub transaction_id: TransactionId,
239    /// Session when established.
240    pub session_id: Option<SessionId>,
241    /// Channel.
242    pub channel_id: ChannelId,
243    /// Terminal kind.
244    pub kind: TransactionEndKind,
245    /// Prior cause when terminal selection raced (optional).
246    pub prior_terminal_cause: Option<TransactionEndKind>,
247    /// Whether the terminal event was accepted by the sink.
248    pub event_delivery: EventDeliveryOutcome,
249    /// Number of events emitted including `Ended`.
250    pub emitted_events: u64,
251    /// Bounded usage facts.
252    pub usage: TransactionUsage,
253    /// Safe diagnostics.
254    pub diagnostics: Vec<TransactionDiagnostic>,
255}
256
257/// Terminal event body for the v2 event stream (no self-referential delivery).
258#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
259pub struct TransactionEndEvent {
260    /// Transaction id.
261    pub transaction_id: TransactionId,
262    /// Session when established.
263    pub session_id: Option<SessionId>,
264    /// Channel.
265    pub channel_id: ChannelId,
266    /// Terminal kind.
267    pub kind: TransactionEndKind,
268    /// Number of events emitted including this terminal event.
269    pub emitted_events: u64,
270    /// Bounded usage facts.
271    pub usage: TransactionUsage,
272    /// Safe diagnostics.
273    pub diagnostics: Vec<TransactionDiagnostic>,
274}
275
276/// Outcome of attempting to enqueue the terminal `Ended` event (v2).
277#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
278pub enum TerminalEventDelivery {
279    /// Terminal event was accepted by the event mailbox.
280    Published,
281    /// Event receiver was dropped / channel closed.
282    QueueClosed,
283    /// Terminal-event budget elapsed before enqueue.
284    DeadlineExceeded,
285    /// Item or byte capacity rejected the terminal event.
286    LimitExceeded,
287    /// No publisher / Seal was ever attempted (e.g. shutdown before Start).
288    ///
289    /// Spec §6.4 / D-041: never-attempted is **not** [`Self::Published`].
290    NotAttempted,
291}
292
293/// Status of owned cleanup after completion publication (v2).
294#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
295pub enum CleanupStatus {
296    /// All owned tasks/processes have been observed finished.
297    Complete,
298    /// Completion was published while owned work remains.
299    Pending {
300        /// Owned Tokio tasks still registered.
301        owned_tasks: u32,
302        /// Owned child processes still registered.
303        owned_processes: u32,
304        /// Cooperative in-process tools still outstanding.
305        cooperative_tools: u32,
306    },
307    /// Cleanup failed with a closed code.
308    Failed {
309        /// Stable cleanup failure code.
310        code: CleanupFailureCode,
311    },
312}
313
314/// Closed cleanup failure codes (v2).
315#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
316pub enum CleanupFailureCode {
317    /// Join observed a panic.
318    TaskPanicked,
319    /// Process reap failed.
320    ProcessReapFailed,
321    /// Internal ownership invariant broken.
322    InvariantFailed,
323}
324
325/// One-shot completion mailbox payload (v2).
326///
327/// Separates terminal event data from terminal-event delivery and cleanup.
328#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
329pub struct TransactionCompletion {
330    /// Terminal event body (also published on the event stream when possible).
331    pub end: TransactionEndEvent,
332    /// Result of the terminal event enqueue attempt, or [`TerminalEventDelivery::NotAttempted`]
333    /// when Seal / `Ended` was never issued.
334    pub terminal_event_delivery: TerminalEventDelivery,
335    /// Whether owned cleanup is complete.
336    pub cleanup: CleanupStatus,
337}
338
339/// Wait outcome for [`crate`] runtime owner shutdown (v2).
340#[derive(Clone, Debug, PartialEq, Eq)]
341pub enum ShutdownWaitOutcome {
342    /// Stopped invariants hold; shutdown generation is complete.
343    Stopped(ShutdownReport),
344    /// Wait deadline elapsed; runtime remains `Quiescing` and retains ownership.
345    TimedOut(ShutdownSnapshot),
346}
347
348/// Final shutdown report when the runtime reaches `Stopped` (v2).
349#[derive(Clone, Debug, PartialEq, Eq, Default)]
350pub struct ShutdownReport {
351    /// Admitted transactions that received a completion publication attempt.
352    pub completions_published: u64,
353    /// Completions where the host had dropped its receiver.
354    pub completions_receiver_dropped: u64,
355    /// Completions that hit an invariant on the sender.
356    pub completions_invariant_failed: u64,
357    /// Transactions terminated because of runtime shutdown.
358    pub runtime_shutdown_terminals: u64,
359}
360
361/// Point-in-time shutdown progress while still `Quiescing` (v2).
362#[derive(Clone, Debug, PartialEq, Eq, Default)]
363pub struct ShutdownSnapshot {
364    /// Shutdown generation id (shared by concurrent waiters).
365    pub generation: u64,
366    /// Ledger entries still present.
367    pub ledger_entries: u32,
368    /// Owned Tokio tasks still registered.
369    pub owned_tasks: u32,
370    /// Owned child processes still registered.
371    pub owned_processes: u32,
372    /// Outstanding MCP routes.
373    pub mcp_routes: u32,
374    /// Completion publications attempted so far in this generation.
375    pub completions_published: u64,
376}
377
378/// Closed terminal kinds.
379#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
380pub enum TransactionEndKind {
381    /// Successful completion.
382    Completed,
383    /// Caller must continue (caller-controlled policy).
384    ContinuationRequired,
385    /// Cancelled.
386    Cancelled,
387    /// Force-terminated.
388    Terminated,
389    /// Runtime shutdown.
390    RuntimeShutdown,
391    /// Deadline exceeded.
392    DeadlineExceeded,
393    /// Channel open/attach failed.
394    ChannelOpenFailed,
395    /// Outbound encoding failed.
396    EncodingFailed,
397    /// Connector failed.
398    ConnectorFailed,
399    /// Interpretation failed.
400    InterpretationFailed,
401    /// Tool exchange failed.
402    ToolExchangeFailed,
403    /// Event delivery failed.
404    EventDeliveryFailed,
405    /// Resource limit exceeded.
406    LimitExceeded,
407    /// Internal invariant failed.
408    InvariantFailed,
409}
410
411/// Terminal event delivery outcome.
412#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
413pub enum EventDeliveryOutcome {
414    /// Sink accepted.
415    Accepted,
416    /// Sink failed or timed out.
417    Failed,
418}
419
420/// Bounded usage facts (unavailable is not zero).
421#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Default)]
422pub struct TransactionUsage {
423    /// Provider input tokens when known.
424    pub provider_input_tokens: Option<u64>,
425    /// Provider output tokens when known.
426    pub provider_output_tokens: Option<u64>,
427    /// Number of provider exchanges.
428    pub provider_exchanges: u32,
429    /// Number of tool executions started.
430    pub tools_started: u32,
431    /// Number of tool executions completed (success or domain failure).
432    pub tools_completed: u32,
433}
434
435/// Event delivery error (safe).
436#[derive(Clone, Debug, Error, PartialEq, Eq)]
437pub enum EventDeliveryError {
438    /// Sink rejected or failed.
439    #[error("event delivery failed")]
440    Failed,
441    /// Delivery deadline exceeded.
442    #[error("event delivery deadline exceeded")]
443    DeadlineExceeded,
444}
445
446/// Completion callback delivery error (safe).
447#[derive(Clone, Debug, Error, PartialEq, Eq)]
448pub enum CompletionDeliveryError {
449    /// Callback failed.
450    #[error("completion callback failed")]
451    Failed,
452    /// Callback deadline exceeded.
453    #[error("completion callback deadline exceeded")]
454    DeadlineExceeded,
455}
456
457/// Synchronous admission error.
458#[derive(Clone, Debug, Error, PartialEq, Eq)]
459#[error("{kind:?}: {message}")]
460pub struct AdmissionError {
461    /// Closed kind.
462    pub kind: AdmissionErrorKind,
463    /// Safe bounded message.
464    pub message: String,
465}
466
467impl AdmissionError {
468    /// Construct an admission error.
469    pub fn new(kind: AdmissionErrorKind, message: impl Into<String>) -> Self {
470        Self {
471            kind,
472            message: message.into(),
473        }
474    }
475}
476
477/// Admission error kinds.
478#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
479pub enum AdmissionErrorKind {
480    /// Runtime not accepting.
481    RuntimeShuttingDown,
482    /// Unknown Channel id.
483    UnknownChannel,
484    /// Session already has an active transaction.
485    SessionAlreadyActive,
486    /// Unknown tool id.
487    UnknownTool,
488    /// Duplicate tool id in request.
489    DuplicateTool,
490    /// Invalid canonical input.
491    InvalidInput,
492    /// Invalid configuration merge.
493    InvalidConfiguration,
494    /// Capability mismatch for Channel/tools/session.
495    CapabilityMismatch,
496    /// Capacity exceeded.
497    CapacityExceeded,
498    /// Actor spawn failed.
499    SpawnFailed,
500}
501
502#[cfg(test)]
503mod tests {
504    use super::*;
505    use crate::input::user_text_input;
506    use std::sync::Arc;
507
508    #[test]
509    fn end_kind_round_trip() {
510        let kind = TransactionEndKind::Completed;
511        let json = serde_json::to_string(&kind).unwrap();
512        let back: TransactionEndKind = serde_json::from_str(&json).unwrap();
513        assert_eq!(kind, back);
514    }
515
516    #[tokio::test]
517    async fn sink_adapters_return_futures() {
518        let sink = FnEventSink(|_e| Box::pin(async { Ok(()) }) as EventDelivery);
519        let events: Arc<dyn TransactionEventSink> = Arc::new(sink);
520        let end = TransactionEnd {
521            transaction_id: TransactionId::generate(),
522            session_id: None,
523            channel_id: ChannelId::try_new("ch").unwrap(),
524            kind: TransactionEndKind::Completed,
525            prior_terminal_cause: None,
526            event_delivery: EventDeliveryOutcome::Accepted,
527            emitted_events: 1,
528            usage: TransactionUsage::default(),
529            diagnostics: vec![],
530        };
531        let ev = TransactionEvent {
532            transaction_id: end.transaction_id,
533            channel_id: end.channel_id.clone(),
534            session_id: SessionId::try_new("s").unwrap(),
535            sequence: 1,
536            payload: TransactionEventPayload::Ended(end.clone()),
537        };
538        events.deliver(ev).await.unwrap();
539
540        let cb: Box<dyn CompletionCallback> = Box::new(FnCompletionCallback(|_e| {
541            Box::pin(async { Ok(()) }) as CompletionDelivery
542        }));
543        cb.call(end).await.unwrap();
544
545        let _input = user_text_input("hello").unwrap();
546    }
547}
monoloop_contracts/transaction.rs

monoloop_contracts/
transaction.rs
