talos_session/
pending_submission.rs

1//! Durable, session-scoped custody for structured submissions (ADR-056).
2
3use std::fs::OpenOptions;
4use std::path::{Path, PathBuf};
5use std::sync::{Arc, Mutex};
6
7use rusqlite::{Connection, OptionalExtension, params};
8use talos_core::submission::{
9    PendingSubmissionState, StructuredSubmission, SubmissionKind, SubmissionReceiptDisposition,
10    SubmissionRejectionReason,
11};
12use thiserror::Error;
13use uuid::Uuid;
14
15use crate::runtime_state::{
16    SessionRuntimeActivation, SessionRuntimeActivationStatus, SessionRuntimeIdentity,
17    SessionRuntimeState,
18};
19use crate::turn_outcome::decode_turn_transcript_outcome;
20use crate::{CompactTextSessionStore, JsonlSessionStore, SessionStore, TurnTranscriptOutcome};
21
22const SCHEMA_VERSION: i64 = 1;
23const RUNTIME_GENERATION_KEY: &str = "runtime_generation";
24mod codec;
25mod storage;
26
27use codec::*;
28use storage::*;
29
30/// Failure while reading or mutating pending-submission custody.
31#[derive(Debug, Error)]
32pub enum PendingSubmissionError {
33    /// SQLite operation failed.
34    #[error("pending submission journal failed: {0}")]
35    Storage(#[from] rusqlite::Error),
36    /// Structured payload encoding failed.
37    #[error("pending submission payload failed: {0}")]
38    Payload(#[from] serde_json::Error),
39    /// Journal directory creation failed.
40    #[error("pending submission journal I/O failed: {0}")]
41    Io(#[from] std::io::Error),
42    /// Transcript inspection failed while reconciling a Running record.
43    #[error("pending submission transcript inspection failed: {0}")]
44    Transcript(#[from] crate::SessionError),
45    /// The in-process journal lock was poisoned.
46    #[error("pending submission journal lock poisoned")]
47    LockPoisoned,
48    /// The on-disk schema cannot be read by this build.
49    #[error("unsupported pending submission journal schema version: {0}")]
50    UnsupportedSchema(i64),
51    /// A lifecycle update addressed no durable record.
52    #[error("pending submission record not found")]
53    MissingRecord,
54    /// A lifecycle update violates the state machine.
55    #[error("invalid pending submission state transition")]
56    InvalidTransition,
57    /// The durable runtime generation changed outside the expected lifecycle owner.
58    #[error("runtime generation conflict: expected {expected}, found {actual}")]
59    GenerationConflict {
60        /// Generation held by the caller.
61        expected: u64,
62        /// Generation stored durably for the Session.
63        actual: u64,
64    },
65    /// Non-terminal durable custody still belongs to the current generation.
66    #[error("runtime generation {generation} still owns {pending} non-terminal submission(s)")]
67    GenerationBusy {
68        /// Generation that still owns custody.
69        generation: u64,
70        /// Number of non-terminal durable records.
71        pending: usize,
72    },
73    /// The durable runtime generation cannot advance further.
74    #[error("runtime generation exhausted")]
75    GenerationExhausted,
76    /// A runtime activation record failed validation or contradicted the durable generation.
77    #[error("invalid Session runtime activation: {0}")]
78    InvalidRuntimeActivation(String),
79    /// A different runtime identity already owns this generation.
80    #[error("Session runtime activation conflict: existing {existing}, requested {requested}")]
81    RuntimeActivationConflict { existing: String, requested: String },
82}
83
84/// Exact durable record returned for Actor recovery.
85#[derive(Debug, Clone, PartialEq)]
86pub struct PendingSubmissionRecord {
87    /// Durable acceptance receipt.
88    pub receipt_id: String,
89    /// SHA-256 fingerprint of the immutable serialized submission.
90    pub payload_fingerprint: String,
91    /// Original immutable structured submission.
92    pub submission: StructuredSubmission,
93    /// Current durable lifecycle state.
94    pub state: PendingSubmissionState,
95    /// Correlated Turn after execution starts.
96    pub turn_id: Option<String>,
97}
98
99/// SQLite sidecar that stores accepted but uncompleted Session work.
100#[derive(Debug, Clone)]
101pub struct PendingSubmissionStore {
102    path: Arc<PathBuf>,
103    session_file: Arc<PathBuf>,
104    session_id: Arc<String>,
105    lock: Arc<Mutex<()>>,
106}
107
108impl PendingSubmissionStore {
109    /// Opens the sidecar adjacent to a Session transcript.
110    #[must_use]
111    pub fn for_session(session: &crate::Session) -> Self {
112        Self::for_session_file(&session.file_path, &session.id.to_string())
113    }
114
115    /// Opens the sidecar adjacent to an arbitrary Session transcript path.
116    #[must_use]
117    pub fn for_session_file(session_file: &Path, session_id: &str) -> Self {
118        let parent = session_file.parent().unwrap_or_else(|| Path::new("."));
119        Self {
120            path: Arc::new(parent.join(format!("{session_id}.pending.sqlite"))),
121            session_file: Arc::new(session_file.to_path_buf()),
122            session_id: Arc::new(session_id.to_owned()),
123            lock: Arc::new(Mutex::new(())),
124        }
125    }
126
127    /// Returns the sidecar path.
128    #[must_use]
129    pub fn path(&self) -> &Path {
130        self.path.as_ref()
131    }
132
133    /// Loads the durable runtime generation for this logical Session.
134    ///
135    /// Legacy Sessions without the metadata key are initialized at generation
136    /// zero. Process reconstruction rehydrates this exact value so accepted
137    /// envelopes remain addressable after memory loss.
138    pub fn runtime_generation(&self) -> Result<u64, PendingSubmissionError> {
139        let _guard = self.guard()?;
140        let mut connection = self.connection()?;
141        let transaction = immediate(&mut connection)?;
142        ensure_schema(&transaction)?;
143        let generation = load_or_initialize_runtime_generation(&transaction)?;
144        transaction.commit()?;
145        Ok(generation)
146    }
147
148    /// Loads the compaction-independent current Session runtime identity.
149    pub fn runtime_state(&self) -> Result<Option<SessionRuntimeState>, PendingSubmissionError> {
150        let _guard = self.guard()?;
151        let connection = self.connection()?;
152        ensure_schema(&connection)?;
153        load_runtime_state(&connection)
154    }
155
156    /// Returns whether this Session still owns accepted, paused, or running work.
157    pub fn has_nonterminal_submissions(&self) -> Result<bool, PendingSubmissionError> {
158        if !self.path.exists() {
159            return Ok(false);
160        }
161        let _guard = self.guard()?;
162        let connection = self.connection()?;
163        ensure_schema(&connection)?;
164        Ok(count_nonterminal(&connection)? > 0)
165    }
166
167    /// Initializes a new/legacy Session with an exact committed runtime identity.
168    ///
169    /// Repeating the same initialization is idempotent. A contradictory identity
170    /// at the same durable generation is rejected rather than silently changing
171    /// the Session's reconstruction authority.
172    pub fn initialize_runtime_identity(
173        &self,
174        identity: SessionRuntimeIdentity,
175    ) -> Result<SessionRuntimeState, PendingSubmissionError> {
176        self.ensure_transcript_owner_marker()?;
177        let _guard = self.guard()?;
178        let mut connection = self.connection()?;
179        let transaction = immediate(&mut connection)?;
180        ensure_schema(&transaction)?;
181        let generation = load_or_initialize_runtime_generation(&transaction)?;
182        let activation = SessionRuntimeActivation::new(generation, identity.clone(), identity);
183        if let Some(existing) = load_runtime_state(&transaction)? {
184            if existing.activation == activation
185                && existing.status == SessionRuntimeActivationStatus::Committed
186            {
187                transaction.commit()?;
188                return Ok(existing);
189            }
190            return Err(PendingSubmissionError::RuntimeActivationConflict {
191                existing: existing.activation.activation_id,
192                requested: activation.activation_id,
193            });
194        }
195        write_runtime_state(
196            &transaction,
197            &SessionRuntimeState {
198                activation: activation.clone(),
199                status: SessionRuntimeActivationStatus::Committed,
200            },
201        )?;
202        transaction.commit()?;
203        Ok(SessionRuntimeState {
204            activation,
205            status: SessionRuntimeActivationStatus::Committed,
206        })
207    }
208
209    /// Atomically fences generation `G` and stages the exact `G+1` activation.
210    ///
211    /// The transcript marker is committed in a separate append-only store, so
212    /// the sidecar records `pending_marker` until the caller proves that marker
213    /// durable and calls [`Self::commit_runtime_activation`]. A crash at that
214    /// cut point is therefore recoverable without starting a runtime from an
215    /// unproven activation.
216    pub fn stage_runtime_activation(
217        &self,
218        expected: u64,
219        activation: &SessionRuntimeActivation,
220    ) -> Result<u64, PendingSubmissionError> {
221        if !activation.is_valid() {
222            return Err(PendingSubmissionError::InvalidRuntimeActivation(
223                "activation digest/version mismatch".to_string(),
224            ));
225        }
226        let expected_next = expected
227            .checked_add(1)
228            .ok_or(PendingSubmissionError::GenerationExhausted)?;
229        if activation.generation != expected_next {
230            return Err(PendingSubmissionError::InvalidRuntimeActivation(format!(
231                "activation generation {} does not follow expected generation {expected}",
232                activation.generation
233            )));
234        }
235
236        let _guard = self.guard()?;
237        let mut connection = self.connection()?;
238        let transaction = immediate(&mut connection)?;
239        ensure_schema(&transaction)?;
240        let current = load_or_initialize_runtime_generation(&transaction)?;
241
242        if current == activation.generation {
243            let existing = load_runtime_state(&transaction)?.ok_or_else(|| {
244                PendingSubmissionError::InvalidRuntimeActivation(
245                    "generation advanced without runtime activation state".to_string(),
246                )
247            })?;
248            if existing.activation == *activation {
249                transaction.commit()?;
250                return Ok(current);
251            }
252            return Err(PendingSubmissionError::RuntimeActivationConflict {
253                existing: existing.activation.activation_id,
254                requested: activation.activation_id.clone(),
255            });
256        }
257        if current != expected {
258            return Err(PendingSubmissionError::GenerationConflict {
259                expected,
260                actual: current,
261            });
262        }
263        let pending = count_nonterminal(&transaction)?;
264        if pending > 0 {
265            return Err(PendingSubmissionError::GenerationBusy {
266                generation: current,
267                pending,
268            });
269        }
270        transaction.execute(
271            "UPDATE pending_journal_meta SET value = ?2 WHERE key = ?1",
272            params![RUNTIME_GENERATION_KEY, to_i64(activation.generation)],
273        )?;
274        write_runtime_state(
275            &transaction,
276            &SessionRuntimeState {
277                activation: activation.clone(),
278                status: SessionRuntimeActivationStatus::PendingMarker,
279            },
280        )?;
281        transaction.commit()?;
282        Ok(activation.generation)
283    }
284
285    /// Marks a staged activation committed after its exact transcript marker is durable.
286    pub fn commit_runtime_activation(
287        &self,
288        activation_id: &str,
289    ) -> Result<SessionRuntimeState, PendingSubmissionError> {
290        let _guard = self.guard()?;
291        let mut connection = self.connection()?;
292        let transaction = immediate(&mut connection)?;
293        ensure_schema(&transaction)?;
294        let mut state = load_runtime_state(&transaction)?.ok_or_else(|| {
295            PendingSubmissionError::InvalidRuntimeActivation(
296                "no staged runtime activation exists".to_string(),
297            )
298        })?;
299        if state.activation.activation_id != activation_id {
300            return Err(PendingSubmissionError::RuntimeActivationConflict {
301                existing: state.activation.activation_id,
302                requested: activation_id.to_string(),
303            });
304        }
305        let generation = load_or_initialize_runtime_generation(&transaction)?;
306        if generation != state.activation.generation {
307            return Err(PendingSubmissionError::InvalidRuntimeActivation(format!(
308                "runtime state generation {} differs from durable generation {generation}",
309                state.activation.generation
310            )));
311        }
312        state.status = SessionRuntimeActivationStatus::Committed;
313        write_runtime_state(&transaction, &state)?;
314        transaction.commit()?;
315        Ok(state)
316    }
317
318    /// Atomically advances the durable generation for a live replacement of
319    /// the same logical Session.
320    ///
321    /// The quiescence check and metadata update execute under the same SQLite
322    /// immediate transaction used by [`Self::accept`]. Therefore either an
323    /// in-flight admission wins and the fence remains at the old generation,
324    /// or the fence wins and every new old-generation admission is rejected.
325    pub fn advance_runtime_generation(&self, expected: u64) -> Result<u64, PendingSubmissionError> {
326        let _guard = self.guard()?;
327        let mut connection = self.connection()?;
328        let transaction = immediate(&mut connection)?;
329        ensure_schema(&transaction)?;
330        let current = load_or_initialize_runtime_generation(&transaction)?;
331        if current != expected {
332            return Err(PendingSubmissionError::GenerationConflict {
333                expected,
334                actual: current,
335            });
336        }
337        let pending = count_nonterminal(&transaction)?;
338        if pending > 0 {
339            return Err(PendingSubmissionError::GenerationBusy {
340                generation: current,
341                pending,
342            });
343        }
344        let next = current
345            .checked_add(1)
346            .ok_or(PendingSubmissionError::GenerationExhausted)?;
347        transaction.execute(
348            "UPDATE pending_journal_meta SET value = ?2 WHERE key = ?1",
349            params![RUNTIME_GENERATION_KEY, to_i64(next)],
350        )?;
351        transaction.commit()?;
352        Ok(next)
353    }
354
355    /// Durably accepts an exact submission or returns its prior receipt.
356    pub fn accept(
357        &self,
358        submission: &StructuredSubmission,
359    ) -> Result<(String, SubmissionReceiptDisposition), PendingSubmissionError> {
360        if let Err(reason) = submission.validate() {
361            return Ok(rejected(reason));
362        }
363        let encoded = serde_json::to_string(submission)?;
364        let fingerprint = fingerprint(encoded.as_bytes());
365        let text_bytes = submission.total_text_bytes();
366        let (image_count, image_bytes) = submission.image_totals();
367        let generation = match i64::try_from(submission.sender_generation) {
368            Ok(value) => value,
369            Err(_) => return Ok(rejected(SubmissionRejectionReason::InvalidStructure)),
370        };
371
372        let _guard = self.guard()?;
373        let mut connection = self.connection()?;
374        let transaction = immediate(&mut connection)?;
375        ensure_schema(&transaction)?;
376
377        if let Some(row) = lookup(&transaction, &submission.id)? {
378            let disposition = if identity_matches(&row, &fingerprint, &encoded) {
379                SubmissionReceiptDisposition::AlreadyAccepted {
380                    state: decode_state(&row.state)?,
381                    turn_id: row.turn_id,
382                }
383            } else {
384                SubmissionReceiptDisposition::Rejected {
385                    reason: SubmissionRejectionReason::IdentityConflict,
386                }
387            };
388            transaction.commit()?;
389            return Ok((row.receipt_id, disposition));
390        }
391
392        let runtime_generation = load_or_initialize_runtime_generation(&transaction)?;
393        if submission.sender_generation != runtime_generation {
394            transaction.commit()?;
395            return Ok(rejected(SubmissionRejectionReason::WrongGeneration));
396        }
397
398        prune_tombstones(&transaction)?;
399        if exceeds_pending_bounds(&transaction, text_bytes, image_count, image_bytes)? {
400            transaction.commit()?;
401            return Ok(rejected(SubmissionRejectionReason::LimitExceeded));
402        }
403
404        let receipt_id = Uuid::new_v4().to_string();
405        let reservation_id = format!("reservation:{}", submission.id);
406        transaction.execute(
407            "INSERT INTO pending_submissions (
408                batch_id, reservation_id, session_id, session_generation,
409                receipt_id, fingerprint, submission_json, text_bytes,
410                image_count, image_bytes, state, turn_id
411             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10,
412                       'accepted_pending', NULL)",
413            params![
414                submission.id,
415                reservation_id,
416                self.session_id.as_str(),
417                generation,
418                receipt_id,
419                fingerprint,
420                encoded,
421                to_i64(text_bytes),
422                to_i64(image_count),
423                to_i64(image_bytes),
424            ],
425        )?;
426        transaction.commit()?;
427        Ok((receipt_id, SubmissionReceiptDisposition::AcceptedPending))
428    }
429
430    /// Reconciles a previously sent submission without creating custody.
431    pub fn reconcile(
432        &self,
433        submission: &StructuredSubmission,
434    ) -> Result<(String, SubmissionReceiptDisposition), PendingSubmissionError> {
435        if let Err(reason) = submission.validate() {
436            return Ok(rejected(reason));
437        }
438        let encoded = serde_json::to_string(submission)?;
439        let payload_fingerprint = fingerprint(encoded.as_bytes());
440        let _guard = self.guard()?;
441        let connection = self.connection()?;
442        ensure_schema(&connection)?;
443        Ok(match lookup(&connection, &submission.id)? {
444            Some(row) if identity_matches(&row, &payload_fingerprint, &encoded) => (
445                row.receipt_id,
446                SubmissionReceiptDisposition::AlreadyAccepted {
447                    state: decode_state(&row.state)?,
448                    turn_id: row.turn_id,
449                },
450            ),
451            Some(row) => (
452                row.receipt_id,
453                SubmissionReceiptDisposition::Rejected {
454                    reason: SubmissionRejectionReason::IdentityConflict,
455                },
456            ),
457            None => (String::new(), SubmissionReceiptDisposition::NotAccepted),
458        })
459    }
460
461    /// Marks an accepted submission as the active Turn.
462    pub fn mark_running(
463        &self,
464        submission_id: &str,
465        turn_id: &str,
466    ) -> Result<(), PendingSubmissionError> {
467        self.transition(
468            submission_id,
469            PendingSubmissionState::Running,
470            Some(turn_id),
471            &[
472                PendingSubmissionState::AcceptedPending,
473                PendingSubmissionState::PausedPending,
474            ],
475        )
476    }
477
478    /// Pauses one durably accepted submission before it starts.
479    pub fn mark_paused(&self, submission_id: &str) -> Result<(), PendingSubmissionError> {
480        self.transition(
481            submission_id,
482            PendingSubmissionState::PausedPending,
483            None,
484            &[
485                PendingSubmissionState::AcceptedPending,
486                PendingSubmissionState::PausedPending,
487            ],
488        )
489    }
490
491    /// Pauses all accepted submissions that have not started.
492    pub fn pause_unstarted(&self) -> Result<usize, PendingSubmissionError> {
493        let _guard = self.guard()?;
494        let mut connection = self.connection()?;
495        let transaction = immediate(&mut connection)?;
496        ensure_schema(&transaction)?;
497        let changed = transaction.execute(
498            "UPDATE pending_submissions SET state = 'paused_pending'
499             WHERE state = 'accepted_pending'",
500            [],
501        )?;
502        transaction.commit()?;
503        Ok(changed)
504    }
505
506    /// Explicitly terminalizes one accepted submission before Provider start.
507    ///
508    /// This is the recovery action for deterministic pre-start failures. The
509    /// original identity remains in the permanent idempotency ledger and can
510    /// never execute as fresh work later.
511    pub fn cancel_unstarted(&self, submission_id: &str) -> Result<(), PendingSubmissionError> {
512        self.transition(
513            submission_id,
514            PendingSubmissionState::TerminalCancelled,
515            None,
516            &[
517                PendingSubmissionState::AcceptedPending,
518                PendingSubmissionState::PausedPending,
519                PendingSubmissionState::TerminalCancelled,
520            ],
521        )
522    }
523
524    /// Marks a started submission terminal without making it resumable.
525    pub fn mark_terminal(
526        &self,
527        submission_id: &str,
528        state: PendingSubmissionState,
529        turn_id: &str,
530    ) -> Result<(), PendingSubmissionError> {
531        if !matches!(
532            state,
533            PendingSubmissionState::TerminalCancelled | PendingSubmissionState::TerminalError
534        ) {
535            return Err(PendingSubmissionError::InvalidTransition);
536        }
537        self.transition(
538            submission_id,
539            state,
540            Some(turn_id),
541            &[PendingSubmissionState::Running, state],
542        )
543    }
544
545    /// Finalizes custody from the authoritative terminal transcript outcome.
546    ///
547    /// Success becomes Committed. Error and Cancelled markers are mapped to
548    /// their matching terminal states even when the caller is the legacy
549    /// startup path named `mark_committed`. A real transcript file with no
550    /// marker is ambiguous and remains Running. Preview requests are the only
551    /// transcript-free successful structured kind and are validated from the
552    /// immutable journal payload before taking that compatibility path.
553    pub fn mark_committed(
554        &self,
555        submission_id: &str,
556        turn_id: &str,
557    ) -> Result<(), PendingSubmissionError> {
558        let preview = self.get(submission_id)?.is_some_and(|record| {
559            record.submission.common_kind() == Some(SubmissionKind::PreviewRequest)
560        });
561        if preview {
562            return self.transition(
563                submission_id,
564                PendingSubmissionState::Committed,
565                Some(turn_id),
566                &[
567                    PendingSubmissionState::Running,
568                    PendingSubmissionState::Committed,
569                ],
570            );
571        }
572
573        let outcome = self.transcript_outcome_for_turn(turn_id)?;
574        match outcome {
575            Some(TurnTranscriptOutcome::Success) => self.transition(
576                submission_id,
577                PendingSubmissionState::Committed,
578                Some(turn_id),
579                &[
580                    PendingSubmissionState::Running,
581                    PendingSubmissionState::Committed,
582                ],
583            ),
584            Some(TurnTranscriptOutcome::Cancelled) => self.mark_terminal(
585                submission_id,
586                PendingSubmissionState::TerminalCancelled,
587                turn_id,
588            ),
589            Some(TurnTranscriptOutcome::Error) => self.mark_terminal(
590                submission_id,
591                PendingSubmissionState::TerminalError,
592                turn_id,
593            ),
594            None if !self.session_file.exists() => self.transition(
595                submission_id,
596                PendingSubmissionState::Committed,
597                Some(turn_id),
598                &[
599                    PendingSubmissionState::Running,
600                    PendingSubmissionState::Committed,
601                ],
602            ),
603            None => Err(PendingSubmissionError::InvalidTransition),
604        }
605    }
606
607    /// Returns unstarted work in durable FIFO order.
608    pub fn recover_unstarted(
609        &self,
610    ) -> Result<Vec<PendingSubmissionRecord>, PendingSubmissionError> {
611        self.recover_states(&["accepted_pending", "paused_pending"])
612    }
613
614    /// Returns Running records for transcript-backed crash reconciliation.
615    pub fn recover_running(&self) -> Result<Vec<PendingSubmissionRecord>, PendingSubmissionError> {
616        self.recover_states(&["running"])
617    }
618
619    fn recover_states(
620        &self,
621        states: &[&str],
622    ) -> Result<Vec<PendingSubmissionRecord>, PendingSubmissionError> {
623        let _guard = self.guard()?;
624        let connection = self.connection()?;
625        ensure_schema(&connection)?;
626        let state_filter = states
627            .iter()
628            .map(|state| format!("'{state}'"))
629            .collect::<Vec<_>>()
630            .join(", ");
631        let sql = format!(
632            "SELECT receipt_id, fingerprint, submission_json, state, turn_id
633             FROM pending_submissions
634             WHERE state IN ({state_filter})
635             ORDER BY rowid ASC"
636        );
637        let mut statement = connection.prepare(&sql)?;
638        let rows = statement.query_map([], read_record_tuple)?;
639        rows.map(|row| tuple_to_record(row?)).collect()
640    }
641
642    /// Returns one durable submission record that still retains its payload.
643    pub fn get(
644        &self,
645        submission_id: &str,
646    ) -> Result<Option<PendingSubmissionRecord>, PendingSubmissionError> {
647        let _guard = self.guard()?;
648        let connection = self.connection()?;
649        ensure_schema(&connection)?;
650        connection
651            .query_row(
652                "SELECT receipt_id, fingerprint, submission_json, state, turn_id
653                 FROM pending_submissions WHERE batch_id = ?1",
654                params![submission_id],
655                read_record_tuple,
656            )
657            .optional()?
658            .map(tuple_to_record)
659            .transpose()
660    }
661
662    fn transcript_outcome_for_turn(
663        &self,
664        turn_id: &str,
665    ) -> Result<Option<TurnTranscriptOutcome>, PendingSubmissionError> {
666        if !self.session_file.exists() {
667            return Ok(None);
668        }
669        let entries = if self
670            .session_file
671            .extension()
672            .and_then(|value| value.to_str())
673            == Some("jsonl")
674        {
675            JsonlSessionStore.read_entries(self.session_file.as_ref())?
676        } else {
677            CompactTextSessionStore.read_entries(self.session_file.as_ref())?
678        };
679        Ok(entries
680            .into_iter()
681            .filter_map(|entry| decode_turn_transcript_outcome(&entry.content))
682            .filter(|record| record.turn_id == turn_id)
683            .map(|record| record.outcome)
684            .next_back())
685    }
686
687    fn transition(
688        &self,
689        submission_id: &str,
690        next: PendingSubmissionState,
691        turn_id: Option<&str>,
692        expected: &[PendingSubmissionState],
693    ) -> Result<(), PendingSubmissionError> {
694        let _guard = self.guard()?;
695        let mut connection = self.connection()?;
696        let transaction = immediate(&mut connection)?;
697        ensure_schema(&transaction)?;
698        let current = transaction
699            .query_row(
700                "SELECT state FROM pending_submissions WHERE batch_id = ?1",
701                params![submission_id],
702                |row| row.get::<_, String>(0),
703            )
704            .optional()?
705            .ok_or(PendingSubmissionError::MissingRecord)?;
706        if !expected.contains(&decode_state(&current)?) {
707            return Err(PendingSubmissionError::InvalidTransition);
708        }
709        transaction.execute(
710            "UPDATE pending_submissions SET state = ?2, turn_id = ?3 WHERE batch_id = ?1",
711            params![submission_id, encode_state(next), turn_id],
712        )?;
713        prune_tombstones(&transaction)?;
714        transaction.commit()?;
715        Ok(())
716    }
717
718    fn guard(&self) -> Result<std::sync::MutexGuard<'_, ()>, PendingSubmissionError> {
719        self.lock
720            .lock()
721            .map_err(|_| PendingSubmissionError::LockPoisoned)
722    }
723
724    fn ensure_transcript_owner_marker(&self) -> Result<(), PendingSubmissionError> {
725        if self.session_file.exists() {
726            return Ok(());
727        }
728        if let Some(parent) = self.session_file.parent() {
729            std::fs::create_dir_all(parent)?;
730        }
731        match OpenOptions::new()
732            .write(true)
733            .create_new(true)
734            .open(self.session_file.as_ref())
735        {
736            Ok(_) => Ok(()),
737            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => Ok(()),
738            Err(error) => Err(error.into()),
739        }
740    }
741
742    fn connection(&self) -> Result<Connection, PendingSubmissionError> {
743        if let Some(parent) = self.path.parent() {
744            std::fs::create_dir_all(parent)?;
745        }
746        let connection = Connection::open(self.path.as_ref())?;
747        connection.busy_timeout(SQLITE_BUSY_TIMEOUT)?;
748        retry_sqlite_busy(SQLITE_BUSY_TIMEOUT, || {
749            connection.execute_batch("PRAGMA journal_mode = WAL; PRAGMA synchronous = FULL;")
750        })?;
751        Ok(connection)
752    }
753}
754
755#[cfg(test)]
756mod tests;
talos_session/pending_submission.rs

talos_session/
pending_submission.rs
