1use super::*;
3use crate::docker_snapshot_resume::{
4 commit_with_progress, progress_interval_seconds, remove_original_after, run_snapshot_prepare,
5 SnapshotReporter,
6};
7use crate::docker_snapshot_safety::{
8 acquire_snapshot_lock_with, df_free_bytes, preflight_snapshot_guarded, PreflightOptions,
9 SnapshotFailure, SnapshotLockRequest,
10};
11use crate::snapshot_resume_options::SnapshotSettings;
12
13pub fn resume_execution(
15 store: Option<&ExecutionStore>,
16 identifier: &str,
17 command: Option<&str>,
18 output_format: Option<&str>,
19) -> ExecutionResumeResult {
20 resume_execution_with(
21 store,
22 identifier,
23 command,
24 output_format,
25 &SystemCommandRunner,
26 &SystemResumeHooks,
27 )
28}
29
30pub fn resume_execution_with<R: CommandRunner, H: ResumeHooks>(
33 store: Option<&ExecutionStore>,
34 identifier: &str,
35 command: Option<&str>,
36 output_format: Option<&str>,
37 runner: &R,
38 hooks: &H,
39) -> ExecutionResumeResult {
40 resume_execution_with_resources(
41 store,
42 identifier,
43 command,
44 output_format,
45 runner,
46 hooks,
47 &json!({}),
48 )
49}
50
51pub fn resume_execution_with_options(
52 store: Option<&ExecutionStore>,
53 identifier: &str,
54 command: Option<&str>,
55 output_format: Option<&str>,
56 options: &crate::args_parser::WrapperOptions,
57) -> ExecutionResumeResult {
58 let mut overrides = crate::docker_resource_options::specs(options);
59 overrides["removeOriginal"] = json!(options.remove_original);
60 overrides["labels"] = json!(options.labels);
61 crate::snapshot_resume_options::insert_overrides(&options.snapshot, &mut overrides);
62 resume_execution_with_resources(
63 store,
64 identifier,
65 command,
66 output_format,
67 &SystemCommandRunner,
68 &SystemResumeHooks,
69 &overrides,
70 )
71}
72
73#[allow(clippy::too_many_arguments)]
74pub fn resume_execution_with_resources<R: CommandRunner, H: ResumeHooks>(
75 store: Option<&ExecutionStore>,
76 identifier: &str,
77 command: Option<&str>,
78 output_format: Option<&str>,
79 runner: &R,
80 hooks: &H,
81 overrides: &Value,
82) -> ExecutionResumeResult {
83 let Some(store) = store else {
84 return ExecutionResumeResult {
85 success: false,
86 output: None,
87 error: Some("Execution tracking is disabled.".to_string()),
88 exit_code: None,
89 };
90 };
91
92 let Some(mut record) = store.get(identifier) else {
93 return ExecutionResumeResult {
94 success: false,
95 output: None,
96 error: Some(format!(
97 "No execution found with UUID or session name: {}",
98 identifier
99 )),
100 exit_code: None,
101 };
102 };
103
104 if crate::launch_owner::has_active_launch(&record) {
105 return ExecutionResumeResult {
106 success: false,
107 output: None,
108 error: Some("An execution launch is already reserved.".into()),
109 exit_code: None,
110 };
111 }
112 let previous = record.clone();
113 let probe = probe_session(&record, runner);
114 let resources = crate::resume_resources::prepare(&record, &probe, overrides, runner);
115 let (limits, resolved, old_memory) = match resources {
116 Ok(result) => result,
117 Err(error) => {
118 return ExecutionResumeResult {
119 success: false,
120 output: None,
121 error: Some(error),
122 exit_code: None,
123 }
124 }
125 };
126 let mut planning_record = record.clone();
127 if let Some(labels) = overrides
128 .get("labels")
129 .and_then(Value::as_array)
130 .filter(|labels| !labels.is_empty())
131 {
132 let mut merged = record_strings(&record, "labels");
133 for label in labels.iter().filter_map(Value::as_str) {
134 let key = label.split('=').next().unwrap_or(label);
135 if let Some(index) = merged
136 .iter()
137 .position(|existing| existing.split('=').next() == Some(key))
138 {
139 merged[index] = label.to_string();
140 } else {
141 merged.push(label.to_string());
142 }
143 }
144 planning_record
145 .options
146 .insert("labels".into(), json!(merged));
147 }
148 if planning_record.options.get("labels") != record.options.get("labels") {
149 if record_option(&record, "isolated") != Some("docker") {
150 return ExecutionResumeResult {
151 success: false,
152 output: None,
153 error: Some("Docker labels require a Docker execution when resuming.".into()),
154 exit_code: None,
155 };
156 }
157 planning_record
158 .options
159 .insert("forceSnapshot".into(), json!(true));
160 }
161 if let Some(limits) = &limits {
162 planning_record
163 .options
164 .insert("resourceLimits".into(), json!(limits));
165 }
166 let mut plan =
167 match build_resume_plan_with_limits(&planning_record, command, &probe, limits.clone()) {
168 Ok(plan) => plan,
169 Err(error) => {
170 return ExecutionResumeResult {
171 success: false,
172 output: None,
173 error: Some(error),
174 exit_code: None,
175 }
176 }
177 };
178 if let Some(limits) = limits {
179 plan.resource_limits = limits.clone();
180 let changed = ["memory", "memorySwap", "memoryReservation", "cpus"]
181 .iter()
182 .any(|k| overrides.get(k).and_then(Value::as_str).is_some())
183 || record.options.contains_key("cpuPenaltyConfig");
184 if plan.mode == ResumeMode::DockerStart && changed {
185 let mut args = vec!["update".into()];
186 args.extend(limits.into_iter().filter(|s| {
187 [
188 "--memory=",
189 "--memory-swap=",
190 "--memory-reservation=",
191 "--cpus=",
192 "--cpu-quota=",
193 "--cpu-period=",
194 ]
195 .iter()
196 .any(|f| s.starts_with(f))
197 }));
198 args.push(plan.session_name.clone());
199 plan.steps.insert(
200 0,
201 ResumeStep {
202 command: docker_command().to_string_lossy().into(),
203 args,
204 description: "Apply resume limits before start".into(),
205 },
206 );
207 }
208 }
209 let mut container_id: Option<String> = None;
210 let mut attempt = crate::execution_attempt::create_attempt(
211 &record,
212 plan.mode.as_str(),
213 active_session_name(&plan),
214 );
215 let mut lifecycle_record = record.clone();
216 lifecycle_record.attempt = Some(attempt.clone());
217 crate::execution_attempt::append_lifecycle(&lifecycle_record, "resume-started", json!({}));
218 apply_resume_to_record(&mut record, &plan, None);
219 if let Some(labels) = planning_record.options.get("labels") {
220 record.options.insert("labels".into(), labels.clone());
221 }
222 record.attempt = Some(attempt.clone());
223 if let Some(resolved) = resolved {
224 record.options.insert("resolvedLimits".into(), resolved);
225 }
226 let mut requested = record
227 .options
228 .get("resourceLimitSpecs")
229 .filter(|value| value.is_object())
230 .cloned()
231 .unwrap_or_else(|| json!({}));
232 for k in ["memory", "memorySwap", "memoryReservation", "cpus"] {
233 if let Some(value) = overrides.get(k).filter(|v| v.is_string()) {
234 requested[k] = value.clone();
235 }
236 }
237 record
238 .options
239 .insert("resourceLimitSpecs".into(), requested);
240 if record.options.contains_key("cpuPenaltyConfig")
241 && overrides.get("cpus").and_then(Value::as_str).is_some()
242 {
243 record
244 .options
245 .insert("baseResourceLimits".into(), json!(&plan.resource_limits));
246 }
247 crate::launch_owner::mark_launch(&mut record);
248 if let Err(error) = store.reserve_launch(&record, &previous) {
249 return ExecutionResumeResult {
250 success: false,
251 output: None,
252 error: Some(format!("Launch reservation failed: {}", error)),
253 exit_code: None,
254 };
255 }
256
257 if let Some(spec) = overrides.get("memory").and_then(Value::as_str) {
258 let new = record
259 .options
260 .get("resolvedLimits")
261 .and_then(|v| v.get("memory"))
262 .cloned()
263 .unwrap_or(Value::Null);
264 if !record.log_path.is_empty() {
265 append_log_file(
266 &PathBuf::from(&record.log_path),
267 &format!(
268 "Resume memory limit: {} -> {} ({})\n",
269 old_memory, new, spec
270 ),
271 );
272 }
273 }
274
275 if plan.mode == ResumeMode::Relaunch {
276 let mut launch_options = plan.launch_options.clone().unwrap_or_default();
277 launch_options.attempt_number = attempt.number;
278 let launch_result = hooks.relaunch(&plan.backend, &plan.command, &launch_options);
279 if !launch_result.success {
280 let _ = store.save(&previous);
281 crate::execution_attempt::append_lifecycle(
282 &lifecycle_record,
283 "launch-failed",
284 json!({ "error": launch_result.message }),
285 );
286 return ExecutionResumeResult {
287 success: false,
288 output: None,
289 error: Some(format!(
290 "Failed to relaunch {} session \"{}\": {}",
291 plan.backend, plan.session_name, launch_result.message
292 )),
293 exit_code: None,
294 };
295 }
296 container_id = launch_result.container_id;
297 } else {
298 let mut snapshot_created = false;
299 let mut derived_created = false;
300 let handoff = plan.mode == ResumeMode::DockerStart
301 && command.is_some()
302 && record_flag(&previous, "commandHandoff");
303 let settings = SnapshotSettings::from_overrides(overrides);
304 let notify = |line: &str| eprintln!("{}", line);
305 let reporter = SnapshotReporter {
306 log_path: &record.log_path,
307 notify: ¬ify,
308 };
309 let mut failure_exit_code = None;
310 let launch_result = (|| -> Result<(), String> {
311 let mut exit_code_of = |failure: SnapshotFailure| {
312 failure_exit_code = failure.exit_code;
313 failure.message
314 };
315 let _snapshot_lock = if plan.mode == ResumeMode::DockerSnapshot {
316 Some(
317 acquire_snapshot_lock_with(&SnapshotLockRequest {
318 lock_path: settings.lock_path.clone(),
319 holder: Some(&plan.session_name),
320 wait_seconds: settings.wait_seconds,
321 notify: ¬ify,
322 })
323 .map_err(&mut exit_code_of)?,
324 )
325 } else {
326 None
327 };
328 if plan.mode == ResumeMode::DockerSnapshot {
329 if let Some(prepare) = &settings.prepare {
330 run_snapshot_prepare(&plan.session_name, prepare, runner, &reporter);
331 }
332 let options = PreflightOptions {
333 size_hint: settings.size_hint,
334 disk_wait_seconds: settings.disk_wait_seconds,
335 disk_poll_seconds: settings.disk_poll_seconds,
336 notify: ¬ify,
337 ..Default::default()
338 };
339 let message =
340 preflight_snapshot_guarded(&plan.session_name, runner, &options, &|root| {
341 df_free_bytes(runner, root)
342 })
343 .map_err(&mut exit_code_of)?;
344 plan.message.push_str(&format!("\n{}", message));
345 if !record.log_path.is_empty() {
346 append_log_file(&PathBuf::from(&record.log_path), &format!("{}\n", message));
347 }
348 }
349 if handoff {
350 let copied = crate::docker_command_handoff::write_command_handoff(
351 &plan.session_name,
352 &plan.command,
353 record_option(&previous, "shell").unwrap_or("auto"),
354 record_flag(&previous, "keepAlive"),
355 runner,
356 )?;
357 if !copied.success {
358 return Err(if copied.stderr.trim().is_empty() {
359 copied.error.unwrap_or_else(|| {
360 "Could not copy replacement command into the stopped container".into()
361 })
362 } else {
363 copied.stderr.trim().into()
364 });
365 }
366 }
367 if plan.mode == ResumeMode::DockerStart && record_flag(&previous, "commandHandoff") {
368 let env = crate::execution_env::build_execution_env(
370 Some(&record.uuid),
371 Some(&plan.session_name),
372 attempt.number,
373 );
374 let _ = crate::docker_command_handoff::write_execution_env_handoff(
375 &plan.session_name,
376 &env,
377 runner,
378 );
379 }
380 for step in &plan.steps {
381 let result = if step.args[0] == "commit" {
382 commit_with_progress(
383 runner,
384 &step.command,
385 &step.args,
386 &plan.session_name,
387 &reporter,
388 settings
389 .progress_interval_seconds
390 .unwrap_or_else(progress_interval_seconds),
391 )
392 } else {
393 runner.run(&step.command, &step.args)
394 };
395 if !result.success {
396 let detail = if !result.stderr.trim().is_empty() {
397 result.stderr.trim().to_string()
398 } else {
399 result.error.clone().unwrap_or_else(|| {
400 format!(
401 "exit code {}",
402 result
403 .status
404 .map(|code| code.to_string())
405 .unwrap_or_else(|| "unknown".to_string())
406 )
407 })
408 };
409 return Err(detail);
410 }
411 if step.args[0] == "commit" {
412 snapshot_created = true;
413 }
414 if step.args[0] == "create" {
415 derived_created = true;
416 }
417 let stdout = result.stdout.trim().to_string();
418 if !stdout.is_empty() && matches!(step.args[0].as_str(), "run" | "create") {
419 container_id = Some(stdout);
420 }
421 }
422 if plan.mode == ResumeMode::DockerSnapshot
423 && overrides.get("removeOriginal").and_then(Value::as_bool) == Some(true)
424 && settings.remove_original_after <= 0.0
425 {
426 let running = runner.run(
427 &docker_command().to_string_lossy(),
428 &[
429 "inspect".into(),
430 "-f".into(),
431 "{{.State.Running}}".into(),
432 active_session_name(&plan).into(),
433 ],
434 );
435 let note = if running.success && running.stdout.trim() == "true" {
436 let removed = runner.run(
437 &docker_command().to_string_lossy(),
438 &["rm".into(), plan.session_name.clone()],
439 );
440 if removed.success {
441 format!("Stopped original container removed: {}", plan.session_name)
442 } else {
443 format!("Original container retained: {}", removed.stderr.trim())
444 }
445 } else {
446 "Original container retained: the snapshot container is not running.".into()
447 };
448 plan.message.push_str(&format!("\n{}", note));
449 if !record.log_path.is_empty() {
450 append_log_file(&PathBuf::from(&record.log_path), &format!("{}\n", note));
451 }
452 }
453 Ok(())
454 })();
455 if let Err(detail) = launch_result {
456 if derived_created {
457 runner.run(
458 &docker_command().to_string_lossy(),
459 &["rm".into(), active_session_name(&plan).into()],
460 );
461 }
462 if snapshot_created {
463 runner.run(
464 &docker_command().to_string_lossy(),
465 &["rmi".into(), plan.snapshot_image.clone().unwrap()],
466 );
467 }
468 if handoff {
469 let _ = crate::docker_command_handoff::write_command_handoff(
470 &plan.session_name,
471 &previous.command,
472 record_option(&previous, "shell").unwrap_or("auto"),
473 record_flag(&previous, "keepAlive"),
474 runner,
475 );
476 }
477 let _ = store.save(&previous);
478 crate::execution_attempt::append_lifecycle(
479 &lifecycle_record,
480 "launch-failed",
481 json!({ "error": detail }),
482 );
483 return ExecutionResumeResult {
484 success: false,
485 output: None,
486 error: Some(format!(
487 "Failed to resume {} session \"{}\": {}",
488 plan.backend, plan.session_name, detail
489 )),
490 exit_code: failure_exit_code,
491 };
492 }
493
494 if let Some(line) = build_resource_limits_status_line(&plan.resource_limits) {
495 if !record.log_path.is_empty() {
496 append_log_file(&PathBuf::from(&record.log_path), &format!("{}\n", line));
497 }
498 }
499 }
500
501 let previous_session_name = plan
502 .new_session_name
503 .as_ref()
504 .map(|_| plan.session_name.clone());
505 if let Some(id) = container_id.as_deref() {
506 record.options.insert("containerId".into(), json!(id));
507 }
508 record.options.insert("launchPending".into(), json!(false));
509 attempt.launch_accepted_at = Some(Utc::now().to_rfc3339());
510 record
511 .options
512 .insert("resumedAt".into(), json!(attempt.started_at));
513 record.attempt = Some(attempt.clone());
514 if let Err(error) = store.save(&record) {
515 if plan.backend == "docker" {
516 let _ = hooks.attach_watcher(active_session_name(&plan), &record);
517 }
518 let stopped = plan.backend == "docker"
519 && runner
520 .run(
521 &docker_command().to_string_lossy(),
522 &["stop".into(), active_session_name(&plan).into()],
523 )
524 .success;
525 if stopped {
526 record.status = ExecutionStatus::Executed;
527 record.exit_code = Some(-1);
528 record.end_time = Some(Utc::now().to_rfc3339());
529 record.options.insert("launchPending".into(), json!(false));
530 let _ = store.save(&record);
531 }
532 return ExecutionResumeResult {
533 success: false,
534 output: None,
535 error: Some(
536 json!({"code": "LAUNCH_PERSISTENCE_FAILED", "uuid": record.uuid,
537 "containerName": active_session_name(&plan), "running": !stopped,
538 "error": error})
539 .to_string(),
540 ),
541 exit_code: None,
542 };
543 }
544 crate::execution_attempt::append_lifecycle(&record, "launch-accepted", json!({}));
545 if plan.backend == "docker" {
546 let attachment = hooks.attach_watcher(active_session_name(&plan), &record);
547 let fields = match &attachment {
548 Ok(()) => json!({"watcherAttachedAt": Utc::now().to_rfc3339()}),
549 Err(error) => json!({"watcherError": error}),
550 };
551 match store.patch_attempt(&record.uuid, attempt.number, fields) {
552 Ok(Some(current)) => crate::execution_attempt::append_lifecycle(
553 ¤t,
554 if attachment.is_ok() {
555 "watcher-attached"
556 } else {
557 "watcher-attachment-failed"
558 },
559 json!({"error": attachment.as_ref().err()}),
560 ),
561 Ok(None) => {}
562 Err(error) => {
563 return ExecutionResumeResult {
564 success: false,
565 output: None,
566 error: Some(error),
567 exit_code: None,
568 }
569 }
570 }
571 if let Err(error) = attachment {
572 return ExecutionResumeResult {
573 success: false,
574 output: None,
575 error: Some(format!(
576 "Launch accepted, but completion watcher attachment failed: {}",
577 error
578 )),
579 exit_code: None,
580 };
581 }
582 }
583
584 let remove_after = SnapshotSettings::from_overrides(overrides).remove_original_after;
585 if plan.mode == ResumeMode::DockerSnapshot
586 && overrides.get("removeOriginal").and_then(Value::as_bool) == Some(true)
587 && remove_after > 0.0
588 {
589 let notify = |line: &str| eprintln!("{}", line);
590 let note = remove_original_after(
591 &plan.session_name,
592 active_session_name(&plan),
593 remove_after,
594 runner,
595 &SnapshotReporter {
596 log_path: &record.log_path,
597 notify: ¬ify,
598 },
599 &std::thread::sleep,
600 );
601 plan.message.push_str(&format!("\n{}", note));
602 }
603
604 let session_name = record_option(&record, "sessionName")
605 .unwrap_or(&plan.session_name)
606 .to_string();
607
608 ExecutionResumeResult {
609 success: true,
610 output: Some(format_resume_result(
611 &ResumeResultFields {
612 identifier,
613 uuid: &record.uuid,
614 mode: plan.mode,
615 backend: &plan.backend,
616 session_name: &session_name,
617 previous_session_name: previous_session_name.as_deref(),
618 snapshot_image: plan.snapshot_image.as_deref(),
619 resource_limits: &plan.resource_limits,
620 command: &plan.command,
621 message: &plan.message,
622 },
623 output_format,
624 )),
625 error: None,
626 exit_code: None,
627 }
628}
629
630#[cfg(test)]
631#[path = "execution_resume_cases.rs"]
632mod tests;