#[cfg(test)]
mod tests;
use anyhow::{Context, Result, bail, ensure};
use sha2::{Digest, Sha256};
use super::{Controller, SessionResumeOptions, now};
fn mutation_holds() -> &'static std::sync::Mutex<std::collections::BTreeSet<String>> {
static HOLDS: std::sync::OnceLock<std::sync::Mutex<std::collections::BTreeSet<String>>> =
std::sync::OnceLock::new();
HOLDS.get_or_init(Default::default)
}
pub fn move_owns_session(session_id: &str) -> bool {
move_has_pending_queue(session_id)
|| mutation_holds()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.contains(session_id)
}
fn queue_holds() -> &'static std::sync::Mutex<std::collections::BTreeSet<String>> {
static HOLDS: std::sync::OnceLock<std::sync::Mutex<std::collections::BTreeSet<String>>> =
std::sync::OnceLock::new();
HOLDS.get_or_init(Default::default)
}
pub fn move_has_pending_queue(session_id: &str) -> bool {
queue_holds()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.contains(session_id)
}
pub fn release_move_queue_hold(session_id: &str) {
queue_holds()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.remove(session_id);
}
pub fn restore_move_queue_hold(operation: &MoveOperation) {
let mut holds = queue_holds()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if operation.queue_admission_started && !operation.queue_admission_finished {
holds.insert(operation.selection.session_id.clone());
} else {
holds.remove(&operation.selection.session_id);
}
}
pub struct MoveMutationGuard(String);
impl MoveMutationGuard {
pub fn reserve(session_id: &str) -> Result<Self> {
ensure!(
mutation_holds()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.insert(session_id.to_owned()),
"session already has a move owner"
);
Ok(Self(session_id.to_owned()))
}
}
impl Drop for MoveMutationGuard {
fn drop(&mut self) {
mutation_holds()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.remove(&self.0);
}
}
pub(crate) fn move_refuses_command(session_id: &str, command: &RelayCommand) -> bool {
move_owns_session(session_id)
&& matches!(
command,
RelayCommand::Prompt { .. }
| RelayCommand::SetConfig { .. }
| RelayCommand::SetSessionMode { .. }
| RelayCommand::RunUserShell { .. }
| RelayCommand::CancelUserShell { .. }
| RelayCommand::Cancel
| RelayCommand::RemoveQueuedPrompt { .. }
| RelayCommand::ClearQueuedPrompts
)
}
use crate::hel_session_manager::{SessionManagerControl, StandaloneSession, new_command_id};
use hel::hel_archive::{CanonicalQueuedCommandKind, verify_archive_streaming};
use hel::hel_state::{MoveOperation, MovePhase, ResumeQueueDisposition, SessionState};
pub use hel::hel_state::{MoveOutcome, MovePreparation, MoveSelection, MoveSessionRequest};
use hel::hel_targets::{CommandExecutor, ProvisionStage, ProvisionStageGuard};
use hel::hel_worker::RelayCommand;
fn digest(value: &impl serde::Serialize) -> Result<String> {
Ok(format!("{:x}", Sha256::digest(serde_json::to_vec(value)?)))
}
struct MovePhaseTimer<'a> {
session_id: &'a str,
phase: &'static str,
started: std::time::Instant,
}
impl<'a> MovePhaseTimer<'a> {
fn new(session_id: &'a str, phase: &'static str) -> Self {
Self {
session_id,
phase,
started: std::time::Instant::now(),
}
}
}
impl Drop for MovePhaseTimer<'_> {
fn drop(&mut self) {
tracing::info!(
session_id = self.session_id,
phase = self.phase,
elapsed_ms = self.started.elapsed().as_millis() as u64,
"move phase finished"
);
}
}
impl Controller {
fn validate_move_destination_paths(
&self,
source: &hel::hel_state::SessionRecord,
target_id: &str,
executor: &(impl CommandExecutor + Sync),
) -> Result<()> {
use super::worktree::ResumePlan;
match super::worktree::resume_compatibility(source, &self.config, target_id)
.map_err(anyhow::Error::msg)?
{
ResumePlan::RawToWorkspace => {
super::worktree::plan_raw_to_workspace(source, &self.config, executor)?;
}
ResumePlan::WorkspaceToRaw => {
self.plan_workspace_to_raw(source, target_id, executor)?;
}
ResumePlan::InPlace if source.managed_worktree.is_none() => {
if let Some(path) = &source.project_directory {
self.validate_project_directory(target_id, path, executor)?;
}
}
ResumePlan::InPlace => {}
}
Ok(())
}
fn move_confirmation(
&self,
selection: &MoveSelection,
) -> Result<(bool, Vec<hel::hel_state::MaterializedQueuedPrompt>, String)> {
let source = self
.state
.sessions
.get(&selection.session_id)
.context("unknown move session")?;
let (mut active, mut queued) = hel::hel_database::move_pending_work(&source.id)?;
if let Some(operation) = hel::hel_database::load_move_operation(&source.id)?
&& operation.queue_admission_started
&& !operation.queue_admission_finished
{
ensure!(
operation.selection == *selection,
"queue admission is incomplete on the live destination; retry that move before selecting another destination"
);
let checkpoint = operation
.checkpoint
.as_ref()
.context("retained queue checkpoint is missing")?;
let verified = verify_archive_streaming(&checkpoint.archive_path)?;
ensure!(
verified.archive_sha256 == checkpoint.sha256
&& verified.manifest.session.id == source.id,
"retained move checkpoint verification failed"
);
queued = verified
.canonical_session
.queued_prompts
.into_iter()
.map(|entry| hel::hel_state::MaterializedQueuedPrompt {
command_id: entry.command_id,
kind: match entry.kind {
CanonicalQueuedCommandKind::Prompt => {
hel::hel_state::QueuedCommandKind::Prompt
}
CanonicalQueuedCommandKind::SetConfig { key, value } => {
hel::hel_state::QueuedCommandKind::SetConfig { key, value }
}
},
content: entry.content,
queued_at_ms: entry.queued_at_ms,
})
.collect();
active = false;
}
let fingerprint = digest(&(
&source.last_profile,
&source.target_template_id,
&source.target,
&source.native_session_id,
&source.resource_allocation,
&source.additional_mounts,
&source.container_cpus,
&source.container_memory,
selection,
self.move_configuration_fingerprint(selection)?,
&queued,
))?;
Ok((active, queued, fingerprint))
}
fn move_configuration_fingerprint(&self, selection: &MoveSelection) -> Result<String> {
let profile = self
.config
.profiles
.get(
selection
.profile_id
.as_deref()
.context("move profile is unresolved")?,
)
.context("move profile no longer exists")?;
let target = self
.config
.targets
.get(
selection
.target_template_id
.as_deref()
.context("move target is unresolved")?,
)
.context("move target no longer exists")?;
let source = self
.state
.sessions
.get(&selection.session_id)
.context("unknown move session")?;
digest(&(
profile,
target,
&self.config.bundles,
self.config.targets.get(&source.target_template_id),
self.config.profiles.get(&source.last_profile),
))
}
pub async fn prepare_move_session_controlled(
&self,
mut selection: MoveSelection,
executor: &(impl CommandExecutor + Sync),
) -> Result<MovePreparation> {
ensure!(
selection.profile_id.is_some() || selection.target_template_id.is_some(),
"move requires a target or profile selection"
);
let source = self
.state
.sessions
.get(&selection.session_id)
.context("unknown session")?;
let previous = hel::hel_database::load_move_operation(&source.id)?;
let retry = previous.as_ref().is_some_and(|op| {
!matches!(op.phase, MovePhase::Completed) && source.checkpoint.is_some()
});
ensure!(
matches!(
source.state,
SessionState::Running | SessionState::Disconnected
) || retry,
"only active sessions can move; use Resume for a stopped or lost session"
);
selection
.profile_id
.get_or_insert_with(|| source.last_profile.clone());
selection
.target_template_id
.get_or_insert_with(|| source.target_template_id.clone());
selection
.additional_mounts
.get_or_insert_with(|| source.additional_mounts.clone());
ensure!(
!selection.clear_resource_allocation || selection.resource_allocation.is_none(),
"select resource allocation or explicitly clear it, not both"
);
if !selection.clear_resource_allocation {
selection.resource_allocation = selection
.resource_allocation
.or_else(|| source.resource_allocation.clone());
}
let profile_id = selection.profile_id.as_deref().unwrap();
let target_id = selection.target_template_id.as_deref().unwrap();
let profile = self
.config
.profiles
.get(profile_id)
.context("unknown destination profile")?;
let target = self
.config
.targets
.get(target_id)
.context("unknown destination target")?;
self.validate_muse_resume_destination(source, profile.kind, target_id)?;
super::worktree::resume_compatibility(source, &self.config, target_id)
.map_err(anyhow::Error::msg)?;
super::backend::validate_resource_allocation(
target,
selection.resource_allocation.as_ref(),
)?;
let mounts = selection.additional_mounts.as_deref().unwrap_or_default();
ensure!(
profile.kind != hel::hel_config::HarnessKind::Muse || mounts.is_empty(),
"Muse Code ACP supports one workspace root; attached directories are unsupported"
);
ensure!(
mounts.is_empty() || hel::hel_config::mount_history_host(target).is_some(),
"attached resources are unsupported for this target; select compatible resources explicitly"
);
hel::hel_targets::validate_additional_mounts(mounts)?;
for mount in mounts {
self.validate_mount_source(target_id, &mount.source, executor)?;
}
self.validate_move_destination_paths(source, target_id, executor)?;
ensure!(
profile.home.is_dir(),
"destination profile home is unavailable; configure the profile before moving"
);
super::worker_binary::preflight_worker_binary(target)?;
super::backend::preflight_target(target, executor)?;
let source_harness = previous
.as_ref()
.filter(|operation| {
!operation.queue_admission_started && operation.phase != MovePhase::Completed
})
.and_then(|operation| operation.recovery_session.as_ref())
.map_or(source.harness_kind, |record| record.harness_kind);
let cross_harness = profile.kind != source_harness;
if cross_harness {
let cancel = tokio_util::sync::CancellationToken::new();
let resolve =
crate::hel_utility_llm::UtilityLlmRuntime::shared().resolve(&self.config, &cancel);
tokio::pin!(resolve);
loop {
tokio::select! {
result = &mut resolve => { result.context("cross-harness move needs an available utility model")?; break; }
_ = tokio::time::sleep(std::time::Duration::from_millis(50)) => {
if executor.cancellation_requested() { cancel.cancel(); bail!("move preparation cancelled"); }
}
}
}
}
let (active, mut queued_commands, fingerprint) = self.move_confirmation(&selection)?;
for entry in &mut queued_commands {
for content in &mut entry.content {
if content.get("type").and_then(serde_json::Value::as_str) == Some("image") {
let mime = content
.get("mimeType")
.and_then(serde_json::Value::as_str)
.unwrap_or("image");
*content = serde_json::json!({"type":"text", "text":format!("[Image attachment: {mime}]")});
}
}
}
let operation_id = previous
.as_ref()
.filter(|operation| {
operation.selection == selection
&& operation.phase != MovePhase::Completed
&& operation.checkpoint.is_some()
})
.map(|operation| operation.operation_id.clone())
.unwrap_or(new_command_id("move")?);
Ok(MovePreparation {
selection,
source_profile_id: source.last_profile.clone(),
source_target_template_id: source.target_template_id.clone(),
cross_harness,
active,
queued_commands,
fingerprint,
operation_id,
})
}
pub async fn move_session_managed_controlled(
&mut self,
request: MoveSessionRequest,
executor: &(impl CommandExecutor + Sync),
manager: &SessionManagerControl,
) -> Result<MoveOutcome> {
let prepared = &request.preparation;
let id = prepared.selection.session_id.clone();
let started = std::time::Instant::now();
executor.notify_notice("Checking destination");
let checked = {
let _checking_destination =
ProvisionStageGuard::new(executor, ProvisionStage::Verifying);
let mut checked = self
.prepare_move_session_controlled(prepared.selection.clone(), executor)
.await?;
if matches!(
self.state.sessions[&id].state,
SessionState::Running | SessionState::Disconnected
) {
let handle = manager
.wait_for_session(&id, std::time::Duration::from_secs(5))
.await?;
handle.sync_now().await?;
let (active, queue, fingerprint) = self.move_confirmation(&checked.selection)?;
checked.active = active
|| handle.view().snapshot.as_ref().is_some_and(|snapshot| {
let mut operational = snapshot.operational.clone();
operational.queued_prompts.clear();
operational.checkpoint_barrier = None;
!operational.is_quiet()
});
checked.queued_commands = queue;
checked.fingerprint = fingerprint;
}
checked
};
ensure!(
checked.fingerprint == prepared.fingerprint,
"session, pending work, or destination configuration changed; prepare and confirm Move again"
);
let queue = request.queue.unwrap_or(ResumeQueueDisposition::Discard);
let source = self.state.sessions[&id].clone();
let old_operation = hel::hel_database::load_move_operation(&id)?;
let retry = old_operation.filter(|op| {
op.selection == prepared.selection
&& op.phase != MovePhase::Completed
&& op.checkpoint.is_some()
});
if retry.is_none()
&& source.state == SessionState::Running
&& source.last_profile == checked.selection.profile_id.as_deref().unwrap()
&& source.target_template_id == checked.selection.target_template_id.as_deref().unwrap()
&& Some(&source.additional_mounts) == checked.selection.additional_mounts.as_ref()
&& source.resource_allocation == checked.selection.resource_allocation
&& (!checked.selection.clear_resource_allocation
|| (source.container_cpus.is_none() && source.container_memory.is_none()))
{
return Ok(outcome(
&prepared.operation_id,
&prepared.selection,
"unchanged",
None,
None,
));
}
ensure!(
!checked.active || request.acknowledge_interruption,
"active work will be interrupted; confirm Move again with interruption acknowledgement"
);
ensure!(
checked.queued_commands.is_empty() || request.queue.is_some(),
"pending work requires an explicit queue choice: discard or start"
);
let timestamp = now();
let mut operation = match retry {
Some(mut op) => {
ensure!(
!op.queue_admission_started || op.queue == queue,
"queued work may already have run; retry with the original queue choice on the same destination"
);
hel::hel_database::clear_move_cancellation_for_retry(&id)?;
op.configuration_fingerprint =
self.move_configuration_fingerprint(&checked.selection)?;
op.queue = queue;
op.cancellation_requested = false;
op.error = None;
op
}
None => MoveOperation {
operation_id: prepared.operation_id.clone(),
selection: checked.selection.clone(),
source_profile_id: source.last_profile.clone(),
source_target_template_id: source.target_template_id.clone(),
source_target: source.target.clone(),
source_native_session_id: source.native_session_id.clone(),
source_additional_mounts: source.additional_mounts.clone(),
source_resource_allocation: source.resource_allocation.clone(),
destination_target: None,
destination_native_session_id: None,
destination_store_id: None,
configuration_fingerprint: self
.move_configuration_fingerprint(&checked.selection)?,
checkpoint: None,
recovery_session: None,
queue,
phase: MovePhase::Preparing,
queue_admission_started: false,
queue_admission_finished: false,
cancellation_requested: false,
created_at: timestamp.clone(),
updated_at: timestamp,
error: None,
},
};
hel::hel_database::save_move_operation(&operation)?;
tracing::info!(
session_id = id,
phase = "preflight",
elapsed_ms = started.elapsed().as_millis() as u64,
"move phase completed"
);
let result = self
.execute_move(&mut operation, Some(&checked), executor, manager)
.await;
self.finish_move_result(&mut operation, result, executor)
}
fn finish_move_result(
&self,
operation: &mut MoveOperation,
result: Result<()>,
executor: &impl CommandExecutor,
) -> Result<MoveOutcome> {
let (status, error, recovery) = match result {
Ok(()) => {
operation.phase = MovePhase::Completed;
("completed", None, None)
}
Err(error) => {
let cancelled =
executor.cancellation_requested() || operation.cancellation_requested;
operation.phase = if cancelled {
MovePhase::Cancelled
} else {
MovePhase::Failed
};
operation.cancellation_requested = cancelled;
let recovery = if operation.queue_admission_started {
"Destination is live; retry queue admission on this same destination. Already accepted work may have effects."
} else if self
.state
.sessions
.get(&operation.selection.session_id)
.is_some_and(|s| s.state == SessionState::Stopped)
{
"Session is stopped with a verified checkpoint. Retry move or Resume with previous settings."
} else {
"Source or partial destination is retained. Retry move after resolving the reported error."
};
let error = format!("{error:#}");
operation.error = Some(error.clone());
(
if cancelled { "cancelled" } else { "failed" },
Some(error),
Some(recovery.to_owned()),
)
}
};
operation.updated_at = now();
hel::hel_database::save_move_operation(operation)?;
Ok(outcome(
&operation.operation_id,
&operation.selection,
status,
error,
recovery,
))
}
pub async fn recover_move_managed_controlled(
&mut self,
mut operation: MoveOperation,
executor: &(impl CommandExecutor + Sync),
manager: &SessionManagerControl,
) -> Result<MoveOutcome> {
let id = operation.selection.session_id.clone();
let result = async {
let session = self.state.sessions.get(&id).context("move session is missing")?.clone();
if operation.queue_admission_started {
ensure!(!operation.cancellation_requested, "Move was cancelled; destination retained without further queue admission");
return self.admit_move_queue(&mut operation, executor).await;
}
if matches!(session.state, SessionState::Closing | SessionState::Destroying) {
let cleanup = hel::hel_targets::CancellableProcessExecutor::with_timeout(std::time::Duration::from_secs(15));
self.recover_move_source_stop(&mut operation, &cleanup, manager).await?;
operation.checkpoint = self.state.sessions[&id].checkpoint.clone();
if self.state.sessions[&id].state == SessionState::Stopped && self.state.sessions[&id].target.is_some() {
self.cleanup_stopped_target(&id, &cleanup)?;
}
bail!("Move source stop recovered; no destination work was started. Retry Move or Resume with previous settings");
}
match operation.phase {
MovePhase::Preparing => bail!("Move preparation was interrupted; source retained. Prepare Move again."),
MovePhase::ResumingDestination if session.state == SessionState::Running => {
ensure!(Some(&session.last_profile) == operation.selection.profile_id.as_ref()
&& Some(&session.target_template_id) == operation.selection.target_template_id.as_ref(),
"ready destination does not match the move intent");
operation.destination_target = session.target.clone();
operation.destination_native_session_id = session.native_session_id.clone();
operation.queue_admission_started = true;
operation.phase = MovePhase::StartingQueue;
hel::hel_database::save_move_operation(&operation)?;
restore_move_queue_hold(&operation);
ensure!(!operation.cancellation_requested, "Move was cancelled; ready destination retained");
self.admit_move_queue(&mut operation, executor).await
}
MovePhase::ResumingDestination => {
let previous = operation.recovery_session.as_ref().context("move lacks its stopped recovery identity; retain resources for inspection")?;
let error = self.rollback_failed_resume(&id, previous, false,
anyhow::anyhow!("destination restoration was interrupted; checkpoint retained for an explicit retry"), executor)?;
Err(error)
}
MovePhase::ClosingSource => {
if matches!(session.state, SessionState::Closing | SessionState::Destroying) {
self.recover_move_source_stop(&mut operation, executor, manager).await?;
}
operation.checkpoint = self.state.sessions[&id].checkpoint.clone();
if self.state.sessions[&id].state == SessionState::Stopped && self.state.sessions[&id].target.is_some() {
self.cleanup_stopped_target(&id, executor)?;
}
bail!("source stop was recovered; verified checkpoint retained. Retry move or Resume with previous settings")
}
_ => bail!("Move requires an explicit retry after the daemon restarted"),
}
}.await;
self.finish_move_result(&mut operation, result, executor)
}
async fn recover_move_source_stop(
&mut self,
operation: &mut MoveOperation,
executor: &(impl CommandExecutor + Sync),
manager: &SessionManagerControl,
) -> Result<()> {
let id = operation.selection.session_id.clone();
if self.state.sessions[&id].state == SessionState::Closing {
let handle = manager
.wait_for_session(&id, std::time::Duration::from_secs(5))
.await?;
let mut lease = handle.lease_connection().await?;
let execution = lease.connection_mut().sync().await?.operational.execution;
lease.release();
if matches!(
execution,
hel::hel_worker::RelayExecutionState::Idle
| hel::hel_worker::RelayExecutionState::Running
) {
if operation.cancellation_requested {
let record = self.state.sessions.get_mut(&id).unwrap();
record.state = SessionState::Running;
record.updated_at = now();
record.last_error =
Some("Move was cancelled before the source was sealed".into());
hel::hel_database::save_lifecycle_session(record)?;
} else {
self.close_session_for_move(&id, executor, manager, operation, None)
.await?;
}
return Ok(());
}
}
self.recover_interrupted_close_managed(&id, executor, manager)
.await?;
Ok(())
}
async fn execute_move(
&mut self,
operation: &mut MoveOperation,
preparation: Option<&MovePreparation>,
executor: &(impl CommandExecutor + Sync),
manager: &SessionManagerControl,
) -> Result<()> {
let id = operation.selection.session_id.clone();
ensure!(
!executor.cancellation_requested(),
"move cancelled before source interruption"
);
if !operation.queue_admission_started {
if self.state.sessions[&id].state == SessionState::Error
&& let Some(previous) = operation.recovery_session.as_ref()
{
let failure = self.rollback_failed_resume(
&id,
previous,
false,
anyhow::anyhow!("clean up the partial Move destination before retry"),
executor,
)?;
ensure!(
self.state.sessions[&id].state == SessionState::Stopped,
"{failure:#}"
);
}
let state = self.state.sessions[&id].state;
if matches!(state, SessionState::Closing | SessionState::Destroying) {
self.recover_move_source_stop(operation, executor, manager)
.await?;
operation.checkpoint = self.state.sessions[&id].checkpoint.clone();
} else if matches!(state, SessionState::Running | SessionState::Disconnected)
&& operation.destination_target.is_none()
{
executor.notify_notice("Stopping source");
let _timing = MovePhaseTimer::new(&id, "checkpoint and source stop");
operation.phase = MovePhase::ClosingSource;
operation.updated_at = now();
hel::hel_database::save_move_operation(operation)?;
self.close_session_for_move(&id, executor, manager, operation, preparation)
.await?;
}
if self.state.sessions[&id].state == SessionState::Stopped
&& self.state.sessions[&id].target.is_some()
{
executor.notify_notice("Cleaning up source");
let _timing = MovePhaseTimer::new(&id, "source storage cleanup");
self.cleanup_stopped_target(&id, executor)?;
}
operation.checkpoint = operation
.checkpoint
.clone()
.or_else(|| self.state.sessions[&id].checkpoint.clone());
ensure!(
operation.checkpoint.is_some(),
"move has no verified checkpoint"
);
ensure!(
!executor.cancellation_requested(),
"move cancelled after source teardown; session is stopped"
);
operation.phase = MovePhase::ResumingDestination;
operation.recovery_session = Some(self.state.sessions[&id].clone());
operation.updated_at = now();
hel::hel_database::save_move_operation(operation)?;
executor.notify_notice("Preparing destination");
if operation.selection.clear_resource_allocation {
let session = self.state.sessions.get_mut(&id).unwrap();
session.resource_allocation = None;
session.container_cpus = None;
session.container_memory = None;
hel::hel_database::save_session(session)?;
}
self.resume_session_controlled(
&id,
operation.selection.profile_id.as_deref().unwrap(),
operation.selection.target_template_id.as_deref().unwrap(),
SessionResumeOptions {
additional_mounts: operation.selection.additional_mounts.clone(),
resource_allocation: operation.selection.resource_allocation.clone(),
discard_queue: true,
},
executor,
)
.await?;
let destination = &self.state.sessions[&id];
operation.destination_target = destination.target.clone();
operation.destination_native_session_id = destination.native_session_id.clone();
operation.phase = MovePhase::StartingQueue;
operation.queue_admission_started = true;
operation.updated_at = now();
hel::hel_database::save_move_operation(operation)?;
}
restore_move_queue_hold(operation);
self.admit_move_queue(operation, executor).await
}
async fn admit_move_queue(
&self,
operation: &mut MoveOperation,
executor: &(impl CommandExecutor + Sync),
) -> Result<()> {
let timing_id = operation.selection.session_id.clone();
let _timing = MovePhaseTimer::new(&timing_id, "queue admission");
let id = &operation.selection.session_id;
let mut relay = {
let _checking_destination =
ProvisionStageGuard::new(executor, ProvisionStage::Verifying);
let destination = &self.state.sessions[id];
ensure!(
destination.state == SessionState::Running
&& destination.target == operation.destination_target
&& destination.native_session_id == operation.destination_native_session_id,
"cannot prove the same ready destination; refusing to replay potentially executed work"
);
let spec = self.reconnect_command(id)?;
let relay = StandaloneSession::connect_command(&spec, id).await?;
let store_id = relay.snapshot().operational.store_id.context("destination worker does not expose its durable store identity; upgrade the worker before admitting queued work")?;
if let Some(expected) = &operation.destination_store_id {
ensure!(
*expected == store_id,
"destination relay storage was replaced; refusing to replay potentially executed work"
);
} else {
operation.destination_store_id = Some(store_id);
hel::hel_database::save_move_operation(operation)?;
}
ensure!(
relay.snapshot().operational.native_session_id
== operation.destination_native_session_id,
"destination relay native identity changed; refusing queue replay"
);
relay
};
if operation.queue == ResumeQueueDisposition::Start && !operation.queue_admission_finished {
let _starting_queue = ProvisionStageGuard::new(executor, ProvisionStage::Starting);
executor.notify_notice("Starting queued work");
let checkpoint = operation
.checkpoint
.as_ref()
.context("move queue archive is missing")?;
let verified = verify_archive_streaming(&checkpoint.archive_path)?;
ensure!(
verified.archive_sha256 == checkpoint.sha256 && verified.manifest.session.id == *id,
"move queue checkpoint verification failed"
);
for queued in verified.canonical_session.queued_prompts {
ensure!(
!executor.cancellation_requested(),
"move cancelled during queue admission; destination retained"
);
let command = match queued.kind {
CanonicalQueuedCommandKind::Prompt => RelayCommand::Prompt {
prompt: queued
.content
.into_iter()
.map(serde_json::from_value)
.collect::<serde_json::Result<_>>()?,
},
CanonicalQueuedCommandKind::SetConfig { key, value } => {
RelayCommand::SetConfig { key, value }
}
};
relay.submit(queued.command_id, command).await?;
}
}
operation.queue_admission_finished = true;
hel::hel_database::save_move_operation(operation)?;
restore_move_queue_hold(operation);
relay.submit(format!("{}-notice", operation.operation_id), RelayCommand::RecordNotice {
text: format!("Moved from {} / {} to {} / {} in a fresh environment. {} The interrupted prompt was not replayed.",
operation.source_profile_id, operation.source_target_template_id,
operation.selection.profile_id.as_deref().unwrap(), operation.selection.target_template_id.as_deref().unwrap(),
if operation.queue == ResumeQueueDisposition::Discard { "Queued work was discarded; ready and idle." } else { "Queued work was accepted." }),
}).await?;
Ok(())
}
pub(super) fn validate_move_checkpoint(
&self,
operation: &MoveOperation,
preparation: Option<&MovePreparation>,
executor: &(impl CommandExecutor + Sync),
) -> Result<()> {
let _verifying = ProvisionStageGuard::new(executor, ProvisionStage::Verifying);
let current = Controller {
config: hel::hel_config::HelConfig::load()?,
state: self.state.clone(),
};
ensure!(
current.move_configuration_fingerprint(&operation.selection)?
== operation.configuration_fingerprint,
"destination configuration changed during move"
);
let id = &operation.selection.session_id;
current.validate_move_destination_paths(
&self.state.sessions[id],
operation.selection.target_template_id.as_deref().unwrap(),
executor,
)?;
if let super::ResumeRepositorySourcePreflight::RepositoryMoved(mismatch) = self
.preflight_resume_repository_sources(
id,
operation.selection.target_template_id.as_deref().unwrap(),
executor,
)?
{
bail!(
"destination repository source is missing checkpoint commit {}; source retained",
mismatch.missing_commit
);
}
if let Some(prepared) = preparation {
let checkpoint = self.state.sessions[id]
.checkpoint
.as_ref()
.context("no move checkpoint")?;
let verified = verify_archive_streaming(&checkpoint.archive_path)?;
let actual: Vec<_> = verified
.canonical_session
.queued_prompts
.iter()
.map(|p| p.command_id.as_str())
.collect();
let expected: Vec<_> = prepared
.queued_commands
.iter()
.map(|p| p.command_id.as_str())
.collect();
ensure!(
actual == expected,
"pending queue changed before checkpoint capture; source retained, confirm Move again"
);
}
Ok(())
}
}
fn outcome(
operation_id: &str,
selection: &MoveSelection,
status: &str,
error: Option<String>,
recovery: Option<String>,
) -> MoveOutcome {
MoveOutcome {
operation_id: operation_id.into(),
session_id: selection.session_id.clone(),
profile_id: selection.profile_id.clone().unwrap_or_default(),
target_template_id: selection.target_template_id.clone().unwrap_or_default(),
outcome: status.into(),
error,
recovery,
}
}