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    Ok(json.and_then(|json| decode_move_operation(session_id, &json)))
50}
51
52/// Decode one stored move intent, or `None` when it no longer decodes.
53///
54/// A move intent recorded before a harness was removed keeps that harness in
55/// its recovery snapshot, so it fails to decode under a binary that no longer
56/// knows the harness. Nothing deletes a completed intent, so such rows linger
57/// indefinitely (BrokkAi/mjolnir#1026). Every reader asks the same question,
58/// "is there a move to act on for this session?", and a move whose snapshot
59/// names a removed harness cannot be acted on, so all readers treat the row as
60/// absent with a warning rather than failing the operation that asked, whether
61/// that is daemon startup, a stop, a checkpoint, or a new move.
62fn decode_move_operation(session_id: &str, json: &str) -> Option<MoveOperation> {
63    match serde_json::from_str::<MoveOperation>(json) {
64        Ok(operation) => Some(operation),
65        Err(error) => {
66            tracing::warn!(
67                session_id,
68                %error,
69                "durable move intent no longer decodes; treating it as absent (its harness may have been removed)"
70            );
71            None
72        }
73    }
74}
75
76pub fn load_move_operations() -> Result<Vec<MoveOperation>> {
77    let connection = open_reader(&database_path())?;
78    load_move_operations_with(&connection)
79}
80
81pub(super) fn load_move_operations_with(connection: &Connection) -> Result<Vec<MoveOperation>> {
82    let mut statement = connection
83        .prepare("SELECT session_id, operation_json FROM session_moves ORDER BY session_id")?;
84    // The daemon loads every intent at startup, so one undecodable row would
85    // otherwise stop the daemon from starting; `decode_move_operation` says why
86    // such a row is skipped rather than fatal.
87    let rows = statement.query_map([], |row| {
88        Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
89    })?;
90    let mut operations = Vec::new();
91    for row in rows {
92        let (session_id, json) = row?;
93        if let Some(operation) = decode_move_operation(&session_id, &json) {
94            operations.push(operation);
95        }
96    }
97    Ok(operations)
98}
99
100pub fn move_checkpoint_is_retained(path: &Path) -> Result<bool> {
101    Ok(load_move_operations()?.iter().any(|operation| {
102        operation.retains_checkpoint()
103            && operation
104                .checkpoint
105                .as_ref()
106                .is_some_and(|checkpoint| checkpoint.archive_path == path)
107    }))
108}
109
110pub fn request_move_cancellation(session_id: &str) -> Result<()> {
111    let session_id = session_id.to_owned();
112    submit_database_write("request_move_cancellation", move |connection| {
113        connection.execute(
114            "UPDATE session_moves SET operation_json=json_set(operation_json, '$.cancellation_requested', json('true'))
115             WHERE session_id=?1 AND json_extract(operation_json, '$.phase') IN ('preparing','closing_source','resuming_destination','starting_queue')",
116            [session_id],
117        )?;
118        Ok(())
119    })
120}
121
122pub fn clear_move_cancellation_for_retry(session_id: &str) -> Result<()> {
123    let session_id = session_id.to_owned();
124    submit_database_write("clear_move_cancellation_for_retry", move |connection| {
125        connection.execute(
126            "UPDATE session_moves SET operation_json=json_set(operation_json, '$.cancellation_requested', json('false')) WHERE session_id=?1",
127            [session_id],
128        )?;
129        Ok(())
130    })
131}
132
133/// Read the confirmation data without loading the conversation history.
134pub fn move_pending_work(session_id: &str) -> Result<(bool, Vec<MaterializedQueuedPrompt>)> {
135    let connection = open_reader(&database_path())?;
136    let running: Option<String> = connection
137        .query_row(
138            "SELECT execution_state FROM materialized_sessions WHERE session_id=?1",
139            [session_id],
140            |row| row.get(0),
141        )
142        .optional()?;
143    Ok((
144        running.as_deref() == Some("running"),
145        read_materialized_queued_prompts(&connection, session_id)?,
146    ))
147}
148
149#[cfg(test)]
150mod tests {
151    use super::*;
152    use mj_core::state::{MovePhase, MoveSelection, ResumeQueueDisposition};
153
154    fn operation(session: &SessionRecord) -> MoveOperation {
155        MoveOperation {
156            in_place: false,
157            source_checkpoint_only: false,
158            operation_id: "move-one".into(),
159            selection: MoveSelection {
160                clear_resource_allocation: false,
161                session_id: session.id.clone(),
162                profile_id: Some("destination".into()),
163                target_template_id: Some("local".into()),
164                additional_mounts: Some(Vec::new()),
165                resource_allocation: None,
166            },
167            source_profile_id: session.last_profile.clone(),
168            source_target_template_id: session.target_template_id.clone(),
169            source_target: session.target.clone(),
170            source_native_session_id: session.native_session_id.clone(),
171            source_additional_mounts: session.additional_mounts.clone(),
172            source_resource_allocation: session.resource_allocation.clone(),
173            destination_target: None,
174            destination_native_session_id: None,
175            destination_store_id: None,
176            configuration_fingerprint: "fingerprint".into(),
177            checkpoint: session.checkpoint.clone(),
178            recovery_session: Some(session.clone()),
179            queue: ResumeQueueDisposition::Start,
180            phase: MovePhase::Preparing,
181            queue_admission_started: false,
182            queue_admission_finished: false,
183            cancellation_requested: false,
184            created_at: session.created_at.clone(),
185            updated_at: session.updated_at.clone(),
186            error: None,
187        }
188    }
189
190    #[test]
191    fn move_boundaries_survive_database_reopen_and_retain_the_source_locator() {
192        let directory = tempfile::tempdir().unwrap();
193        let path = directory.path().join("mj.sqlite3");
194        let session = super::super::tests::session("move-reopen", "project");
195        save_session_to(&path, &session).unwrap();
196        let mut intent = operation(&session);
197        for phase in [
198            MovePhase::Preparing,
199            MovePhase::ClosingSource,
200            MovePhase::ResumingDestination,
201            MovePhase::StartingQueue,
202            MovePhase::Failed,
203            MovePhase::Completed,
204        ] {
205            intent.phase = phase;
206            if phase == MovePhase::StartingQueue {
207                intent.queue_admission_started = true;
208                intent.destination_target = session.target.clone();
209                intent.destination_store_id = Some("durable-destination".into());
210            }
211            if phase == MovePhase::Completed {
212                intent.queue_admission_finished = true;
213            }
214            let connection = open(&path).unwrap();
215            save_move_operation_with(&connection, &intent).unwrap();
216            drop(connection);
217            let reopened = open_reader(&path).unwrap();
218            let restored = load_move_operation_with(&reopened, &session.id)
219                .unwrap()
220                .unwrap();
221            assert_eq!(restored, intent);
222            assert_eq!(restored.source_target, session.target);
223            assert_eq!(restored.retains_checkpoint(), phase != MovePhase::Completed);
224        }
225    }
226
227    #[test]
228    fn bulk_load_skips_a_move_intent_whose_harness_no_longer_decodes() {
229        let directory = tempfile::tempdir().unwrap();
230        let path = directory.path().join("mj.sqlite3");
231        let good = super::super::tests::session("move-good", "project");
232        let stale_session = super::super::tests::session("move-removed-harness", "project");
233        save_session_to(&path, &good).unwrap();
234        save_session_to(&path, &stale_session).unwrap();
235        let connection = open(&path).unwrap();
236        save_move_operation_with(&connection, &operation(&good)).unwrap();
237        let mut stale_operation = operation(&stale_session);
238        stale_operation.operation_id = "move-two".into();
239        save_move_operation_with(&connection, &stale_operation).unwrap();
240        // Simulate a row recorded before a harness was removed: rewrite the
241        // stored recovery snapshot to name a harness the current binary no
242        // longer knows, so the row fails to decode.
243        let rewritten = connection
244            .execute(
245                "UPDATE session_moves
246                 SET operation_json = replace(operation_json, '\"harness_kind\":\"codex\"', '\"harness_kind\":\"zcode\"')
247                 WHERE session_id = 'move-removed-harness'",
248                [],
249            )
250            .unwrap();
251        assert_eq!(
252            rewritten, 1,
253            "the test session must store a codex harness to rewrite"
254        );
255        let loaded = load_move_operations_with(&connection).unwrap();
256        assert_eq!(
257            loaded.len(),
258            1,
259            "the undecodable row must be skipped, not fail the load"
260        );
261        assert_eq!(loaded[0].selection.session_id, good.id);
262    }
263
264    #[test]
265    fn in_place_intent_round_trips_and_a_legacy_row_without_it_reads_as_a_fresh_environment() {
266        let directory = tempfile::tempdir().unwrap();
267        let path = directory.path().join("mj.sqlite3");
268        let session = super::super::tests::session("move-in-place", "project");
269        save_session_to(&path, &session).unwrap();
270        let connection = open(&path).unwrap();
271        let mut intent = operation(&session);
272        intent.in_place = true;
273        save_move_operation_with(&connection, &intent).unwrap();
274        let stored: String = connection
275            .query_row(
276                "SELECT operation_json FROM session_moves WHERE session_id=?1",
277                [&session.id],
278                |row| row.get(0),
279            )
280            .unwrap();
281        assert!(
282            stored.contains("\"in_place\":true"),
283            "the in-place choice must be durable: {stored}"
284        );
285        let restored = load_move_operation_with(&connection, &session.id)
286            .unwrap()
287            .unwrap();
288        assert_eq!(restored, intent);
289        // A row written before this field existed must still decode, as the
290        // full fresh-environment move it was.
291        let rewritten = connection
292            .execute(
293                "UPDATE session_moves
294                 SET operation_json = replace(operation_json, '\"in_place\":true,', '')
295                 WHERE session_id = ?1",
296                [&session.id],
297            )
298            .unwrap();
299        assert_eq!(rewritten, 1);
300        let legacy = load_move_operation_with(&connection, &session.id)
301            .unwrap()
302            .expect("a row without in_place must still decode");
303        assert!(!legacy.in_place);
304    }
305
306    #[test]
307    fn per_session_load_treats_an_intent_whose_harness_no_longer_decodes_as_absent() {
308        // The stop, checkpoint, recovery, and new-move paths each read the one
309        // intent for their session. A completed move whose recovery snapshot
310        // names a removed harness must not fail those operations.
311        let directory = tempfile::tempdir().unwrap();
312        let path = directory.path().join("mj.sqlite3");
313        let stale_session = super::super::tests::session("move-removed-harness", "project");
314        save_session_to(&path, &stale_session).unwrap();
315        let connection = open(&path).unwrap();
316        let mut stale_operation = operation(&stale_session);
317        stale_operation.phase = MovePhase::Completed;
318        save_move_operation_with(&connection, &stale_operation).unwrap();
319        let rewritten = connection
320            .execute(
321                "UPDATE session_moves
322                 SET operation_json = replace(operation_json, '\"harness_kind\":\"codex\"', '\"harness_kind\":\"zcode\"')
323                 WHERE session_id = 'move-removed-harness'",
324                [],
325            )
326            .unwrap();
327        assert_eq!(
328            rewritten, 1,
329            "the test session must store a codex harness to rewrite"
330        );
331        let loaded = load_move_operation_with(&connection, &stale_session.id)
332            .expect("an undecodable intent must not fail the read");
333        assert!(
334            loaded.is_none(),
335            "the undecodable intent is treated as absent"
336        );
337    }
338
339    #[test]
340    fn concurrent_phase_save_cannot_erase_durable_cancellation() {
341        let directory = tempfile::tempdir().unwrap();
342        let path = directory.path().join("mj.sqlite3");
343        let session = super::super::tests::session("move-cancel", "project");
344        save_session_to(&path, &session).unwrap();
345        let connection = open(&path).unwrap();
346        let mut intent = operation(&session);
347        save_move_operation_with(&connection, &intent).unwrap();
348        let mut cancelled = intent.clone();
349        cancelled.cancellation_requested = true;
350        save_move_operation_with(&connection, &cancelled).unwrap();
351        intent.phase = MovePhase::ClosingSource;
352        save_move_operation_with(&connection, &intent).unwrap();
353        let restored = load_move_operation_with(&connection, &session.id)
354            .unwrap()
355            .unwrap();
356        assert!(restored.cancellation_requested);
357        assert_eq!(restored.phase, MovePhase::ClosingSource);
358        intent.operation_id = "explicit-new-operation".into();
359        save_move_operation_with(&connection, &intent).unwrap();
360        assert!(
361            !load_move_operation_with(&connection, &session.id)
362                .unwrap()
363                .unwrap()
364                .cancellation_requested
365        );
366    }
367
368    #[test]
369    fn cancelling_partial_queue_admission_does_not_release_its_archive() {
370        let session = super::super::tests::session("move-queue", "project");
371        let mut intent = operation(&session);
372        intent.phase = MovePhase::Cancelled;
373        intent.queue_admission_started = true;
374        assert!(intent.retains_checkpoint());
375        intent.queue_admission_finished = true;
376        assert!(!intent.retains_checkpoint());
377    }
378
379    #[test]
380    fn destination_record_install_keeps_drafts_and_titles_edited_during_move() {
381        let directory = tempfile::tempdir().unwrap();
382        let path = directory.path().join("mj.sqlite3");
383        let mut stale = super::super::tests::session("move-draft", "project");
384        save_session_to(&path, &stale).unwrap();
385        let connection = open(&path).unwrap();
386        save_move_operation_with(&connection, &operation(&stale)).unwrap();
387        connection.execute("UPDATE sessions SET draft_input='keep this draft', session_title_override='new title' WHERE session_id=?1", [&stale.id]).unwrap();
388        stale.last_profile = "destination".into();
389        stale.state = SessionState::Provisioning;
390        save_session_to(&path, &stale).unwrap();
391        let current = load_state_from(&path)
392            .unwrap()
393            .sessions
394            .remove(&stale.id)
395            .unwrap();
396        assert_eq!(current.draft_input, "keep this draft");
397        assert_eq!(current.session_title_override.as_deref(), Some("new title"));
398        assert_eq!(current.last_profile, "destination");
399    }
400}
mj_controller/database/session_move.rs

mj_controller/database/
session_move.rs
