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 mut launch_options = plan.launch_options.clone().unwrap_or_default();
276 launch_options.attempt_number = attempt.number;
277 let launch_result = hooks.relaunch(&plan.backend, &plan.command, &launch_options);
278 if !launch_result.success {
279 let _ = store.save(&previous);
280 crate::execution_attempt::append_lifecycle(
281 &lifecycle_record,
282 "launch-failed",
283 json!({ "error": launch_result.message }),
284 );
285 return ExecutionResumeResult {
286 success: false,
287 output: None,
288 error: Some(format!(
289 "Failed to relaunch {} session \"{}\": {}",
290 plan.backend, plan.session_name, launch_result.message
291 )),
292 exit_code: None,
293 };
294 }
295 container_id = launch_result.container_id;
296 } else {
297 let mut snapshot_created = false;
298 let mut derived_created = false;
299 let handoff = plan.mode == ResumeMode::DockerStart
300 && command.is_some()
301 && record_flag(&previous, "commandHandoff");
302 let settings = SnapshotSettings::from_overrides(overrides);
303 let notify = |line: &str| eprintln!("{}", line);
304 let reporter = SnapshotReporter {
305 log_path: &record.log_path,
306 notify: ¬ify,
307 };
308 let mut failure_exit_code = None;
309 let launch_result = (|| -> Result<(), String> {
310 let mut exit_code_of = |failure: SnapshotFailure| {
311 failure_exit_code = failure.exit_code;
312 failure.message
313 };
314 let _snapshot_lock = if plan.mode == ResumeMode::DockerSnapshot {
315 Some(
316 acquire_snapshot_lock_with(&SnapshotLockRequest {
317 lock_path: settings.lock_path.clone(),
318 holder: Some(&plan.session_name),
319 wait_seconds: settings.wait_seconds,
320 notify: ¬ify,
321 })
322 .map_err(&mut exit_code_of)?,
323 )
324 } else {
325 None
326 };
327 if plan.mode == ResumeMode::DockerSnapshot {
328 if let Some(prepare) = &settings.prepare {
329 run_snapshot_prepare(&plan.session_name, prepare, runner, &reporter);
330 }
331 let options = PreflightOptions {
332 size_hint: settings.size_hint,
333 disk_wait_seconds: settings.disk_wait_seconds,
334 disk_poll_seconds: settings.disk_poll_seconds,
335 notify: ¬ify,
336 ..Default::default()
337 };
338 let message =
339 preflight_snapshot_guarded(&plan.session_name, runner, &options, &|root| {
340 df_free_bytes(runner, root)
341 })
342 .map_err(&mut exit_code_of)?;
343 plan.message.push_str(&format!("\n{}", message));
344 if !record.log_path.is_empty() {
345 append_log_file(&PathBuf::from(&record.log_path), &format!("{}\n", message));
346 }
347 }
348 if handoff {
349 let copied = crate::docker_command_handoff::write_command_handoff(
350 &plan.session_name,
351 &plan.command,
352 record_option(&previous, "shell").unwrap_or("auto"),
353 record_flag(&previous, "keepAlive"),
354 runner,
355 )?;
356 if !copied.success {
357 return Err(if copied.stderr.trim().is_empty() {
358 copied.error.unwrap_or_else(|| {
359 "Could not copy replacement command into the stopped container".into()
360 })
361 } else {
362 copied.stderr.trim().into()
363 });
364 }
365 }
366 if plan.mode == ResumeMode::DockerStart && record_flag(&previous, "commandHandoff") {
367 let env = crate::execution_env::build_execution_env(
369 Some(&record.uuid),
370 Some(&plan.session_name),
371 attempt.number,
372 );
373 let _ = crate::docker_command_handoff::write_execution_env_handoff(
374 &plan.session_name,
375 &env,
376 runner,
377 );
378 }
379 for step in &plan.steps {
380 let result = if step.args[0] == "commit" {
381 commit_with_progress(
382 runner,
383 &step.command,
384 &step.args,
385 &plan.session_name,
386 &reporter,
387 settings
388 .progress_interval_seconds
389 .unwrap_or_else(progress_interval_seconds),
390 )
391 } else {
392 runner.run(&step.command, &step.args)
393 };
394 if !result.success {
395 let detail = if !result.stderr.trim().is_empty() {
396 result.stderr.trim().to_string()
397 } else {
398 result.error.clone().unwrap_or_else(|| {
399 format!(
400 "exit code {}",
401 result
402 .status
403 .map(|code| code.to_string())
404 .unwrap_or_else(|| "unknown".to_string())
405 )
406 })
407 };
408 return Err(detail);
409 }
410 if step.args[0] == "commit" {
411 snapshot_created = true;
412 }
413 if step.args[0] == "create" {
414 derived_created = true;
415 }
416 let stdout = result.stdout.trim().to_string();
417 if !stdout.is_empty() && matches!(step.args[0].as_str(), "run" | "create") {
418 container_id = Some(stdout);
419 }
420 }
421 if plan.mode == ResumeMode::DockerSnapshot
422 && overrides.get("removeOriginal").and_then(Value::as_bool) == Some(true)
423 && settings.remove_original_after <= 0.0
424 {
425 let running = runner.run(
426 &docker_command().to_string_lossy(),
427 &[
428 "inspect".into(),
429 "-f".into(),
430 "{{.State.Running}}".into(),
431 active_session_name(&plan).into(),
432 ],
433 );
434 let note = if running.success && running.stdout.trim() == "true" {
435 let removed = runner.run(
436 &docker_command().to_string_lossy(),
437 &["rm".into(), plan.session_name.clone()],
438 );
439 if removed.success {
440 format!("Stopped original container removed: {}", plan.session_name)
441 } else {
442 format!("Original container retained: {}", removed.stderr.trim())
443 }
444 } else {
445 "Original container retained: the snapshot container is not running.".into()
446 };
447 plan.message.push_str(&format!("\n{}", note));
448 if !record.log_path.is_empty() {
449 append_log_file(&PathBuf::from(&record.log_path), &format!("{}\n", note));
450 }
451 }
452 Ok(())
453 })();
454 if let Err(detail) = launch_result {
455 if derived_created {
456 runner.run(
457 &docker_command().to_string_lossy(),
458 &["rm".into(), active_session_name(&plan).into()],
459 );
460 }
461 if snapshot_created {
462 runner.run(
463 &docker_command().to_string_lossy(),
464 &["rmi".into(), plan.snapshot_image.clone().unwrap()],
465 );
466 }
467 if handoff {
468 let _ = crate::docker_command_handoff::write_command_handoff(
469 &plan.session_name,
470 &previous.command,
471 record_option(&previous, "shell").unwrap_or("auto"),
472 record_flag(&previous, "keepAlive"),
473 runner,
474 );
475 }
476 let _ = store.save(&previous);
477 crate::execution_attempt::append_lifecycle(
478 &lifecycle_record,
479 "launch-failed",
480 json!({ "error": detail }),
481 );
482 return ExecutionResumeResult {
483 success: false,
484 output: None,
485 error: Some(format!(
486 "Failed to resume {} session \"{}\": {}",
487 plan.backend, plan.session_name, detail
488 )),
489 exit_code: failure_exit_code,
490 };
491 }
492
493 if let Some(line) = build_resource_limits_status_line(&plan.resource_limits) {
494 if !record.log_path.is_empty() {
495 append_log_file(&PathBuf::from(&record.log_path), &format!("{}\n", line));
496 }
497 }
498 }
499
500 let previous_session_name = plan
501 .new_session_name
502 .as_ref()
503 .map(|_| plan.session_name.clone());
504 if let Some(id) = container_id.as_deref() {
505 record.options.insert("containerId".into(), json!(id));
506 }
507 record.options.insert("launchPending".into(), json!(false));
508 attempt.launch_accepted_at = Some(Utc::now().to_rfc3339());
509 record
510 .options
511 .insert("resumedAt".into(), json!(attempt.started_at));
512 record.attempt = Some(attempt.clone());
513 if let Err(error) = store.save(&record) {
514 if plan.backend == "docker" {
515 let _ = hooks.attach_watcher(active_session_name(&plan), &record);
516 }
517 let stopped = plan.backend == "docker"
518 && runner
519 .run(
520 &docker_command().to_string_lossy(),
521 &["stop".into(), active_session_name(&plan).into()],
522 )
523 .success;
524 if stopped {
525 record.status = ExecutionStatus::Executed;
526 record.exit_code = Some(-1);
527 record.end_time = Some(Utc::now().to_rfc3339());
528 record.options.insert("launchPending".into(), json!(false));
529 let _ = store.save(&record);
530 }
531 return ExecutionResumeResult {
532 success: false,
533 output: None,
534 error: Some(
535 json!({"code": "LAUNCH_PERSISTENCE_FAILED", "uuid": record.uuid,
536 "containerName": active_session_name(&plan), "running": !stopped,
537 "error": error})
538 .to_string(),
539 ),
540 exit_code: None,
541 };
542 }
543 crate::execution_attempt::append_lifecycle(&record, "launch-accepted", json!({}));
544 if plan.backend == "docker" {
545 let attachment = hooks.attach_watcher(active_session_name(&plan), &record);
546 let fields = match &attachment {
547 Ok(()) => json!({"watcherAttachedAt": Utc::now().to_rfc3339()}),
548 Err(error) => json!({"watcherError": error}),
549 };
550 match store.patch_attempt(&record.uuid, attempt.number, fields) {
551 Ok(Some(current)) => crate::execution_attempt::append_lifecycle(
552 ¤t,
553 if attachment.is_ok() {
554 "watcher-attached"
555 } else {
556 "watcher-attachment-failed"
557 },
558 json!({"error": attachment.as_ref().err()}),
559 ),
560 Ok(None) => {}
561 Err(error) => {
562 return ExecutionResumeResult {
563 success: false,
564 output: None,
565 error: Some(error),
566 exit_code: None,
567 }
568 }
569 }
570 if let Err(error) = attachment {
571 return ExecutionResumeResult {
572 success: false,
573 output: None,
574 error: Some(format!(
575 "Launch accepted, but completion watcher attachment failed: {}",
576 error
577 )),
578 exit_code: None,
579 };
580 }
581 }
582
583 let remove_after = SnapshotSettings::from_overrides(overrides).remove_original_after;
584 if plan.mode == ResumeMode::DockerSnapshot
585 && overrides.get("removeOriginal").and_then(Value::as_bool) == Some(true)
586 && remove_after > 0.0
587 {
588 let notify = |line: &str| eprintln!("{}", line);
589 let note = remove_original_after(
590 &plan.session_name,
591 active_session_name(&plan),
592 remove_after,
593 runner,
594 &SnapshotReporter {
595 log_path: &record.log_path,
596 notify: ¬ify,
597 },
598 &std::thread::sleep,
599 );
600 plan.message.push_str(&format!("\n{}", note));
601 }
602
603 let session_name = record_option(&record, "sessionName")
604 .unwrap_or(&plan.session_name)
605 .to_string();
606
607 ExecutionResumeResult {
608 success: true,
609 output: Some(format_resume_result(
610 &ResumeResultFields {
611 identifier,
612 uuid: &record.uuid,
613 mode: plan.mode,
614 backend: &plan.backend,
615 session_name: &session_name,
616 previous_session_name: previous_session_name.as_deref(),
617 snapshot_image: plan.snapshot_image.as_deref(),
618 resource_limits: &plan.resource_limits,
619 command: &plan.command,
620 message: &plan.message,
621 },
622 output_format,
623 )),
624 error: None,
625 exit_code: None,
626 }
627}
628
629#[cfg(test)]
630#[path = "execution_resume_cases.rs"]
631mod tests;