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", "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 "--cpus=",
191 "--cpu-quota=",
192 "--cpu-period=",
193 ]
194 .iter()
195 .any(|f| s.starts_with(f))
196 }));
197 args.push(plan.session_name.clone());
198 plan.steps.insert(
199 0,
200 ResumeStep {
201 command: docker_command().to_string_lossy().into(),
202 args,
203 description: "Apply resume limits before start".into(),
204 },
205 );
206 }
207 }
208 let mut container_id: Option<String> = None;
209 let mut attempt = crate::execution_attempt::create_attempt(
210 &record,
211 plan.mode.as_str(),
212 active_session_name(&plan),
213 );
214 let mut lifecycle_record = record.clone();
215 lifecycle_record.attempt = Some(attempt.clone());
216 crate::execution_attempt::append_lifecycle(&lifecycle_record, "resume-started", json!({}));
217 apply_resume_to_record(&mut record, &plan, None);
218 if let Some(labels) = planning_record.options.get("labels") {
219 record.options.insert("labels".into(), labels.clone());
220 }
221 record.attempt = Some(attempt.clone());
222 if let Some(resolved) = resolved {
223 record.options.insert("resolvedLimits".into(), resolved);
224 }
225 let mut requested = record
226 .options
227 .get("resourceLimitSpecs")
228 .filter(|value| value.is_object())
229 .cloned()
230 .unwrap_or_else(|| json!({}));
231 for k in ["memory", "memorySwap", "cpus"] {
232 if let Some(value) = overrides.get(k).filter(|v| v.is_string()) {
233 requested[k] = value.clone();
234 }
235 }
236 record
237 .options
238 .insert("resourceLimitSpecs".into(), requested);
239 if record.options.contains_key("cpuPenaltyConfig")
240 && overrides.get("cpus").and_then(Value::as_str).is_some()
241 {
242 record
243 .options
244 .insert("baseResourceLimits".into(), json!(&plan.resource_limits));
245 }
246 crate::launch_owner::mark_launch(&mut record);
247 if let Err(error) = store.reserve_launch(&record, &previous) {
248 return ExecutionResumeResult {
249 success: false,
250 output: None,
251 error: Some(format!("Launch reservation failed: {}", error)),
252 exit_code: None,
253 };
254 }
255
256 if let Some(spec) = overrides.get("memory").and_then(Value::as_str) {
257 let new = record
258 .options
259 .get("resolvedLimits")
260 .and_then(|v| v.get("memory"))
261 .cloned()
262 .unwrap_or(Value::Null);
263 if !record.log_path.is_empty() {
264 append_log_file(
265 &PathBuf::from(&record.log_path),
266 &format!(
267 "Resume memory limit: {} -> {} ({})\n",
268 old_memory, new, spec
269 ),
270 );
271 }
272 }
273
274 if plan.mode == ResumeMode::Relaunch {
275 let launch_options = plan.launch_options.clone().unwrap_or_default();
276 let launch_result = hooks.relaunch(&plan.backend, &plan.command, &launch_options);
277 if !launch_result.success {
278 let _ = store.save(&previous);
279 crate::execution_attempt::append_lifecycle(
280 &lifecycle_record,
281 "launch-failed",
282 json!({ "error": launch_result.message }),
283 );
284 return ExecutionResumeResult {
285 success: false,
286 output: None,
287 error: Some(format!(
288 "Failed to relaunch {} session \"{}\": {}",
289 plan.backend, plan.session_name, launch_result.message
290 )),
291 exit_code: None,
292 };
293 }
294 container_id = launch_result.container_id;
295 } else {
296 let mut snapshot_created = false;
297 let mut derived_created = false;
298 let handoff = plan.mode == ResumeMode::DockerStart
299 && command.is_some()
300 && record_flag(&previous, "commandHandoff");
301 let settings = SnapshotSettings::from_overrides(overrides);
302 let notify = |line: &str| eprintln!("{}", line);
303 let reporter = SnapshotReporter {
304 log_path: &record.log_path,
305 notify: ¬ify,
306 };
307 let mut failure_exit_code = None;
308 let launch_result = (|| -> Result<(), String> {
309 let mut exit_code_of = |failure: SnapshotFailure| {
310 failure_exit_code = failure.exit_code;
311 failure.message
312 };
313 let _snapshot_lock = if plan.mode == ResumeMode::DockerSnapshot {
314 Some(
315 acquire_snapshot_lock_with(&SnapshotLockRequest {
316 lock_path: settings.lock_path.clone(),
317 holder: Some(&plan.session_name),
318 wait_seconds: settings.wait_seconds,
319 notify: ¬ify,
320 })
321 .map_err(&mut exit_code_of)?,
322 )
323 } else {
324 None
325 };
326 if plan.mode == ResumeMode::DockerSnapshot {
327 if let Some(prepare) = &settings.prepare {
328 run_snapshot_prepare(&plan.session_name, prepare, runner, &reporter);
329 }
330 let options = PreflightOptions {
331 size_hint: settings.size_hint,
332 disk_wait_seconds: settings.disk_wait_seconds,
333 disk_poll_seconds: settings.disk_poll_seconds,
334 notify: ¬ify,
335 ..Default::default()
336 };
337 let message =
338 preflight_snapshot_guarded(&plan.session_name, runner, &options, &|root| {
339 df_free_bytes(runner, root)
340 })
341 .map_err(&mut exit_code_of)?;
342 plan.message.push_str(&format!("\n{}", message));
343 if !record.log_path.is_empty() {
344 append_log_file(&PathBuf::from(&record.log_path), &format!("{}\n", message));
345 }
346 }
347 if handoff {
348 let copied = crate::docker_command_handoff::write_command_handoff(
349 &plan.session_name,
350 &plan.command,
351 record_option(&previous, "shell").unwrap_or("auto"),
352 record_flag(&previous, "keepAlive"),
353 runner,
354 )?;
355 if !copied.success {
356 return Err(if copied.stderr.trim().is_empty() {
357 copied.error.unwrap_or_else(|| {
358 "Could not copy replacement command into the stopped container".into()
359 })
360 } else {
361 copied.stderr.trim().into()
362 });
363 }
364 }
365 for step in &plan.steps {
366 let result = if step.args[0] == "commit" {
367 commit_with_progress(
368 runner,
369 &step.command,
370 &step.args,
371 &plan.session_name,
372 &reporter,
373 settings
374 .progress_interval_seconds
375 .unwrap_or_else(progress_interval_seconds),
376 )
377 } else {
378 runner.run(&step.command, &step.args)
379 };
380 if !result.success {
381 let detail = if !result.stderr.trim().is_empty() {
382 result.stderr.trim().to_string()
383 } else {
384 result.error.clone().unwrap_or_else(|| {
385 format!(
386 "exit code {}",
387 result
388 .status
389 .map(|code| code.to_string())
390 .unwrap_or_else(|| "unknown".to_string())
391 )
392 })
393 };
394 return Err(detail);
395 }
396 if step.args[0] == "commit" {
397 snapshot_created = true;
398 }
399 if step.args[0] == "create" {
400 derived_created = true;
401 }
402 let stdout = result.stdout.trim().to_string();
403 if !stdout.is_empty() && matches!(step.args[0].as_str(), "run" | "create") {
404 container_id = Some(stdout);
405 }
406 }
407 if plan.mode == ResumeMode::DockerSnapshot
408 && overrides.get("removeOriginal").and_then(Value::as_bool) == Some(true)
409 && settings.remove_original_after <= 0.0
410 {
411 let running = runner.run(
412 &docker_command().to_string_lossy(),
413 &[
414 "inspect".into(),
415 "-f".into(),
416 "{{.State.Running}}".into(),
417 active_session_name(&plan).into(),
418 ],
419 );
420 let note = if running.success && running.stdout.trim() == "true" {
421 let removed = runner.run(
422 &docker_command().to_string_lossy(),
423 &["rm".into(), plan.session_name.clone()],
424 );
425 if removed.success {
426 format!("Stopped original container removed: {}", plan.session_name)
427 } else {
428 format!("Original container retained: {}", removed.stderr.trim())
429 }
430 } else {
431 "Original container retained: the snapshot container is not running.".into()
432 };
433 plan.message.push_str(&format!("\n{}", note));
434 if !record.log_path.is_empty() {
435 append_log_file(&PathBuf::from(&record.log_path), &format!("{}\n", note));
436 }
437 }
438 Ok(())
439 })();
440 if let Err(detail) = launch_result {
441 if derived_created {
442 runner.run(
443 &docker_command().to_string_lossy(),
444 &["rm".into(), active_session_name(&plan).into()],
445 );
446 }
447 if snapshot_created {
448 runner.run(
449 &docker_command().to_string_lossy(),
450 &["rmi".into(), plan.snapshot_image.clone().unwrap()],
451 );
452 }
453 if handoff {
454 let _ = crate::docker_command_handoff::write_command_handoff(
455 &plan.session_name,
456 &previous.command,
457 record_option(&previous, "shell").unwrap_or("auto"),
458 record_flag(&previous, "keepAlive"),
459 runner,
460 );
461 }
462 let _ = store.save(&previous);
463 crate::execution_attempt::append_lifecycle(
464 &lifecycle_record,
465 "launch-failed",
466 json!({ "error": detail }),
467 );
468 return ExecutionResumeResult {
469 success: false,
470 output: None,
471 error: Some(format!(
472 "Failed to resume {} session \"{}\": {}",
473 plan.backend, plan.session_name, detail
474 )),
475 exit_code: failure_exit_code,
476 };
477 }
478
479 if let Some(line) = build_resource_limits_status_line(&plan.resource_limits) {
480 if !record.log_path.is_empty() {
481 append_log_file(&PathBuf::from(&record.log_path), &format!("{}\n", line));
482 }
483 }
484 }
485
486 let previous_session_name = plan
487 .new_session_name
488 .as_ref()
489 .map(|_| plan.session_name.clone());
490 if let Some(id) = container_id.as_deref() {
491 record.options.insert("containerId".into(), json!(id));
492 }
493 record.options.insert("launchPending".into(), json!(false));
494 attempt.launch_accepted_at = Some(Utc::now().to_rfc3339());
495 record
496 .options
497 .insert("resumedAt".into(), json!(attempt.started_at));
498 record.attempt = Some(attempt.clone());
499 if let Err(error) = store.save(&record) {
500 if plan.backend == "docker" {
501 let _ = hooks.attach_watcher(active_session_name(&plan), &record);
502 }
503 let stopped = plan.backend == "docker"
504 && runner
505 .run(
506 &docker_command().to_string_lossy(),
507 &["stop".into(), active_session_name(&plan).into()],
508 )
509 .success;
510 if stopped {
511 record.status = ExecutionStatus::Executed;
512 record.exit_code = Some(-1);
513 record.end_time = Some(Utc::now().to_rfc3339());
514 record.options.insert("launchPending".into(), json!(false));
515 let _ = store.save(&record);
516 }
517 return ExecutionResumeResult {
518 success: false,
519 output: None,
520 error: Some(
521 json!({"code": "LAUNCH_PERSISTENCE_FAILED", "uuid": record.uuid,
522 "containerName": active_session_name(&plan), "running": !stopped,
523 "error": error})
524 .to_string(),
525 ),
526 exit_code: None,
527 };
528 }
529 crate::execution_attempt::append_lifecycle(&record, "launch-accepted", json!({}));
530 if plan.backend == "docker" {
531 let attachment = hooks.attach_watcher(active_session_name(&plan), &record);
532 let fields = match &attachment {
533 Ok(()) => json!({"watcherAttachedAt": Utc::now().to_rfc3339()}),
534 Err(error) => json!({"watcherError": error}),
535 };
536 match store.patch_attempt(&record.uuid, attempt.number, fields) {
537 Ok(Some(current)) => crate::execution_attempt::append_lifecycle(
538 ¤t,
539 if attachment.is_ok() {
540 "watcher-attached"
541 } else {
542 "watcher-attachment-failed"
543 },
544 json!({"error": attachment.as_ref().err()}),
545 ),
546 Ok(None) => {}
547 Err(error) => {
548 return ExecutionResumeResult {
549 success: false,
550 output: None,
551 error: Some(error),
552 exit_code: None,
553 }
554 }
555 }
556 if let Err(error) = attachment {
557 return ExecutionResumeResult {
558 success: false,
559 output: None,
560 error: Some(format!(
561 "Launch accepted, but completion watcher attachment failed: {}",
562 error
563 )),
564 exit_code: None,
565 };
566 }
567 }
568
569 let remove_after = SnapshotSettings::from_overrides(overrides).remove_original_after;
570 if plan.mode == ResumeMode::DockerSnapshot
571 && overrides.get("removeOriginal").and_then(Value::as_bool) == Some(true)
572 && remove_after > 0.0
573 {
574 let notify = |line: &str| eprintln!("{}", line);
575 let note = remove_original_after(
576 &plan.session_name,
577 active_session_name(&plan),
578 remove_after,
579 runner,
580 &SnapshotReporter {
581 log_path: &record.log_path,
582 notify: ¬ify,
583 },
584 &std::thread::sleep,
585 );
586 plan.message.push_str(&format!("\n{}", note));
587 }
588
589 let session_name = record_option(&record, "sessionName")
590 .unwrap_or(&plan.session_name)
591 .to_string();
592
593 ExecutionResumeResult {
594 success: true,
595 output: Some(format_resume_result(
596 &ResumeResultFields {
597 identifier,
598 uuid: &record.uuid,
599 mode: plan.mode,
600 backend: &plan.backend,
601 session_name: &session_name,
602 previous_session_name: previous_session_name.as_deref(),
603 snapshot_image: plan.snapshot_image.as_deref(),
604 resource_limits: &plan.resource_limits,
605 command: &plan.command,
606 message: &plan.message,
607 },
608 output_format,
609 )),
610 error: None,
611 exit_code: None,
612 }
613}
614
615#[cfg(test)]
616#[path = "execution_resume_cases.rs"]
617mod tests;