1use std::path::PathBuf;
19
20use chrono::Utc;
21use serde_json::{json, Value};
22
23use crate::docker_cleanup::{
24 build_docker_runtime_args, docker_command, docker_networks,
25 get_docker_container_cleanup_policy, start_detached_docker_completion_watcher_with,
26 DockerWatcherOptions,
27};
28use crate::docker_resource_limits::{build_resource_limits_status_line, normalize_resource_limits};
29use crate::execution_control::{CommandRunner, SystemCommandRunner};
30use crate::execution_store::{ExecutionRecord, ExecutionStatus, ExecutionStore};
31use crate::isolation::isolation_log::append_log_file;
32use crate::isolation::{run_isolated, IsolationOptions, IsolationResult};
33use crate::output_blocks::escape_for_links_notation;
34use crate::session_probe::{probe_session, SessionProbe, SessionState};
35
36#[derive(Debug, Clone, Copy, PartialEq, Eq)]
38pub enum ResumeMode {
39 DockerStart,
40 DockerSnapshot,
41 Relaunch,
42}
43
44impl ResumeMode {
45 pub fn as_str(self) -> &'static str {
46 match self {
47 ResumeMode::DockerStart => "docker-start",
48 ResumeMode::DockerSnapshot => "docker-snapshot",
49 ResumeMode::Relaunch => "relaunch",
50 }
51 }
52}
53
54#[derive(Debug, Clone, PartialEq, Eq)]
56pub struct ResumeStep {
57 pub command: String,
58 pub args: Vec<String>,
59 pub description: String,
60}
61
62#[derive(Debug, Clone)]
64pub struct ResumePlan {
65 pub mode: ResumeMode,
66 pub backend: String,
67 pub session_name: String,
68 pub new_session_name: Option<String>,
70 pub snapshot_image: Option<String>,
71 pub command: String,
72 pub attempt: u64,
73 pub resource_limits: Vec<String>,
76 pub steps: Vec<ResumeStep>,
77 pub launch_options: Option<IsolationOptions>,
78 pub message: String,
79}
80
81pub struct ExecutionResumeResult {
83 pub success: bool,
84 pub output: Option<String>,
85 pub error: Option<String>,
86 pub exit_code: Option<i32>,
89}
90
91pub trait ResumeHooks {
96 fn start_watcher(&self, session_name: &str, record: &ExecutionRecord);
98 fn attach_watcher(&self, session_name: &str, record: &ExecutionRecord) -> Result<(), String> {
100 self.start_watcher(session_name, record);
101 Ok(())
102 }
103
104 fn relaunch(&self, backend: &str, command: &str, options: &IsolationOptions)
106 -> IsolationResult;
107
108 fn reconcile(&self, record: &ExecutionRecord) -> ExecutionRecord;
110}
111
112#[derive(Debug, Default)]
114pub struct SystemResumeHooks;
115
116impl ResumeHooks for SystemResumeHooks {
117 fn start_watcher(&self, session_name: &str, record: &ExecutionRecord) {
118 let _ = self.attach_watcher(session_name, record);
119 }
120
121 fn attach_watcher(&self, session_name: &str, record: &ExecutionRecord) -> Result<(), String> {
122 let log_path = (!record.log_path.is_empty()).then(|| PathBuf::from(&record.log_path));
123 start_detached_docker_completion_watcher_with(
124 session_name,
125 get_docker_container_cleanup_policy(&build_launch_options(record)),
126 log_path.as_ref(),
127 Some(record.uuid.as_str()),
130 &DockerWatcherOptions {
131 since: record
132 .attempt
133 .as_ref()
134 .map(|attempt| attempt.started_at.clone()),
135 attempt_number: record.attempt.as_ref().map(|attempt| attempt.number),
136 recover_on_kill: keeps_kill_recovery(session_name, record),
137 },
138 )
139 }
140
141 fn relaunch(
142 &self,
143 backend: &str,
144 command: &str,
145 options: &IsolationOptions,
146 ) -> IsolationResult {
147 run_isolated(backend, command, options)
148 }
149
150 fn reconcile(&self, record: &ExecutionRecord) -> ExecutionRecord {
151 crate::status_formatter::enrich_detached_status(record)
152 }
153}
154
155pub fn keeps_kill_recovery(session_name: &str, record: &ExecutionRecord) -> bool {
159 record.attempt.as_ref().is_none_or(|attempt| {
160 matches!(
161 attempt.mode.as_str(),
162 "docker-start" | "automatic-recovery" | "relaunch"
163 )
164 }) && record_option(record, "sessionName") == Some(session_name)
165 && record
166 .options
167 .get("onKillResume")
168 .is_some_and(|value| value.as_u64().unwrap_or(0) > 0)
169}
170
171fn record_option<'a>(record: &'a ExecutionRecord, key: &str) -> Option<&'a str> {
172 record.options.get(key).and_then(|value| value.as_str())
173}
174
175fn record_flag(record: &ExecutionRecord, key: &str) -> bool {
176 record
177 .options
178 .get(key)
179 .and_then(|value| value.as_bool())
180 .unwrap_or(false)
181}
182
183fn record_strings(record: &ExecutionRecord, key: &str) -> Vec<String> {
184 record
185 .options
186 .get(key)
187 .and_then(|value| value.as_array())
188 .map(|items| {
189 items
190 .iter()
191 .filter_map(|item| item.as_str().map(str::to_string))
192 .collect()
193 })
194 .unwrap_or_default()
195}
196
197pub fn build_snapshot_image_name(session_name: &str, attempt: u64) -> String {
201 let mut sanitized = String::new();
202 let mut pending_dash = false;
203 for ch in session_name.to_lowercase().chars() {
204 if ch.is_ascii_lowercase() || ch.is_ascii_digit() || ch == '.' || ch == '_' || ch == '-' {
205 if pending_dash {
206 sanitized.push('-');
207 pending_dash = false;
208 }
209 sanitized.push(ch);
210 } else if !sanitized.is_empty() || !pending_dash {
211 pending_dash = true;
212 }
213 }
214 if pending_dash {
215 sanitized.push('-');
216 }
217 let sanitized = sanitized.trim_start_matches(['-', '.']).to_string();
218 let sanitized = if sanitized.is_empty() {
219 "session".to_string()
220 } else {
221 sanitized
222 };
223 format!("start-command-resume/{}:{}", sanitized, attempt)
224}
225
226pub fn build_resumed_session_name(session_name: &str, attempt: u64) -> String {
228 format!("{}-resume-{}", session_name, attempt)
229}
230
231pub fn build_launch_options(record: &ExecutionRecord) -> IsolationOptions {
234 let networks = record_strings(record, "networks");
235 IsolationOptions {
236 session: record_option(record, "sessionName").map(str::to_string),
237 labels: record_strings(record, "labels"),
238 uuid: Some(record.uuid.clone()),
239 root_session: Some(
240 record_option(record, "rootSession")
241 .or_else(|| {
242 record
243 .options
244 .get("sessionNameHistory")
245 .and_then(Value::as_array)
246 .and_then(|history| history.first())
247 .and_then(Value::as_str)
248 })
249 .unwrap_or_else(|| record_option(record, "sessionName").unwrap_or("session"))
250 .to_string(),
251 ),
252 resume_count: record
253 .options
254 .get("resumeCount")
255 .and_then(Value::as_u64)
256 .unwrap_or(0) as u32
257 + 1,
258 image: record_option(record, "image").map(str::to_string),
259 volumes: record_strings(record, "volumes"),
260 mounts: record_strings(record, "mounts"),
261 env: record_strings(record, "env"),
262 privileged: record_flag(record, "privileged"),
263 network: networks.first().cloned(),
264 networks,
265 network_aliases: record_strings(record, "networkAliases"),
266 resource_limits: normalize_resource_limits(record.options.get("resourceLimits")),
268 cpu_penalty_config: record
269 .options
270 .get("cpuPenaltyConfig")
271 .and_then(|v| serde_json::from_value(v.clone()).ok()),
272 on_kill_resume: record
273 .options
274 .get("onKillResume")
275 .and_then(Value::as_u64)
276 .map(|n| n as u32),
277 recovery_command: record_option(record, "recoveryCommand").map(str::to_string),
278 endpoint: record_option(record, "endpoint").map(str::to_string),
279 detached: true,
280 user: record_option(record, "user").map(str::to_string),
281 keep_alive: record_flag(record, "keepAlive"),
282 auto_remove_docker_container: record_flag(record, "autoRemoveDockerContainer"),
283 always_cleanup_container: record_flag(record, "alwaysCleanupContainer"),
284 keep_container: record_flag(record, "keepContainer"),
285 keep_container_on_fail: record_flag(record, "keepContainerOnFail"),
286 shell: record_option(record, "shell").unwrap_or("auto").to_string(),
287 log_path: (!record.log_path.is_empty()).then(|| PathBuf::from(&record.log_path)),
289 execution_id: Some(record.uuid.clone()),
292 defer_completion_watcher: true,
293 }
294}
295
296fn build_snapshot_start_steps(
304 docker: &str,
305 container_args: Vec<String>,
306 extra_networks: &[String],
307 new_session_name: &str,
308 snapshot_image: &str,
309) -> Vec<ResumeStep> {
310 if extra_networks.is_empty() {
311 let mut args = vec!["run".to_string(), "-d".to_string()];
312 args.extend(container_args);
313 return vec![ResumeStep {
314 command: docker.to_string(),
315 args,
316 description: format!("Run the new command in {}", new_session_name),
317 }];
318 }
319
320 let mut create_args = vec!["create".to_string()];
321 create_args.extend(container_args);
322 let mut steps = vec![ResumeStep {
323 command: docker.to_string(),
324 args: create_args,
325 description: format!("Create {} from {}", new_session_name, snapshot_image),
326 }];
327 for network in extra_networks {
328 steps.push(ResumeStep {
329 command: docker.to_string(),
330 args: vec![
331 "network".to_string(),
332 "connect".to_string(),
333 network.clone(),
334 new_session_name.to_string(),
335 ],
336 description: format!("Connect {} to network {}", new_session_name, network),
337 });
338 }
339 steps.push(ResumeStep {
340 command: docker.to_string(),
341 args: vec!["start".to_string(), new_session_name.to_string()],
342 description: format!("Run the new command in {}", new_session_name),
343 });
344 steps
345}
346
347fn docker_snapshot_plan(
348 record: &ExecutionRecord,
349 backend: &str,
350 session_name: &str,
351 command: &str,
352 attempt: u64,
353 live_resource_limits: Option<Vec<String>>,
354) -> ResumePlan {
355 let snapshot_image = build_snapshot_image_name(session_name, attempt);
356 let new_session_name = build_resumed_session_name(session_name, attempt);
357 let docker = docker_command().to_string_lossy().to_string();
358
359 let mut launch_options = build_launch_options(record);
360 launch_options.session = Some(new_session_name.clone());
361 launch_options.resume_count = attempt as u32;
362 launch_options.labels =
363 crate::isolation_metadata::docker_attribution_labels(&launch_options, &new_session_name);
364 if let Some(limits) = live_resource_limits {
367 launch_options.resource_limits = limits;
368 }
369 let mut container_args = vec!["--name".to_string(), new_session_name.clone()];
370 container_args.extend([
371 "--label".into(),
372 format!(
373 "{}={}",
374 crate::docker_snapshot_safety::SNAPSHOT_IMAGE_LABEL,
375 snapshot_image
376 ),
377 ]);
378 if let Some(user) = record_option(record, "user") {
379 container_args.push("--user".to_string());
380 container_args.push(user.to_string());
381 }
382 container_args.extend(
383 build_docker_runtime_args(&launch_options)
384 .into_iter()
385 .map(str::to_string),
386 );
387 container_args.push(snapshot_image.clone());
388 container_args.extend(crate::docker_command_handoff::build_command_handoff_args(
389 &["sh".into(), "-c".into(), command.into()],
390 &crate::docker_command_handoff::command_handoff_path(&new_session_name),
391 ));
392
393 let mut steps = vec![ResumeStep {
394 command: docker.clone(),
395 args: vec![
396 "commit".to_string(),
397 session_name.to_string(),
398 snapshot_image.clone(),
399 ],
400 description: format!("Snapshot container {} as {}", session_name, snapshot_image),
401 }];
402 steps.extend(build_snapshot_start_steps(
403 &docker,
404 container_args,
405 &docker_networks(&launch_options)
406 .into_iter()
407 .skip(1)
408 .map(str::to_string)
409 .collect::<Vec<_>>(),
410 &new_session_name,
411 &snapshot_image,
412 ));
413
414 ResumePlan {
415 mode: ResumeMode::DockerSnapshot,
416 backend: backend.to_string(),
417 session_name: session_name.to_string(),
418 new_session_name: Some(new_session_name.clone()),
419 snapshot_image: Some(snapshot_image.clone()),
420 command: command.to_string(),
421 attempt,
422 resource_limits: launch_options.resource_limits.clone(),
423 steps,
424 launch_options: None,
425 message: format!(
426 "Resumed session in new container {} from snapshot of {}",
427 new_session_name, session_name
428 ),
429 }
430}
431
432pub fn build_resume_plan(
434 record: &ExecutionRecord,
435 new_command: Option<&str>,
436 probe: &SessionProbe,
437) -> Result<ResumePlan, String> {
438 build_resume_plan_with_limits(record, new_command, probe, None)
439}
440
441pub fn build_resume_plan_with_limits(
444 record: &ExecutionRecord,
445 new_command: Option<&str>,
446 probe: &SessionProbe,
447 live_resource_limits: Option<Vec<String>>,
448) -> Result<ResumePlan, String> {
449 let session_name = record_option(record, "sessionName")
450 .ok_or_else(|| "Execution record does not contain an isolation session name.".to_string())?
451 .to_string();
452
453 if record_option(record, "isolationMode") != Some("detached") {
454 return Err("Only detached isolated executions can be resumed.".to_string());
455 }
456
457 if probe.alive {
458 return Err(format!(
459 "Session \"{}\" is still running. Use `$ --attach {}` to re-enter it, or `$ --stop {}` first.",
460 session_name, record.uuid, record.uuid
461 ));
462 }
463
464 let backend = record_option(record, "isolated")
465 .unwrap_or("unknown")
466 .to_string();
467 let command = new_command.unwrap_or(record.command.as_str()).to_string();
468 if command.is_empty() {
469 return Err(format!(
470 "Execution \"{}\" has no stored command to resume.",
471 record.uuid
472 ));
473 }
474
475 let attempt = record
476 .options
477 .get("resumeCount")
478 .and_then(|value| value.as_u64())
479 .unwrap_or(0)
480 + 1;
481
482 if backend == "docker" && probe.state == SessionState::Stopped {
483 return Ok(match new_command {
484 None | Some(_)
485 if !record_flag(record, "forceSnapshot")
486 && (new_command.is_none() || record_flag(record, "commandHandoff")) =>
487 {
488 ResumePlan {
489 mode: ResumeMode::DockerStart,
490 backend,
491 new_session_name: None,
492 snapshot_image: None,
493 command,
494 attempt,
495 resource_limits: Vec::new(),
496 steps: vec![ResumeStep {
497 command: docker_command().to_string_lossy().to_string(),
498 args: vec!["start".to_string(), session_name.clone()],
499 description: format!("Start stopped container {}", session_name),
500 }],
501 launch_options: None,
502 message: if new_command.is_some() {
503 format!("Resumed new command in the same detached docker container: {} (no filesystem snapshot)", session_name)
504 } else {
505 format!("Resumed detached docker container: {}", session_name)
506 },
507 session_name,
508 }
509 }
510 Some(new_command) => docker_snapshot_plan(
511 record,
512 &backend,
513 &session_name,
514 new_command,
515 attempt,
516 live_resource_limits,
517 ),
518 None => docker_snapshot_plan(
519 record,
520 &backend,
521 &session_name,
522 &command,
523 attempt,
524 live_resource_limits,
525 ),
526 });
527 }
528
529 Ok(ResumePlan {
530 mode: ResumeMode::Relaunch,
531 message: format!("Relaunched {} session: {}", backend, session_name),
532 backend,
533 new_session_name: None,
534 snapshot_image: None,
535 command,
536 attempt,
537 resource_limits: Vec::new(),
538 steps: Vec::new(),
539 launch_options: Some(build_launch_options(record)),
540 session_name,
541 })
542}
543
544pub struct ResumeResultFields<'a> {
546 pub identifier: &'a str,
547 pub uuid: &'a str,
548 pub mode: ResumeMode,
549 pub backend: &'a str,
550 pub session_name: &'a str,
551 pub previous_session_name: Option<&'a str>,
552 pub snapshot_image: Option<&'a str>,
553 pub resource_limits: &'a [String],
555 pub command: &'a str,
556 pub message: &'a str,
557}
558
559pub fn format_resume_result_as_links_notation(result: &ResumeResultFields) -> String {
561 let mut lines = vec![
562 "executionResume".to_string(),
563 format!(
564 " identifier {}",
565 escape_for_links_notation(result.identifier)
566 ),
567 format!(" uuid {}", escape_for_links_notation(result.uuid)),
568 format!(" mode {}", escape_for_links_notation(result.mode.as_str())),
569 format!(" backend {}", escape_for_links_notation(result.backend)),
570 format!(
571 " sessionName {}",
572 escape_for_links_notation(result.session_name)
573 ),
574 ];
575 if let Some(previous) = result.previous_session_name {
576 lines.push(format!(
577 " previousSessionName {}",
578 escape_for_links_notation(previous)
579 ));
580 }
581 if let Some(image) = result.snapshot_image {
582 lines.push(format!(
583 " snapshotImage {}",
584 escape_for_links_notation(image)
585 ));
586 }
587 if !result.resource_limits.is_empty() {
588 lines.push(format!(
589 " resourceLimits {}",
590 escape_for_links_notation(&result.resource_limits.join(" "))
591 ));
592 }
593 lines.push(format!(
594 " command {}",
595 escape_for_links_notation(result.command)
596 ));
597 lines.push(format!(
598 " message {}",
599 escape_for_links_notation(result.message)
600 ));
601 lines.join("\n")
602}
603
604pub fn format_resume_result(result: &ResumeResultFields, output_format: Option<&str>) -> String {
606 match output_format {
607 Some("json") => {
608 let mut value = json!({
609 "identifier": result.identifier,
610 "uuid": result.uuid,
611 "mode": result.mode.as_str(),
612 "backend": result.backend,
613 "sessionName": result.session_name,
614 "previousSessionName": result.previous_session_name,
615 "snapshotImage": result.snapshot_image,
616 "resourceLimits": (!result.resource_limits.is_empty())
617 .then_some(result.resource_limits),
618 "command": result.command,
619 "message": result.message,
620 });
621 if let Value::Object(ref mut map) = value {
622 map.retain(|_, entry| !entry.is_null());
623 }
624 serde_json::to_string_pretty(&value).unwrap_or_else(|_| "{}".to_string())
625 }
626 Some("text") => {
627 let mut lines = vec![
628 format!("Resume Mode: {}", result.mode.as_str()),
629 format!("UUID: {}", result.uuid),
630 format!("Backend: {}", result.backend),
631 format!("Session Name: {}", result.session_name),
632 format!("Command: {}", result.command),
633 ];
634 if !result.resource_limits.is_empty() {
635 lines.push(format!(
636 "Resource Limits: {}",
637 result.resource_limits.join(" ")
638 ));
639 }
640 lines.push(result.message.to_string());
641 lines.join("\n")
642 }
643 _ => format_resume_result_as_links_notation(result),
644 }
645}
646
647pub fn apply_resume_to_record(
650 record: &mut ExecutionRecord,
651 plan: &ResumePlan,
652 container_id: Option<&str>,
653) {
654 let attempt = crate::execution_attempt::create_attempt(
655 record,
656 plan.mode.as_str(),
657 active_session_name(plan),
658 );
659 crate::execution_attempt::archive_attempt(record);
660 record.attempt = Some(attempt);
661 record
662 .options
663 .insert("resumeCount".to_string(), json!(plan.attempt));
664 record.options.insert(
665 "resumedAt".to_string(),
666 json!(record.attempt.as_ref().unwrap().started_at),
667 );
668 record.options.remove("stopRequestedAt");
670 record.options.remove("exitEvidence");
671 if !plan.resource_limits.is_empty() {
672 record
673 .options
674 .insert("resourceLimits".to_string(), json!(plan.resource_limits));
675 }
676
677 if !record.options.contains_key("rootSession") {
678 let root = record
679 .options
680 .get("sessionNameHistory")
681 .and_then(Value::as_array)
682 .and_then(|history| history.first())
683 .and_then(Value::as_str)
684 .or_else(|| record_option(record, "sessionName"))
685 .unwrap_or("session")
686 .to_string();
687 record.options.insert("rootSession".into(), json!(root));
688 }
689 if let Some(new_session_name) = &plan.new_session_name {
690 let mut history = record
691 .options
692 .get("sessionNameHistory")
693 .and_then(|value| value.as_array())
694 .cloned()
695 .unwrap_or_default();
696 if let Some(previous) = record.options.get("sessionName").cloned() {
697 history.push(previous);
698 }
699 record
700 .options
701 .insert("sessionNameHistory".to_string(), Value::Array(history));
702 record
703 .options
704 .insert("sessionName".to_string(), json!(new_session_name));
705 }
706 if let Some(snapshot_image) = &plan.snapshot_image {
707 record
708 .options
709 .insert("image".to_string(), json!(snapshot_image));
710 record
711 .options
712 .insert("snapshotImage".into(), json!(snapshot_image));
713 record.options.insert("commandHandoff".into(), json!(true));
714 }
715 if plan.mode == ResumeMode::Relaunch && plan.backend == "docker" {
716 record.options.insert("commandHandoff".into(), json!(true));
717 }
718 if let Some(container_id) = container_id {
719 record
720 .options
721 .insert("containerId".to_string(), json!(container_id));
722 }
723
724 record.command = plan.command.clone();
725 record.status = ExecutionStatus::Executing;
726 record.exit_code = None;
727 record.end_time = None;
728 record.exit_reason = None;
729 record.oom_killed = None;
730}
731
732fn active_session_name(plan: &ResumePlan) -> &str {
733 plan.new_session_name
734 .as_deref()
735 .unwrap_or(&plan.session_name)
736}
737
738#[path = "execution_resume_flow.rs"]
739mod resume_flow;
740pub use resume_flow::{
741 resume_execution, resume_execution_with, resume_execution_with_options,
742 resume_execution_with_resources,
743};