talos_core/
submission.rs

1//! Structured transactional Session submissions (ADR-056 / TUI-044).
2
3use serde::{Deserialize, Serialize};
4
5use crate::message::ContentPart;
6
7/// Maximum UTF-8 bytes accepted for one structured submission item.
8pub const MAX_SUBMISSION_ITEM_BYTES: usize = 64 * 1024;
9/// Maximum items retained in one interactive steering queue.
10pub const MAX_STEERING_QUEUE_ITEMS: usize = 128;
11/// Maximum UTF-8 text bytes retained in one interactive steering queue.
12pub const MAX_STEERING_QUEUE_BYTES: usize = 1024 * 1024;
13/// Maximum image attachments owned across running and pending work.
14pub const MAX_STEERING_QUEUE_IMAGES: usize = 16;
15/// Maximum declared image bytes owned across running and pending work.
16pub const MAX_STEERING_QUEUE_IMAGE_BYTES: u64 = 100 * 1024 * 1024;
17/// Maximum compatible items projected into one Actor Turn.
18pub const MAX_SUBMISSION_BATCH_ITEMS: usize = 32;
19/// Maximum UTF-8 text bytes projected into one Actor Turn.
20pub const MAX_SUBMISSION_BATCH_BYTES: usize = 256 * 1024;
21/// Maximum metadata bytes retained for one attachment.
22pub const MAX_SUBMISSION_ATTACHMENT_METADATA_BYTES: usize = 16 * 1024;
23/// Maximum metadata bytes retained across one structured submission.
24pub const MAX_SUBMISSION_TOTAL_ATTACHMENT_METADATA_BYTES: usize = 64 * 1024;
25/// Maximum image attachments accepted in one structured submission.
26pub const MAX_SUBMISSION_IMAGE_COUNT: usize = 4;
27/// Maximum declared bytes for one image attachment.
28pub const MAX_SUBMISSION_IMAGE_BYTES: u64 = 20 * 1024 * 1024;
29/// Maximum declared image bytes across one structured submission.
30pub const MAX_SUBMISSION_TOTAL_IMAGE_BYTES: u64 = 50 * 1024 * 1024;
31/// Maximum durable pending submissions retained for one Session.
32pub const MAX_PENDING_SUBMISSIONS: usize = 128;
33/// Maximum durable pending text bytes retained for one Session.
34pub const MAX_PENDING_SUBMISSION_BYTES: usize = 1024 * 1024;
35
36/// Origin of a structured Session submission.
37#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
38#[serde(rename_all = "snake_case")]
39pub enum SubmissionSource {
40    /// Interactive user input accepted by a product bridge.
41    User,
42    /// A scheduled follow-up produced by the Session scheduler.
43    Scheduler,
44    /// A legacy or external caller using a compatibility operation.
45    Compatibility,
46}
47
48/// Dispatch semantics fixed before an item enters the authoritative queue.
49#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
50#[serde(rename_all = "snake_case")]
51pub enum SubmissionKind {
52    /// A normal model-visible user Turn.
53    UserTurn,
54    /// A request-preview diagnostic that must not call the Provider.
55    PreviewRequest,
56}
57
58/// One recoverable input item inside a structured submission.
59#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
60pub struct SubmissionItem {
61    /// Opaque producer-assigned item identity.
62    pub id: String,
63    /// Monotonic producer-side FIFO order.
64    pub enqueue_sequence: u64,
65    /// Dispatch kind fixed before queue admission.
66    pub kind: SubmissionKind,
67    /// Original text without delimiter rewriting.
68    pub text: String,
69    /// Attachments bound to this exact item before queue admission.
70    #[serde(default, skip_serializing_if = "Vec::is_empty")]
71    pub attachments: Vec<ContentPart>,
72}
73
74impl SubmissionItem {
75    /// Returns the original UTF-8 text size.
76    #[must_use]
77    pub fn text_bytes(&self) -> usize {
78        self.text.len()
79    }
80}
81
82/// An immutable compatible queue prefix prepared for one Actor Turn.
83#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
84pub struct StructuredSubmission {
85    /// Stable identity shared by preparation, retries, receipt, and reconciliation.
86    pub id: String,
87    /// Source used by Actor arbitration.
88    pub source: SubmissionSource,
89    /// Runtime generation of the addressed logical Session.
90    #[serde(default)]
91    pub sender_generation: u64,
92    /// Ordered, homogeneous, individually recoverable items.
93    pub items: Vec<SubmissionItem>,
94}
95
96impl StructuredSubmission {
97    /// Returns aggregate UTF-8 text bytes using saturating accounting.
98    #[must_use]
99    pub fn total_text_bytes(&self) -> usize {
100        self.items.iter().fold(0usize, |total, item| {
101            total.saturating_add(item.text_bytes())
102        })
103    }
104
105    /// Returns attachment count and declared bytes using saturating accounting.
106    #[must_use]
107    pub fn image_totals(&self) -> (usize, u64) {
108        self.items.iter().flat_map(|item| &item.attachments).fold(
109            (0usize, 0u64),
110            |(count, bytes), part| match part {
111                ContentPart::Image { byte_count, .. } => {
112                    (count.saturating_add(1), bytes.saturating_add(*byte_count))
113                }
114                ContentPart::Text { .. } => (count, bytes),
115            },
116        )
117    }
118
119    /// Returns total attachment metadata bytes or `None` on arithmetic overflow.
120    #[must_use]
121    pub fn attachment_metadata_bytes(&self) -> Option<usize> {
122        self.items
123            .iter()
124            .flat_map(|item| &item.attachments)
125            .try_fold(0usize, |total, part| match part {
126                ContentPart::Image { path, mime, .. } => total
127                    .checked_add(path.as_os_str().to_string_lossy().len())?
128                    .checked_add(mime.len())?
129                    .checked_add(std::mem::size_of::<u64>())?
130                    .checked_add(32),
131                ContentPart::Text { .. } => None,
132            })
133    }
134
135    /// Returns the common dispatch kind for a non-empty homogeneous batch.
136    #[must_use]
137    pub fn common_kind(&self) -> Option<SubmissionKind> {
138        let first = self.items.first()?.kind;
139        self.items
140            .iter()
141            .all(|item| item.kind == first)
142            .then_some(first)
143    }
144
145    /// Validates immutable identity, ordering, compatibility, and hard bounds.
146    pub fn validate(&self) -> Result<(), SubmissionRejectionReason> {
147        if self.id.is_empty()
148            || self.items.is_empty()
149            || self.items.len() > MAX_SUBMISSION_BATCH_ITEMS
150            || self.common_kind().is_none()
151        {
152            return Err(SubmissionRejectionReason::InvalidStructure);
153        }
154
155        let mut previous_sequence = None;
156        let mut item_ids = std::collections::HashSet::with_capacity(self.items.len());
157        for item in &self.items {
158            if item.id.is_empty() {
159                return Err(SubmissionRejectionReason::InvalidStructure);
160            }
161            if !item_ids.insert(item.id.as_str()) {
162                return Err(SubmissionRejectionReason::Duplicate);
163            }
164            if item.text_bytes() > MAX_SUBMISSION_ITEM_BYTES {
165                return Err(SubmissionRejectionReason::LimitExceeded);
166            }
167            if previous_sequence.is_some_and(|previous| item.enqueue_sequence <= previous) {
168                return Err(SubmissionRejectionReason::InvalidStructure);
169            }
170            previous_sequence = Some(item.enqueue_sequence);
171        }
172
173        if self.total_text_bytes() > MAX_SUBMISSION_BATCH_BYTES {
174            return Err(SubmissionRejectionReason::LimitExceeded);
175        }
176        if self.common_kind() == Some(SubmissionKind::PreviewRequest)
177            && (self.items.len() != 1 || !self.items[0].attachments.is_empty())
178        {
179            return Err(SubmissionRejectionReason::InvalidStructure);
180        }
181
182        let (image_count, image_bytes) = self.image_totals();
183        let metadata_bytes = self
184            .attachment_metadata_bytes()
185            .ok_or(SubmissionRejectionReason::InvalidStructure)?;
186        let oversized_attachment =
187            self.items
188                .iter()
189                .flat_map(|item| &item.attachments)
190                .any(|part| match part {
191                    ContentPart::Image {
192                        path,
193                        mime,
194                        byte_count,
195                        ..
196                    } => {
197                        let metadata = path
198                            .as_os_str()
199                            .to_string_lossy()
200                            .len()
201                            .saturating_add(mime.len())
202                            .saturating_add(std::mem::size_of::<u64>())
203                            .saturating_add(32);
204                        *byte_count > MAX_SUBMISSION_IMAGE_BYTES
205                            || metadata > MAX_SUBMISSION_ATTACHMENT_METADATA_BYTES
206                    }
207                    ContentPart::Text { .. } => true,
208                });
209        if image_count > MAX_SUBMISSION_IMAGE_COUNT
210            || image_bytes > MAX_SUBMISSION_TOTAL_IMAGE_BYTES
211            || metadata_bytes > MAX_SUBMISSION_TOTAL_ATTACHMENT_METADATA_BYTES
212            || oversized_attachment
213        {
214            return Err(SubmissionRejectionReason::LimitExceeded);
215        }
216
217        Ok(())
218    }
219}
220
221/// Durable state of an Actor-owned pending submission.
222#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
223#[serde(rename_all = "snake_case")]
224pub enum PendingSubmissionState {
225    /// Accepted durably but not started.
226    AcceptedPending,
227    /// Correlated to an active model Turn.
228    Running,
229    /// Retained but automatic advancement is paused.
230    PausedPending,
231    /// The started Turn ended by explicit cancellation.
232    TerminalCancelled,
233    /// The started Turn ended with an error.
234    TerminalError,
235    /// Successful transcript commit completed.
236    Committed,
237}
238
239/// Content-free reason a structured submission was rejected.
240#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
241#[serde(rename_all = "snake_case")]
242pub enum SubmissionRejectionReason {
243    /// Durable pending storage is unavailable for this runtime.
244    DurabilityUnavailable,
245    /// The addressed Session is not current.
246    WrongSession,
247    /// The addressed Session generation is stale or unknown.
248    WrongGeneration,
249    /// The identity already exists with different immutable content.
250    IdentityConflict,
251    /// The exact identity was accepted already.
252    Duplicate,
253    /// The pending submission was explicitly cancelled before start.
254    Cancelled,
255    /// The submission is empty or mixes incompatible kinds.
256    InvalidStructure,
257    /// An item, batch, attachment, queue, or journal bound was exceeded.
258    LimitExceeded,
259    /// The complete Provider request would exceed its context budget.
260    ContextBudgetExceeded,
261    /// The Actor is shutting down or no longer accepts work.
262    SessionClosed,
263}
264
265/// Result of durable Actor acceptance or reconciliation.
266#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
267#[serde(tag = "status", rename_all = "snake_case")]
268pub enum SubmissionReceiptDisposition {
269    /// The journal accepted this exact submission for the first time.
270    AcceptedPending,
271    /// The same immutable submission was accepted previously.
272    AlreadyAccepted {
273        /// Current durable state.
274        state: PendingSubmissionState,
275        /// Turn identity when the submission has started.
276        #[serde(default, skip_serializing_if = "Option::is_none")]
277        turn_id: Option<String>,
278    },
279    /// The authoritative generation confirms no matching submission exists.
280    NotAccepted,
281    /// Admission failed before ownership transferred.
282    Rejected {
283        /// Bounded, content-free reason.
284        reason: SubmissionRejectionReason,
285    },
286}
287
288impl SubmissionReceiptDisposition {
289    /// Returns true once the Actor has durably accepted this exact identity.
290    #[must_use]
291    pub fn has_durable_custody(&self) -> bool {
292        matches!(self, Self::AcceptedPending | Self::AlreadyAccepted { .. })
293    }
294}
295
296/// Canonical durable result delivered to a tracked structured submitter.
297#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
298pub struct SubmissionReceipt {
299    /// Logical Session that produced the receipt.
300    pub session_id: String,
301    /// Runtime generation echoed from the submitted batch.
302    pub session_generation: u64,
303    /// Stable batch identity.
304    pub submission_id: String,
305    /// Exact frozen-prefix identity. Currently identical to `submission_id`.
306    pub reservation_id: String,
307    /// Durable receipt identity, empty only for rejection or `NotAccepted`.
308    pub receipt_id: String,
309    /// Source used by Actor arbitration.
310    pub source: SubmissionSource,
311    /// Number of original recoverable items.
312    pub item_count: usize,
313    /// Aggregate original UTF-8 text bytes.
314    pub total_text_bytes: usize,
315    /// Durable, content-free disposition.
316    pub disposition: SubmissionReceiptDisposition,
317}
318
319#[cfg(test)]
320mod tests {
321    use super::*;
322
323    fn submission() -> StructuredSubmission {
324        StructuredSubmission {
325            id: "batch-1".into(),
326            source: SubmissionSource::User,
327            sender_generation: 1,
328            items: vec![SubmissionItem {
329                id: "item-1".into(),
330                enqueue_sequence: 1,
331                kind: SubmissionKind::UserTurn,
332                text: "hello".into(),
333                attachments: Vec::new(),
334            }],
335        }
336    }
337
338    #[test]
339    fn valid_submission_preserves_item_boundaries() {
340        let mut value = submission();
341        value.items.push(SubmissionItem {
342            id: "item-2".into(),
343            enqueue_sequence: 2,
344            kind: SubmissionKind::UserTurn,
345            text: "world".into(),
346            attachments: Vec::new(),
347        });
348        assert_eq!(value.validate(), Ok(()));
349        let encoded = serde_json::to_string(&value).expect("operation should succeed");
350        let decoded: StructuredSubmission =
351            serde_json::from_str(&encoded).expect("operation should succeed");
352        assert_eq!(decoded.items.len(), 2);
353        assert_eq!(decoded.items[0].text, "hello");
354        assert_eq!(decoded.items[1].text, "world");
355    }
356
357    #[test]
358    fn duplicate_item_identity_fails_closed() {
359        let mut value = submission();
360        value.items.push(SubmissionItem {
361            id: "item-1".into(),
362            enqueue_sequence: 2,
363            kind: SubmissionKind::UserTurn,
364            text: "again".into(),
365            attachments: Vec::new(),
366        });
367        assert_eq!(value.validate(), Err(SubmissionRejectionReason::Duplicate));
368    }
369
370    #[test]
371    fn incompatible_kinds_and_regressive_sequence_fail_closed() {
372        let mut value = submission();
373        value.items.push(SubmissionItem {
374            id: "item-2".into(),
375            enqueue_sequence: 1,
376            kind: SubmissionKind::PreviewRequest,
377            text: "preview".into(),
378            attachments: Vec::new(),
379        });
380        assert_eq!(
381            value.validate(),
382            Err(SubmissionRejectionReason::InvalidStructure)
383        );
384    }
385
386    #[test]
387    fn durable_custody_excludes_rejection_and_not_accepted() {
388        assert!(SubmissionReceiptDisposition::AcceptedPending.has_durable_custody());
389        assert!(
390            SubmissionReceiptDisposition::AlreadyAccepted {
391                state: PendingSubmissionState::Committed,
392                turn_id: Some("turn-1".into()),
393            }
394            .has_durable_custody()
395        );
396        assert!(!SubmissionReceiptDisposition::NotAccepted.has_durable_custody());
397        assert!(
398            !SubmissionReceiptDisposition::Rejected {
399                reason: SubmissionRejectionReason::LimitExceeded,
400            }
401            .has_durable_custody()
402        );
403    }
404}
talos_core/submission.rs

talos_core/
submission.rs
