use std::io::{Read as _, Write as _};
use std::os::unix::fs::PermissionsExt as _;
use std::path::{Path, PathBuf};
use std::process::Command;
use chrono::Utc;
use serde::Serialize;
use serde_json::{json, Value};
use octl_core::report::sanitize_report_advisory;
use octl_core::{
append_and_apply_event, read_all_events, read_manifest_opt, read_node_opt, AdvisoryWarning,
MergeTxn, Node, NodeId, RunLock,
};
use crate::run::merge_recovery;
use crate::supervise::cleanup::git_bin;
use crate::error::{CliError, ExitKind};
use crate::output::{self, OutputFormat, OutputSpec};
use crate::run::dto::SupervisorView;
use crate::run::{from_core, parse_node_id, reattach, require_nonempty, run_paths_from_cli_arg};
use crate::supervise::cleanup;
const MERGE_SH: &str = include_str!("../../scripts/merge.sh");
const DEFAULT_NODE_ID: &str = "n-0001";
const MAX_REPORT_BYTES: u64 = 1024 * 1024;
const MERGE_SH_LOCK_TIMEOUT_EXIT: i32 = 75;
const MERGE_SH_CAS_MISMATCH_EXIT: i32 = 76;
pub struct Args<'a> {
pub run_id: String,
pub source: Option<String>,
pub node_id: Option<String>,
pub report_file: Option<PathBuf>,
pub dry_run: bool,
pub spec: &'a OutputSpec,
pub warnings: &'a [String],
}
#[derive(Serialize)]
struct MergePayload<'a> {
run_id: &'a str,
node_id: &'a str,
branch: &'a str,
#[serde(skip_serializing_if = "Option::is_none")]
source: Option<&'a str>,
merged: bool,
#[serde(skip_serializing_if = "Option::is_none")]
report_seq: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
supervisor: Option<ConsumerOutcome>,
#[serde(skip_serializing_if = "<[_]>::is_empty")]
report_advisory_warnings: &'a [AdvisoryWarning],
#[serde(skip_serializing_if = "Option::is_none")]
dry_run: Option<bool>,
}
#[derive(Serialize, Clone)]
#[serde(tag = "state", rename_all = "kebab-case")]
pub(crate) enum ConsumerOutcome {
Alive,
Terminal,
NotSupervised,
Reattached { pid: Option<u32> },
Deferred { recovery_command: String },
}
pub(crate) struct MergeOutcome {
pub run_id: String,
pub node_id: String,
pub branch: String,
pub source: Option<String>,
pub merged: bool,
pub report_seq: Option<u64>,
pub supervisor: Option<ConsumerOutcome>,
pub dry_run: bool,
pub warnings: Vec<String>,
pub report_advisory_warnings: Vec<AdvisoryWarning>,
}
pub fn run(args: Args<'_>) -> Result<(), CliError> {
let spec = args.spec;
let outcome = execute(&args)?;
let payload = MergePayload {
run_id: &outcome.run_id,
node_id: &outcome.node_id,
branch: &outcome.branch,
source: outcome.source.as_deref(),
merged: outcome.merged,
report_seq: outcome.report_seq,
supervisor: outcome.supervisor.clone(),
report_advisory_warnings: &outcome.report_advisory_warnings,
dry_run: outcome.dry_run.then_some(true),
};
emit(&payload, spec, &outcome.warnings)
}
pub(crate) fn execute(args: &Args<'_>) -> Result<MergeOutcome, CliError> {
let run_id = args.run_id.clone();
let node_id = parse_node_id(args.node_id.as_deref().unwrap_or(DEFAULT_NODE_ID))?;
let source = match &args.source {
Some(s) => Some(require_nonempty(s, "source")?),
None => None,
};
let root = crate::home::root_dir()?;
let paths = run_paths_from_cli_arg(&root, &run_id)?;
let run_id = paths.run_id.as_str().to_string();
let manifest = read_manifest_opt(&paths)
.map_err(from_core)?
.ok_or_else(|| {
CliError::user("run_not_found", format!("no run with id {run_id}"))
.with_invalid_value(&run_id)
})?;
crate::run::reject_legacy_kind(manifest.kind, &run_id)?;
let node = read_node_opt(&paths, &node_id)
.map_err(from_core)?
.ok_or_else(|| {
CliError::user(
"node_not_found",
format!("no node {node_id} in run {run_id}"),
)
.with_invalid_value(node_id.as_str())
})?;
let worktree_path = node.worktree_path.as_deref().ok_or_else(|| {
CliError::user(
"no_worktree",
format!("node {node_id} has no worktree to merge (driver node?)"),
)
.with_invalid_value(node_id.as_str())
})?;
let branch = branch_for(&node).ok_or_else(|| {
CliError::user(
"no_branch",
format!("node {node_id} has no branch recorded; cannot merge"),
)
.with_invalid_value(node_id.as_str())
})?;
if manifest.status == octl_core::Status::Cancelled {
return Err(CliError::user(
"run_already_terminal",
format!(
"run {run_id} is cancelled — merging is not permitted; a cancelled run \
never adopts a merge, so `run merge` would do nothing."
),
)
.with_invalid_value(&run_id));
}
if !args.dry_run {
let worktree_gone = !Path::new(worktree_path).try_exists().unwrap_or(true);
if worktree_gone {
let status = RunLock::with_shared_lock(&paths.lock(), || {
Ok(read_manifest_opt(&paths)?.map(|m| m.status))
})
.map_err(from_core)?
.unwrap_or(manifest.status);
if status.is_terminal() {
let verb = if status == octl_core::Status::Done {
"already done"
} else {
"failed"
};
return Err(CliError::user(
"run_already_terminal",
format!(
"run {run_id} is {verb} and its worktree has been torn down — there is \
no worktree left to merge. If teardown looks incomplete (tmux window \
still open, branch still present), run `orchestratectl run reattach \
{run_id}` to finish it."
),
)
.with_invalid_value(&run_id));
}
return Err(CliError::user(
"worktree_missing",
format!(
"worktree {worktree_path} does not exist — cannot merge. If the run has \
finished, its supervisor may not have rolled up yet; run \
`orchestratectl run reattach {run_id}`. Otherwise the worktree was \
removed out from under a live run."
),
)
.with_invalid_value(&run_id));
}
}
let effective_source = source.clone().or_else(|| manifest.source_branch.clone());
let (mut report, advisory_warnings) = build_report(
args.report_file.as_deref(),
branch,
effective_source.as_deref(),
)?;
let base_warnings: Vec<String> = args
.warnings
.iter()
.cloned()
.chain(advisory_warnings.iter().map(AdvisoryWarning::to_message))
.collect();
if args.dry_run {
return Ok(MergeOutcome {
run_id: run_id.clone(),
node_id: node_id.as_str().to_string(),
branch: branch.to_string(),
source: effective_source.clone(),
merged: false,
report_seq: None,
supervisor: None,
dry_run: true,
warnings: base_warnings,
report_advisory_warnings: advisory_warnings,
});
}
let git = git_bin();
match merge_recovery::recover_node(&paths, &node_id, &git) {
merge_recovery::Recovery::Completed => {
let (outcome, warnings) = ensure_report_consumer(&paths, &run_id, &base_warnings);
return Ok(MergeOutcome {
run_id: run_id.clone(),
node_id: node_id.as_str().to_string(),
branch: branch.to_string(),
source: effective_source.clone(),
merged: true,
report_seq: None,
supervisor: Some(outcome),
dry_run: false,
warnings,
report_advisory_warnings: advisory_warnings.clone(),
});
}
merge_recovery::Recovery::DriverAlive => {
return Err(CliError::user(
"merge_in_progress",
format!(
"another `run merge` is already driving a merge for node {node_id}; \
retry once it finishes"
),
)
.with_invalid_value(&run_id));
}
merge_recovery::Recovery::CannotVerify => {
return Err(CliError::system(
"merge_recovery_unverifiable",
format!(
"a prior merge transaction for node {node_id} could not be verified via git; \
refusing to start a new merge over it — resolve the worktree/source repo and retry"
),
));
}
merge_recovery::Recovery::Rejected { .. }
| merge_recovery::Recovery::Superseded
| merge_recovery::Recovery::NothingPending => {}
}
let merge_start = record_merge_start(
&paths,
&node_id,
worktree_path,
branch,
&node,
effective_source.as_deref(),
&git,
)?;
if let Err(e) = run_merge_sh(
Path::new(worktree_path),
branch,
effective_source.as_deref(),
merge_start.as_ref().map(|h| h.expected_source_oid.as_str()),
) {
if let Some(h) = &merge_start {
abort_merge_start(&paths, &node_id, &h.op_id, "merge did not complete");
}
return Err(e);
}
octl_core::ReportOrigin::RunMerge {
op_id: merge_start.as_ref().map(|h| h.op_id.clone()),
worker_oid: merge_start.as_ref().map(|h| h.worker_oid.clone()),
}
.stamp(&mut report);
let idem_key = format!("explicit-merge:{run_id}:{node_id}");
let result = append_and_apply_event(
&paths,
"node.report",
Some(&node_id),
Some(&idem_key),
report,
)
.map_err(from_core)?;
let (outcome, warnings) = ensure_report_consumer(&paths, &run_id, &base_warnings);
Ok(MergeOutcome {
run_id: run_id.clone(),
node_id: node_id.as_str().to_string(),
branch: branch.to_string(),
source: effective_source.clone(),
merged: true,
report_seq: Some(result.seq),
supervisor: Some(outcome),
dry_run: false,
warnings,
report_advisory_warnings: advisory_warnings,
})
}
fn ensure_report_consumer(
paths: &octl_core::RunPaths,
run_id: &str,
base: &[String],
) -> (ConsumerOutcome, Vec<String>) {
let probed = RunLock::with_shared_lock(&paths.lock(), || {
let manifest = read_manifest_opt(paths)?;
let terminal = manifest.as_ref().is_some_and(|m| m.status.is_terminal());
let warranted = manifest
.as_ref()
.is_some_and(|m| m.kind.lifecycle() == octl_core::Lifecycle::Autonomous)
|| cleanup::any_node_merged_explicitly(paths);
let live = SupervisorView::probe(paths);
let ever_supervised = live.pid.is_some()
|| read_all_events(&paths.events())?
.iter()
.any(|e| e.kind == "supervisor.started");
Ok((terminal, warranted, live, ever_supervised))
});
let (terminal, warranted, live, ever_supervised) = match probed {
Ok(v) => v,
Err(e) => {
let mut warnings = base.to_vec();
warnings.push(format!(
"could not verify supervisor liveness after merge ({e}); if teardown \
does not complete, run `orchestratectl run reattach {run_id}`"
));
return (
ConsumerOutcome::Deferred {
recovery_command: format!("orchestratectl run reattach {run_id}"),
},
warnings,
);
}
};
if live.alive {
return (ConsumerOutcome::Alive, base.to_vec());
}
if !ever_supervised {
return (ConsumerOutcome::NotSupervised, base.to_vec());
}
if terminal && !warranted {
return (ConsumerOutcome::Terminal, base.to_vec());
}
let who = live.pid.map_or_else(
|| "the supervisor".to_string(),
|p| format!("supervisor (pid {p})"),
);
let mut warnings = base.to_vec();
let outcome = match reattach::spawn_supervisor(paths, run_id, false, None) {
Ok(0) => {
warnings.push(format!(
"{who} was not running; restarted it to consume the terminal report \
and complete teardown (new pid not yet confirmed — check \
`orchestratectl run show {run_id}`)"
));
ConsumerOutcome::Reattached { pid: None }
}
Ok(new_pid) => {
warnings.push(format!(
"{who} was not running; restarted it (pid {new_pid}) to consume the \
terminal report and complete teardown"
));
ConsumerOutcome::Reattached { pid: Some(new_pid) }
}
Err(e) if e.code == "supervisor_already_running" => ConsumerOutcome::Alive,
Err(e) => {
let recovery_command = format!("orchestratectl run reattach {run_id}");
warnings.push(format!(
"{who} is not running and auto-reattach failed ({}); teardown (tmux \
window, worktree, branch) is deferred — run `{recovery_command}` to \
complete it",
e.message
));
ConsumerOutcome::Deferred { recovery_command }
}
};
(outcome, warnings)
}
struct MergeStartHandle {
op_id: String,
expected_source_oid: String,
worker_oid: String,
}
fn record_merge_start(
paths: &octl_core::RunPaths,
node_id: &NodeId,
worktree_path: &str,
worker_branch: &str,
node: &Node,
source_branch: Option<&str>,
git: &str,
) -> Result<Option<MergeStartHandle>, CliError> {
let Some(source_branch) = source_branch else {
return Ok(None);
};
let Some(expected_source_oid) = merge_recovery::read_oid(git, worktree_path, source_branch)
else {
return Ok(None);
};
let Some(worker_oid) = merge_recovery::read_oid(git, worktree_path, worker_branch)
.or_else(|| merge_recovery::read_oid(git, worktree_path, "HEAD"))
else {
return Ok(None);
};
let pid = std::process::id();
let txn = MergeTxn {
op_id: octl_core::new_op_id(),
source_branch: source_branch.to_string(),
worker_branch: worker_branch.to_string(),
expected_source_oid: expected_source_oid.clone(),
worker_oid: worker_oid.clone(),
base_sha: node.base_sha.clone(),
driver_pid: Some(pid as i32),
driver_pid_start_secs: crate::supervise::watchdog::pid_start_time(pid),
started_at: Utc::now(),
};
let data = serde_json::to_value(&txn)
.map_err(|e| CliError::system("merge_txn_serialize_failed", e.to_string()))?;
append_and_apply_event(
paths,
octl_core::KIND_MERGE_STARTED,
Some(node_id),
None,
data,
)
.map_err(|e| {
CliError::system(
"merge_txn_record_failed",
format!(
"could not durably record the merge transaction before mutating git ({e}); \
refusing to merge unguarded — retry"
),
)
})?;
Ok(Some(MergeStartHandle {
op_id: txn.op_id,
expected_source_oid,
worker_oid,
}))
}
fn abort_merge_start(paths: &octl_core::RunPaths, node_id: &NodeId, op_id: &str, reason: &str) {
let data = json!({ "op_id": op_id, "reason": reason });
if let Err(e) = append_and_apply_event(
paths,
octl_core::KIND_MERGE_ABORTED,
Some(node_id),
Some(&format!(
"merge-aborted:{}:{node_id}:{op_id}",
paths.run_id.as_str()
)),
data,
) {
tracing::warn!(
target: "orchestratectl::merge",
error = %e,
"could not record merge.aborted after a failed merge; recovery will clear it"
);
}
}
fn branch_for(node: &Node) -> Option<&str> {
node.branch.as_deref().filter(|s| !s.is_empty())
}
fn build_report(
report_file: Option<&Path>,
branch: &str,
source: Option<&str>,
) -> Result<(Value, Vec<AdvisoryWarning>), CliError> {
let mut report = if let Some(path) = report_file {
read_report_file(path)?
} else {
let summary = source.map_or_else(
|| format!("merged {branch} via run merge"),
|src| format!("merged {branch} into {src} via run merge"),
);
json!({ "success": true, "summary": summary })
};
let obj = report.as_object_mut().ok_or_else(|| {
CliError::user(
"report_not_object",
"--report-file payload must be a JSON object",
)
})?;
if obj
.get("success")
.is_some_and(|v| v.as_bool() != Some(true))
|| obj
.get("cancelled")
.is_some_and(|v| v.as_bool() == Some(true))
{
return Err(CliError::user(
"invalid_merge_report",
"--report-file for `run merge` must set `success: true` (a merge is a success) \
and must not set `cancelled: true`",
));
}
obj.insert(
"via".to_string(),
Value::String(octl_core::VIA_EXPLICIT_MERGE.to_string()),
);
let sanitized = sanitize_report_advisory(&report)
.map_err(crate::node::report::map_report_validation_error)?;
Ok((sanitized.report, sanitized.warnings))
}
fn read_report_file(path: &Path) -> Result<Value, CliError> {
let mut f = std::fs::File::open(path).map_err(|e| {
CliError::user(
"report_file_unreadable",
format!("open {}: {}", path.display(), e),
)
.with_invalid_value(path.display().to_string())
})?;
let mut buf = Vec::new();
std::io::Read::by_ref(&mut f)
.take(MAX_REPORT_BYTES + 1)
.read_to_end(&mut buf)
.map_err(|e| {
CliError::user(
"report_file_unreadable",
format!("read {}: {}", path.display(), e),
)
.with_invalid_value(path.display().to_string())
})?;
if buf.len() as u64 > MAX_REPORT_BYTES {
return Err(CliError::user(
"report_file_too_large",
format!("--report-file exceeds maximum of {MAX_REPORT_BYTES} bytes"),
)
.with_invalid_value(path.display().to_string()));
}
serde_json::from_slice(&buf).map_err(|e| {
CliError::user(
"report_file_invalid_json",
format!("parse {}: {}", path.display(), e),
)
.with_invalid_value(path.display().to_string())
})
}
fn materialize_merge_sh() -> Result<MergeScript, CliError> {
if let Ok(path) = std::env::var("OCTL_MERGE_SH") {
return Ok(MergeScript::External(path.into()));
}
let mut tmp = tempfile::Builder::new()
.prefix("orchestratectl-merge-")
.suffix(".sh")
.tempfile()
.map_err(|e| {
CliError::system("tempfile_failed", format!("create merge.sh tempfile: {e}"))
})?;
tmp.write_all(MERGE_SH.as_bytes())
.map_err(|e| CliError::system("write_failed", format!("write merge.sh tempfile: {e}")))?;
tmp.flush()
.map_err(|e| CliError::system("write_failed", format!("flush merge.sh tempfile: {e}")))?;
let perms = std::fs::Permissions::from_mode(0o700);
std::fs::set_permissions(tmp.path(), perms)
.map_err(|e| CliError::system("chmod_failed", format!("chmod merge.sh tempfile: {e}")))?;
Ok(MergeScript::Temp(tmp))
}
enum MergeScript {
External(std::path::PathBuf),
Temp(tempfile::NamedTempFile),
}
impl MergeScript {
fn path(&self) -> &Path {
match self {
MergeScript::External(p) => p.as_path(),
MergeScript::Temp(t) => t.path(),
}
}
}
fn run_merge_sh(
worktree_path: &Path,
branch: &str,
source: Option<&str>,
expected_source_oid: Option<&str>,
) -> Result<(), CliError> {
let script = materialize_merge_sh()?;
let mut cmd = Command::new(script.path());
cmd.current_dir(worktree_path);
if let Some(src) = source {
cmd.arg("--target").arg(src);
}
if let Some(oid) = expected_source_oid {
cmd.arg("--expected-source-oid").arg(oid);
}
cmd.arg("--driver-pid").arg(std::process::id().to_string());
cmd.arg(branch);
let output = cmd.output().map_err(|e| {
if e.kind() == std::io::ErrorKind::NotFound
&& worktree_path.try_exists().is_ok_and(|exists| !exists)
{
CliError::user(
"worktree_missing",
format!(
"worktree {} no longer exists — it was likely torn down as the run \
finished; no merge is needed",
worktree_path.display()
),
)
} else {
CliError::system(
"merge_spawn_failed",
format!("invoke merge.sh ({}): {}", script.path().display(), e),
)
}
})?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
let stdout = String::from_utf8_lossy(&output.stdout);
let detail = stderr.trim();
let detail = if detail.is_empty() {
stdout.trim()
} else {
detail
};
let exit_code = output.status.code().unwrap_or(-1);
let code = if exit_code == MERGE_SH_LOCK_TIMEOUT_EXIT {
"merge_in_progress"
} else if exit_code == MERGE_SH_CAS_MISMATCH_EXIT {
"merge_source_moved"
} else {
"merge_failed"
};
return Err(CliError {
kind: ExitKind::User,
code: code.to_string(),
message: format!("merge.sh exited {exit_code} merging {branch}: {detail}"),
invalid_value: Some(branch.to_string()),
expected: None,
});
}
Ok(())
}
fn emit(
payload: &MergePayload<'_>,
spec: &OutputSpec,
warnings: &[String],
) -> Result<(), CliError> {
match spec.format {
OutputFormat::Json | OutputFormat::Jsonl => {
output::emit_envelope(payload, spec, warnings)?;
}
OutputFormat::Text => {
println!("run-id: {}", payload.run_id);
println!("node-id: {}", payload.node_id);
println!("branch: {}", payload.branch);
match payload.source {
Some(s) => println!("source: {s}"),
None => println!("source: (auto-detect main/master)"),
}
if payload.dry_run == Some(true) {
println!("note: --dry-run (no merge, no report)");
} else {
println!("merged: {}", payload.merged);
if let Some(seq) = payload.report_seq {
println!("report_seq: {seq}");
}
}
output::emit_text_warnings(warnings);
}
}
Ok(())
}