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 attempt_number: crate::execution_attempt::next_attempt_number(record),
295 }
296}
297
298fn build_snapshot_start_steps(
306 docker: &str,
307 container_args: Vec<String>,
308 extra_networks: &[String],
309 new_session_name: &str,
310 snapshot_image: &str,
311) -> Vec<ResumeStep> {
312 if extra_networks.is_empty() {
313 let mut args = vec!["run".to_string(), "-d".to_string()];
314 args.extend(container_args);
315 return vec![ResumeStep {
316 command: docker.to_string(),
317 args,
318 description: format!("Run the new command in {}", new_session_name),
319 }];
320 }
321
322 let mut create_args = vec!["create".to_string()];
323 create_args.extend(container_args);
324 let mut steps = vec![ResumeStep {
325 command: docker.to_string(),
326 args: create_args,
327 description: format!("Create {} from {}", new_session_name, snapshot_image),
328 }];
329 for network in extra_networks {
330 steps.push(ResumeStep {
331 command: docker.to_string(),
332 args: vec![
333 "network".to_string(),
334 "connect".to_string(),
335 network.clone(),
336 new_session_name.to_string(),
337 ],
338 description: format!("Connect {} to network {}", new_session_name, network),
339 });
340 }
341 steps.push(ResumeStep {
342 command: docker.to_string(),
343 args: vec!["start".to_string(), new_session_name.to_string()],
344 description: format!("Run the new command in {}", new_session_name),
345 });
346 steps
347}
348
349fn docker_snapshot_plan(
350 record: &ExecutionRecord,
351 backend: &str,
352 session_name: &str,
353 command: &str,
354 attempt: u64,
355 live_resource_limits: Option<Vec<String>>,
356) -> ResumePlan {
357 let snapshot_image = build_snapshot_image_name(session_name, attempt);
358 let new_session_name = build_resumed_session_name(session_name, attempt);
359 let docker = docker_command().to_string_lossy().to_string();
360
361 let mut launch_options = build_launch_options(record);
362 launch_options.session = Some(new_session_name.clone());
363 launch_options.resume_count = attempt as u32;
364 launch_options.labels =
365 crate::isolation_metadata::docker_attribution_labels(&launch_options, &new_session_name);
366 if let Some(limits) = live_resource_limits {
369 launch_options.resource_limits = limits;
370 }
371 let mut container_args = vec!["--name".to_string(), new_session_name.clone()];
372 container_args.extend([
373 "--label".into(),
374 format!(
375 "{}={}",
376 crate::docker_snapshot_safety::SNAPSHOT_IMAGE_LABEL,
377 snapshot_image
378 ),
379 ]);
380 if let Some(user) = record_option(record, "user") {
381 container_args.push("--user".to_string());
382 container_args.push(user.to_string());
383 }
384 container_args.extend(build_docker_runtime_args(&launch_options));
385 container_args.push(snapshot_image.clone());
386 container_args.extend(crate::docker_command_handoff::build_command_handoff_args(
387 &["sh".into(), "-c".into(), command.into()],
388 &crate::docker_command_handoff::command_handoff_path(&new_session_name),
389 ));
390
391 let mut steps = vec![ResumeStep {
392 command: docker.clone(),
393 args: vec![
394 "commit".to_string(),
395 session_name.to_string(),
396 snapshot_image.clone(),
397 ],
398 description: format!("Snapshot container {} as {}", session_name, snapshot_image),
399 }];
400 steps.extend(build_snapshot_start_steps(
401 &docker,
402 container_args,
403 &docker_networks(&launch_options)
404 .into_iter()
405 .skip(1)
406 .map(str::to_string)
407 .collect::<Vec<_>>(),
408 &new_session_name,
409 &snapshot_image,
410 ));
411
412 ResumePlan {
413 mode: ResumeMode::DockerSnapshot,
414 backend: backend.to_string(),
415 session_name: session_name.to_string(),
416 new_session_name: Some(new_session_name.clone()),
417 snapshot_image: Some(snapshot_image.clone()),
418 command: command.to_string(),
419 attempt,
420 resource_limits: launch_options.resource_limits.clone(),
421 steps,
422 launch_options: None,
423 message: format!(
424 "Resumed session in new container {} from snapshot of {}",
425 new_session_name, session_name
426 ),
427 }
428}
429
430pub fn build_resume_plan(
432 record: &ExecutionRecord,
433 new_command: Option<&str>,
434 probe: &SessionProbe,
435) -> Result<ResumePlan, String> {
436 build_resume_plan_with_limits(record, new_command, probe, None)
437}
438
439pub fn build_resume_plan_with_limits(
442 record: &ExecutionRecord,
443 new_command: Option<&str>,
444 probe: &SessionProbe,
445 live_resource_limits: Option<Vec<String>>,
446) -> Result<ResumePlan, String> {
447 let session_name = record_option(record, "sessionName")
448 .ok_or_else(|| "Execution record does not contain an isolation session name.".to_string())?
449 .to_string();
450
451 if record_option(record, "isolationMode") != Some("detached") {
452 return Err("Only detached isolated executions can be resumed.".to_string());
453 }
454
455 if probe.alive {
456 return Err(format!(
457 "Session \"{}\" is still running. Use `$ --attach {}` to re-enter it, or `$ --stop {}` first.",
458 session_name, record.uuid, record.uuid
459 ));
460 }
461
462 let backend = record_option(record, "isolated")
463 .unwrap_or("unknown")
464 .to_string();
465 let command = new_command.unwrap_or(record.command.as_str()).to_string();
466 if command.is_empty() {
467 return Err(format!(
468 "Execution \"{}\" has no stored command to resume.",
469 record.uuid
470 ));
471 }
472
473 let attempt = record
474 .options
475 .get("resumeCount")
476 .and_then(|value| value.as_u64())
477 .unwrap_or(0)
478 + 1;
479
480 if backend == "docker" && probe.state == SessionState::Stopped {
481 return Ok(match new_command {
482 None | Some(_)
483 if !record_flag(record, "forceSnapshot")
484 && (new_command.is_none() || record_flag(record, "commandHandoff")) =>
485 {
486 ResumePlan {
487 mode: ResumeMode::DockerStart,
488 backend,
489 new_session_name: None,
490 snapshot_image: None,
491 command,
492 attempt,
493 resource_limits: Vec::new(),
494 steps: vec![ResumeStep {
495 command: docker_command().to_string_lossy().to_string(),
496 args: vec!["start".to_string(), session_name.clone()],
497 description: format!("Start stopped container {}", session_name),
498 }],
499 launch_options: None,
500 message: if new_command.is_some() {
501 format!("Resumed new command in the same detached docker container: {} (no filesystem snapshot)", session_name)
502 } else {
503 format!("Resumed detached docker container: {}", session_name)
504 },
505 session_name,
506 }
507 }
508 Some(new_command) => docker_snapshot_plan(
509 record,
510 &backend,
511 &session_name,
512 new_command,
513 attempt,
514 live_resource_limits,
515 ),
516 None => docker_snapshot_plan(
517 record,
518 &backend,
519 &session_name,
520 &command,
521 attempt,
522 live_resource_limits,
523 ),
524 });
525 }
526
527 Ok(ResumePlan {
528 mode: ResumeMode::Relaunch,
529 message: format!("Relaunched {} session: {}", backend, session_name),
530 backend,
531 new_session_name: None,
532 snapshot_image: None,
533 command,
534 attempt,
535 resource_limits: Vec::new(),
536 steps: Vec::new(),
537 launch_options: Some(build_launch_options(record)),
538 session_name,
539 })
540}
541
542pub struct ResumeResultFields<'a> {
544 pub identifier: &'a str,
545 pub uuid: &'a str,
546 pub mode: ResumeMode,
547 pub backend: &'a str,
548 pub session_name: &'a str,
549 pub previous_session_name: Option<&'a str>,
550 pub snapshot_image: Option<&'a str>,
551 pub resource_limits: &'a [String],
553 pub command: &'a str,
554 pub message: &'a str,
555}
556
557pub fn format_resume_result_as_links_notation(result: &ResumeResultFields) -> String {
559 let mut lines = vec![
560 "executionResume".to_string(),
561 format!(
562 " identifier {}",
563 escape_for_links_notation(result.identifier)
564 ),
565 format!(" uuid {}", escape_for_links_notation(result.uuid)),
566 format!(" mode {}", escape_for_links_notation(result.mode.as_str())),
567 format!(" backend {}", escape_for_links_notation(result.backend)),
568 format!(
569 " sessionName {}",
570 escape_for_links_notation(result.session_name)
571 ),
572 ];
573 if let Some(previous) = result.previous_session_name {
574 lines.push(format!(
575 " previousSessionName {}",
576 escape_for_links_notation(previous)
577 ));
578 }
579 if let Some(image) = result.snapshot_image {
580 lines.push(format!(
581 " snapshotImage {}",
582 escape_for_links_notation(image)
583 ));
584 }
585 if !result.resource_limits.is_empty() {
586 lines.push(format!(
587 " resourceLimits {}",
588 escape_for_links_notation(&result.resource_limits.join(" "))
589 ));
590 }
591 lines.push(format!(
592 " command {}",
593 escape_for_links_notation(result.command)
594 ));
595 lines.push(format!(
596 " message {}",
597 escape_for_links_notation(result.message)
598 ));
599 lines.join("\n")
600}
601
602pub fn format_resume_result(result: &ResumeResultFields, output_format: Option<&str>) -> String {
604 match output_format {
605 Some("json") => {
606 let mut value = json!({
607 "identifier": result.identifier,
608 "uuid": result.uuid,
609 "mode": result.mode.as_str(),
610 "backend": result.backend,
611 "sessionName": result.session_name,
612 "previousSessionName": result.previous_session_name,
613 "snapshotImage": result.snapshot_image,
614 "resourceLimits": (!result.resource_limits.is_empty())
615 .then_some(result.resource_limits),
616 "command": result.command,
617 "message": result.message,
618 });
619 if let Value::Object(ref mut map) = value {
620 map.retain(|_, entry| !entry.is_null());
621 }
622 serde_json::to_string_pretty(&value).unwrap_or_else(|_| "{}".to_string())
623 }
624 Some("text") => {
625 let mut lines = vec![
626 format!("Resume Mode: {}", result.mode.as_str()),
627 format!("UUID: {}", result.uuid),
628 format!("Backend: {}", result.backend),
629 format!("Session Name: {}", result.session_name),
630 format!("Command: {}", result.command),
631 ];
632 if !result.resource_limits.is_empty() {
633 lines.push(format!(
634 "Resource Limits: {}",
635 result.resource_limits.join(" ")
636 ));
637 }
638 lines.push(result.message.to_string());
639 lines.join("\n")
640 }
641 _ => format_resume_result_as_links_notation(result),
642 }
643}
644
645pub fn apply_resume_to_record(
648 record: &mut ExecutionRecord,
649 plan: &ResumePlan,
650 container_id: Option<&str>,
651) {
652 let attempt = crate::execution_attempt::create_attempt(
653 record,
654 plan.mode.as_str(),
655 active_session_name(plan),
656 );
657 crate::execution_attempt::archive_attempt(record);
658 record.attempt = Some(attempt);
659 record
660 .options
661 .insert("resumeCount".to_string(), json!(plan.attempt));
662 record.options.insert(
663 "resumedAt".to_string(),
664 json!(record.attempt.as_ref().unwrap().started_at),
665 );
666 record.options.remove("stopRequestedAt");
668 record.options.remove("exitEvidence");
669 if !plan.resource_limits.is_empty() {
670 record
671 .options
672 .insert("resourceLimits".to_string(), json!(plan.resource_limits));
673 }
674
675 if !record.options.contains_key("rootSession") {
676 let root = record
677 .options
678 .get("sessionNameHistory")
679 .and_then(Value::as_array)
680 .and_then(|history| history.first())
681 .and_then(Value::as_str)
682 .or_else(|| record_option(record, "sessionName"))
683 .unwrap_or("session")
684 .to_string();
685 record.options.insert("rootSession".into(), json!(root));
686 }
687 if let Some(new_session_name) = &plan.new_session_name {
688 let mut history = record
689 .options
690 .get("sessionNameHistory")
691 .and_then(|value| value.as_array())
692 .cloned()
693 .unwrap_or_default();
694 if let Some(previous) = record.options.get("sessionName").cloned() {
695 history.push(previous);
696 }
697 record
698 .options
699 .insert("sessionNameHistory".to_string(), Value::Array(history));
700 record
701 .options
702 .insert("sessionName".to_string(), json!(new_session_name));
703 }
704 if let Some(snapshot_image) = &plan.snapshot_image {
705 record
706 .options
707 .insert("image".to_string(), json!(snapshot_image));
708 record
709 .options
710 .insert("snapshotImage".into(), json!(snapshot_image));
711 record.options.insert("commandHandoff".into(), json!(true));
712 }
713 if plan.mode == ResumeMode::Relaunch && plan.backend == "docker" {
714 record.options.insert("commandHandoff".into(), json!(true));
715 }
716 if let Some(container_id) = container_id {
717 record
718 .options
719 .insert("containerId".to_string(), json!(container_id));
720 }
721
722 record.command = plan.command.clone();
723 record.status = ExecutionStatus::Executing;
724 record.exit_code = None;
725 record.end_time = None;
726 record.exit_reason = None;
727 record.oom_killed = None;
728}
729
730fn active_session_name(plan: &ResumePlan) -> &str {
731 plan.new_session_name
732 .as_deref()
733 .unwrap_or(&plan.session_name)
734}
735
736#[path = "execution_resume_flow.rs"]
737mod resume_flow;
738pub use resume_flow::{
739 resume_execution, resume_execution_with, resume_execution_with_options,
740 resume_execution_with_resources,
741};