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            source_checkpoint_only: false,
157            operation_id: "move-one".into(),
158            selection: MoveSelection {
159                clear_resource_allocation: false,
160                session_id: session.id.clone(),
161                profile_id: Some("destination".into()),
162                target_template_id: Some("local".into()),
163                additional_mounts: Some(Vec::new()),
164                resource_allocation: None,
165            },
166            source_profile_id: session.last_profile.clone(),
167            source_target_template_id: session.target_template_id.clone(),
168            source_target: session.target.clone(),
169            source_native_session_id: session.native_session_id.clone(),
170            source_additional_mounts: session.additional_mounts.clone(),
171            source_resource_allocation: session.resource_allocation.clone(),
172            destination_target: None,
173            destination_native_session_id: None,
174            destination_store_id: None,
175            configuration_fingerprint: "fingerprint".into(),
176            checkpoint: session.checkpoint.clone(),
177            recovery_session: Some(session.clone()),
178            queue: ResumeQueueDisposition::Start,
179            phase: MovePhase::Preparing,
180            queue_admission_started: false,
181            queue_admission_finished: false,
182            cancellation_requested: false,
183            created_at: session.created_at.clone(),
184            updated_at: session.updated_at.clone(),
185            error: None,
186        }
187    }
188
189    #[test]
190    fn move_boundaries_survive_database_reopen_and_retain_the_source_locator() {
191        let directory = tempfile::tempdir().unwrap();
192        let path = directory.path().join("mj.sqlite3");
193        let session = super::super::tests::session("move-reopen", "project");
194        save_session_to(&path, &session).unwrap();
195        let mut intent = operation(&session);
196        for phase in [
197            MovePhase::Preparing,
198            MovePhase::ClosingSource,
199            MovePhase::ResumingDestination,
200            MovePhase::StartingQueue,
201            MovePhase::Failed,
202            MovePhase::Completed,
203        ] {
204            intent.phase = phase;
205            if phase == MovePhase::StartingQueue {
206                intent.queue_admission_started = true;
207                intent.destination_target = session.target.clone();
208                intent.destination_store_id = Some("durable-destination".into());
209            }
210            if phase == MovePhase::Completed {
211                intent.queue_admission_finished = true;
212            }
213            let connection = open(&path).unwrap();
214            save_move_operation_with(&connection, &intent).unwrap();
215            drop(connection);
216            let reopened = open_reader(&path).unwrap();
217            let restored = load_move_operation_with(&reopened, &session.id)
218                .unwrap()
219                .unwrap();
220            assert_eq!(restored, intent);
221            assert_eq!(restored.source_target, session.target);
222            assert_eq!(restored.retains_checkpoint(), phase != MovePhase::Completed);
223        }
224    }
225
226    #[test]
227    fn bulk_load_skips_a_move_intent_whose_harness_no_longer_decodes() {
228        let directory = tempfile::tempdir().unwrap();
229        let path = directory.path().join("mj.sqlite3");
230        let good = super::super::tests::session("move-good", "project");
231        let stale_session = super::super::tests::session("move-removed-harness", "project");
232        save_session_to(&path, &good).unwrap();
233        save_session_to(&path, &stale_session).unwrap();
234        let connection = open(&path).unwrap();
235        save_move_operation_with(&connection, &operation(&good)).unwrap();
236        let mut stale_operation = operation(&stale_session);
237        stale_operation.operation_id = "move-two".into();
238        save_move_operation_with(&connection, &stale_operation).unwrap();
239        // Simulate a row recorded before a harness was removed: rewrite the
240        // stored recovery snapshot to name a harness the current binary no
241        // longer knows, so the row fails to decode.
242        let rewritten = connection
243            .execute(
244                "UPDATE session_moves
245                 SET operation_json = replace(operation_json, '\"harness_kind\":\"codex\"', '\"harness_kind\":\"zcode\"')
246                 WHERE session_id = 'move-removed-harness'",
247                [],
248            )
249            .unwrap();
250        assert_eq!(
251            rewritten, 1,
252            "the test session must store a codex harness to rewrite"
253        );
254        let loaded = load_move_operations_with(&connection).unwrap();
255        assert_eq!(
256            loaded.len(),
257            1,
258            "the undecodable row must be skipped, not fail the load"
259        );
260        assert_eq!(loaded[0].selection.session_id, good.id);
261    }
262
263    #[test]
264    fn per_session_load_treats_an_intent_whose_harness_no_longer_decodes_as_absent() {
265        // The stop, checkpoint, recovery, and new-move paths each read the one
266        // intent for their session. A completed move whose recovery snapshot
267        // names a removed harness must not fail those operations.
268        let directory = tempfile::tempdir().unwrap();
269        let path = directory.path().join("mj.sqlite3");
270        let stale_session = super::super::tests::session("move-removed-harness", "project");
271        save_session_to(&path, &stale_session).unwrap();
272        let connection = open(&path).unwrap();
273        let mut stale_operation = operation(&stale_session);
274        stale_operation.phase = MovePhase::Completed;
275        save_move_operation_with(&connection, &stale_operation).unwrap();
276        let rewritten = connection
277            .execute(
278                "UPDATE session_moves
279                 SET operation_json = replace(operation_json, '\"harness_kind\":\"codex\"', '\"harness_kind\":\"zcode\"')
280                 WHERE session_id = 'move-removed-harness'",
281                [],
282            )
283            .unwrap();
284        assert_eq!(
285            rewritten, 1,
286            "the test session must store a codex harness to rewrite"
287        );
288        let loaded = load_move_operation_with(&connection, &stale_session.id)
289            .expect("an undecodable intent must not fail the read");
290        assert!(
291            loaded.is_none(),
292            "the undecodable intent is treated as absent"
293        );
294    }
295
296    #[test]
297    fn concurrent_phase_save_cannot_erase_durable_cancellation() {
298        let directory = tempfile::tempdir().unwrap();
299        let path = directory.path().join("mj.sqlite3");
300        let session = super::super::tests::session("move-cancel", "project");
301        save_session_to(&path, &session).unwrap();
302        let connection = open(&path).unwrap();
303        let mut intent = operation(&session);
304        save_move_operation_with(&connection, &intent).unwrap();
305        let mut cancelled = intent.clone();
306        cancelled.cancellation_requested = true;
307        save_move_operation_with(&connection, &cancelled).unwrap();
308        intent.phase = MovePhase::ClosingSource;
309        save_move_operation_with(&connection, &intent).unwrap();
310        let restored = load_move_operation_with(&connection, &session.id)
311            .unwrap()
312            .unwrap();
313        assert!(restored.cancellation_requested);
314        assert_eq!(restored.phase, MovePhase::ClosingSource);
315        intent.operation_id = "explicit-new-operation".into();
316        save_move_operation_with(&connection, &intent).unwrap();
317        assert!(
318            !load_move_operation_with(&connection, &session.id)
319                .unwrap()
320                .unwrap()
321                .cancellation_requested
322        );
323    }
324
325    #[test]
326    fn cancelling_partial_queue_admission_does_not_release_its_archive() {
327        let session = super::super::tests::session("move-queue", "project");
328        let mut intent = operation(&session);
329        intent.phase = MovePhase::Cancelled;
330        intent.queue_admission_started = true;
331        assert!(intent.retains_checkpoint());
332        intent.queue_admission_finished = true;
333        assert!(!intent.retains_checkpoint());
334    }
335
336    #[test]
337    fn destination_record_install_keeps_drafts_and_titles_edited_during_move() {
338        let directory = tempfile::tempdir().unwrap();
339        let path = directory.path().join("mj.sqlite3");
340        let mut stale = super::super::tests::session("move-draft", "project");
341        save_session_to(&path, &stale).unwrap();
342        let connection = open(&path).unwrap();
343        save_move_operation_with(&connection, &operation(&stale)).unwrap();
344        connection.execute("UPDATE sessions SET draft_input='keep this draft', session_title_override='new title' WHERE session_id=?1", [&stale.id]).unwrap();
345        stale.last_profile = "destination".into();
346        stale.state = SessionState::Provisioning;
347        save_session_to(&path, &stale).unwrap();
348        let current = load_state_from(&path)
349            .unwrap()
350            .sessions
351            .remove(&stale.id)
352            .unwrap();
353        assert_eq!(current.draft_input, "keep this draft");
354        assert_eq!(current.session_title_override.as_deref(), Some("new title"));
355        assert_eq!(current.last_profile, "destination");
356    }
357}
mj_controller/database/session_move.rs

mj_controller/database/
session_move.rs
