mj_controller/database/
session_move.rs

1//! Move intent uses the same guarded writer as session lifecycle transitions.
2
3use super::*;
4use mj_core::state::MoveOperation;
5
6pub fn save_move_operation(operation: &MoveOperation) -> Result<()> {
7    let operation = operation.clone();
8    submit_database_write("save_move_operation", move |connection| {
9        save_move_operation_with(connection, &operation)
10    })
11}
12
13pub(super) fn save_move_operation_with(
14    connection: &Connection,
15    operation: &MoveOperation,
16) -> Result<()> {
17    connection.execute(
18        "INSERT INTO session_moves(session_id, operation_id, operation_json) VALUES (?1, ?2, ?3)
19         ON CONFLICT(session_id) DO UPDATE SET operation_id=excluded.operation_id,
20             operation_json=CASE WHEN session_moves.operation_id=excluded.operation_id
21                 AND json_extract(session_moves.operation_json, '$.cancellation_requested')=1
22                 THEN json_set(excluded.operation_json, '$.cancellation_requested', json('true'))
23                 ELSE excluded.operation_json END",
24        params![
25            operation.selection.session_id,
26            operation.operation_id,
27            serde_json::to_string(operation)?
28        ],
29    )?;
30    Ok(())
31}
32
33pub fn load_move_operation(session_id: &str) -> Result<Option<MoveOperation>> {
34    let connection = open_reader(&database_path())?;
35    load_move_operation_with(&connection, session_id)
36}
37
38pub(super) fn load_move_operation_with(
39    connection: &Connection,
40    session_id: &str,
41) -> Result<Option<MoveOperation>> {
42    let json: Option<String> = connection
43        .query_row(
44            "SELECT operation_json FROM session_moves WHERE session_id=?1",
45            [session_id],
46            |row| row.get(0),
47        )
48        .optional()?;
49    json.map(|json| serde_json::from_str(&json).context("decode durable move intent"))
50        .transpose()
51}
52
53pub fn load_move_operations() -> Result<Vec<MoveOperation>> {
54    let connection = open_reader(&database_path())?;
55    let mut statement =
56        connection.prepare("SELECT operation_json FROM session_moves ORDER BY session_id")?;
57    statement
58        .query_map([], |row| row.get::<_, String>(0))?
59        .map(|row| serde_json::from_str(&row?).context("decode durable move intent"))
60        .collect()
61}
62
63pub fn move_checkpoint_is_retained(path: &Path) -> Result<bool> {
64    Ok(load_move_operations()?.iter().any(|operation| {
65        operation.retains_checkpoint()
66            && operation
67                .checkpoint
68                .as_ref()
69                .is_some_and(|checkpoint| checkpoint.archive_path == path)
70    }))
71}
72
73pub fn request_move_cancellation(session_id: &str) -> Result<()> {
74    let session_id = session_id.to_owned();
75    submit_database_write("request_move_cancellation", move |connection| {
76        connection.execute(
77            "UPDATE session_moves SET operation_json=json_set(operation_json, '$.cancellation_requested', json('true'))
78             WHERE session_id=?1 AND json_extract(operation_json, '$.phase') IN ('preparing','closing_source','resuming_destination','starting_queue')",
79            [session_id],
80        )?;
81        Ok(())
82    })
83}
84
85pub fn clear_move_cancellation_for_retry(session_id: &str) -> Result<()> {
86    let session_id = session_id.to_owned();
87    submit_database_write("clear_move_cancellation_for_retry", move |connection| {
88        connection.execute(
89            "UPDATE session_moves SET operation_json=json_set(operation_json, '$.cancellation_requested', json('false')) WHERE session_id=?1",
90            [session_id],
91        )?;
92        Ok(())
93    })
94}
95
96/// Read the confirmation data without loading the conversation history.
97pub fn move_pending_work(session_id: &str) -> Result<(bool, Vec<MaterializedQueuedPrompt>)> {
98    let connection = open_reader(&database_path())?;
99    let running: Option<String> = connection
100        .query_row(
101            "SELECT execution_state FROM materialized_sessions WHERE session_id=?1",
102            [session_id],
103            |row| row.get(0),
104        )
105        .optional()?;
106    Ok((
107        running.as_deref() == Some("running"),
108        read_materialized_queued_prompts(&connection, session_id)?,
109    ))
110}
111
112#[cfg(test)]
113mod tests {
114    use super::*;
115    use mj_core::state::{MovePhase, MoveSelection, ResumeQueueDisposition};
116
117    fn operation(session: &SessionRecord) -> MoveOperation {
118        MoveOperation {
119            source_checkpoint_only: false,
120            operation_id: "move-one".into(),
121            selection: MoveSelection {
122                clear_resource_allocation: false,
123                session_id: session.id.clone(),
124                profile_id: Some("destination".into()),
125                target_template_id: Some("local".into()),
126                additional_mounts: Some(Vec::new()),
127                resource_allocation: None,
128            },
129            source_profile_id: session.last_profile.clone(),
130            source_target_template_id: session.target_template_id.clone(),
131            source_target: session.target.clone(),
132            source_native_session_id: session.native_session_id.clone(),
133            source_additional_mounts: session.additional_mounts.clone(),
134            source_resource_allocation: session.resource_allocation.clone(),
135            destination_target: None,
136            destination_native_session_id: None,
137            destination_store_id: None,
138            configuration_fingerprint: "fingerprint".into(),
139            checkpoint: session.checkpoint.clone(),
140            recovery_session: Some(session.clone()),
141            queue: ResumeQueueDisposition::Start,
142            phase: MovePhase::Preparing,
143            queue_admission_started: false,
144            queue_admission_finished: false,
145            cancellation_requested: false,
146            created_at: session.created_at.clone(),
147            updated_at: session.updated_at.clone(),
148            error: None,
149        }
150    }
151
152    #[test]
153    fn move_boundaries_survive_database_reopen_and_retain_the_source_locator() {
154        let directory = tempfile::tempdir().unwrap();
155        let path = directory.path().join("mj.sqlite3");
156        let session = super::super::tests::session("move-reopen", "project");
157        save_session_to(&path, &session).unwrap();
158        let mut intent = operation(&session);
159        for phase in [
160            MovePhase::Preparing,
161            MovePhase::ClosingSource,
162            MovePhase::ResumingDestination,
163            MovePhase::StartingQueue,
164            MovePhase::Failed,
165            MovePhase::Completed,
166        ] {
167            intent.phase = phase;
168            if phase == MovePhase::StartingQueue {
169                intent.queue_admission_started = true;
170                intent.destination_target = session.target.clone();
171                intent.destination_store_id = Some("durable-destination".into());
172            }
173            if phase == MovePhase::Completed {
174                intent.queue_admission_finished = true;
175            }
176            let connection = open(&path).unwrap();
177            save_move_operation_with(&connection, &intent).unwrap();
178            drop(connection);
179            let reopened = open_reader(&path).unwrap();
180            let restored = load_move_operation_with(&reopened, &session.id)
181                .unwrap()
182                .unwrap();
183            assert_eq!(restored, intent);
184            assert_eq!(restored.source_target, session.target);
185            assert_eq!(restored.retains_checkpoint(), phase != MovePhase::Completed);
186        }
187    }
188
189    #[test]
190    fn concurrent_phase_save_cannot_erase_durable_cancellation() {
191        let directory = tempfile::tempdir().unwrap();
192        let path = directory.path().join("mj.sqlite3");
193        let session = super::super::tests::session("move-cancel", "project");
194        save_session_to(&path, &session).unwrap();
195        let connection = open(&path).unwrap();
196        let mut intent = operation(&session);
197        save_move_operation_with(&connection, &intent).unwrap();
198        let mut cancelled = intent.clone();
199        cancelled.cancellation_requested = true;
200        save_move_operation_with(&connection, &cancelled).unwrap();
201        intent.phase = MovePhase::ClosingSource;
202        save_move_operation_with(&connection, &intent).unwrap();
203        let restored = load_move_operation_with(&connection, &session.id)
204            .unwrap()
205            .unwrap();
206        assert!(restored.cancellation_requested);
207        assert_eq!(restored.phase, MovePhase::ClosingSource);
208        intent.operation_id = "explicit-new-operation".into();
209        save_move_operation_with(&connection, &intent).unwrap();
210        assert!(
211            !load_move_operation_with(&connection, &session.id)
212                .unwrap()
213                .unwrap()
214                .cancellation_requested
215        );
216    }
217
218    #[test]
219    fn cancelling_partial_queue_admission_does_not_release_its_archive() {
220        let session = super::super::tests::session("move-queue", "project");
221        let mut intent = operation(&session);
222        intent.phase = MovePhase::Cancelled;
223        intent.queue_admission_started = true;
224        assert!(intent.retains_checkpoint());
225        intent.queue_admission_finished = true;
226        assert!(!intent.retains_checkpoint());
227    }
228
229    #[test]
230    fn destination_record_install_keeps_drafts_and_titles_edited_during_move() {
231        let directory = tempfile::tempdir().unwrap();
232        let path = directory.path().join("mj.sqlite3");
233        let mut stale = super::super::tests::session("move-draft", "project");
234        save_session_to(&path, &stale).unwrap();
235        let connection = open(&path).unwrap();
236        save_move_operation_with(&connection, &operation(&stale)).unwrap();
237        connection.execute("UPDATE sessions SET draft_input='keep this draft', session_title_override='new title' WHERE session_id=?1", [&stale.id]).unwrap();
238        stale.last_profile = "destination".into();
239        stale.state = SessionState::Provisioning;
240        save_session_to(&path, &stale).unwrap();
241        let current = load_state_from(&path)
242            .unwrap()
243            .sessions
244            .remove(&stale.id)
245            .unwrap();
246        assert_eq!(current.draft_input, "keep this draft");
247        assert_eq!(current.session_title_override.as_deref(), Some("new title"));
248        assert_eq!(current.last_profile, "destination");
249    }
250}
mj_controller/database/session_move.rs

mj_controller/database/
session_move.rs
