orchestratectl 0.1.6

Rust CLI for orchestrating AI-agent workflows on a developer's machine.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
//! `run merge` — own the full merge lifecycle of one worktree run.
//!
//! Closes the spawn → work → merge → cleanup loop end-to-end inside
//! orchestratectl (issue `bundle-worktree-merge`). Before this verb the
//! merge half lived in the homebase `/worktree-merge` bash skill, which
//! had no knowledge of the run: it merged, but never submitted a terminal
//! `node.report`, so the supervisor kept polling and (for interactive
//! kinds) never tore the window down. `run merge` does both halves in one
//! call:
//!
//!   1. **Merge mechanics** — shell out to the bundled `merge.sh`
//!      (embedded below, materialized to a temp file at runtime). It owns
//!      the rebase, the cross-worktree `flock`, and the `workmux merge`. v1
//!      deliberately wraps the script rather than re-implementing git
//!      wrappers in Rust (issue §4); v2 can move it into core. The worktree /
//!      window / branch teardown is NOT merge.sh's — it belongs to the
//!      supervisor (state-integrity invariant #5).
//!   2. **Terminal report** — on a clean merge, append a `node.report`
//!      with `via: "explicit-merge"`. That flag is the signal the
//!      supervisor's cleanup gate checks to extend teardown to
//!      *interactive* kinds: a user who runs `run merge` is done with the
//!      review window, so it may close (see supervise/cleanup.rs).
//!   3. **Ensure the supervisor tears down.** The supervisor is the SOLE
//!      teardown actor (state-integrity invariant #5); `run merge` never reclaims
//!      resources itself. Once its `via: "explicit-merge"` report is appended, the
//!      octl-core reducer ADOPTS it — even against a node an earlier watchdog
//!      `agent-died` false positive already terminalized (issue
//!      `reducer-adopt-explicit-merge`) — so the cleanup gate sees the merge marker
//!      on every path and warrants teardown. The one path with no LIVE supervisor
//!      is exactly that swallowed case: the watchdog terminal rolled the run up and
//!      its supervisor exited before the merge. [`ensure_report_consumer`]
//!      reattaches one there (reattaching on a terminal-but-teardown-warranted run
//!      when the reducer freshly adopted the report), so the reattached supervisor
//!      runs the same `cleanup_terminal_nodes` and exits. This replaced the inline
//!      worktree/branch reclaim shipped for `merge-skips-teardown` /
//!      `agent-died-merge-no-teardown-interactive`, restoring single-owner teardown.
//!
//! On a merge failure (conflicts, dirty tree, lock timeout) the report is
//! NOT submitted — the node stays live so the agent can recover (e.g.
//! `/complex-rebase`) and re-run `run merge`.

use std::io::{Read as _, Write as _};
use std::os::unix::fs::PermissionsExt as _;
use std::path::{Path, PathBuf};
use std::process::Command;

use serde::Serialize;
use serde_json::{json, Value};

use octl_core::report::validate_report_payload;
use octl_core::{
    append_and_apply_event, read_all_events, read_manifest_opt, read_node_opt, Node, RunLock,
};

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;

/// The bundled merge backend, embedded at compile time so the binary is
/// self-contained (the homebase `merge.sh` is sunset). Materialized to a
/// temp file and executed per invocation. Tests override the resolved
/// script via `OCTL_MERGE_SH`, mirroring `spawn.rs`'s `OCTL_CREATE_SH`.
const MERGE_SH: &str = include_str!("../../scripts/merge.sh");

/// Default reporting node for a single-worker run. Every worktree kind
/// `run merge` targets has exactly one node.
const DEFAULT_NODE_ID: &str = "n-0001";

/// Cap on `--report-file` size, mirroring `node report`'s 1 MiB bound.
const MAX_REPORT_BYTES: u64 = 1024 * 1024;

/// The exit status `merge.sh` reserves for "could not acquire the merge lock in
/// time" — another self-merge into the SAME target branch held the serializing
/// lock past the timeout (issue `concurrent-self-merge-race`). It is the sole
/// producer of this status in the script (the mkdir-lock acquire loop is the
/// only path that emits 75, and the `workmux` invocation normalizes its exit so
/// a downstream 75 can't leak), so mapping it to a distinct,
/// retryable `merge_in_progress` code is unambiguous. Value is `EX_TEMPFAIL`
/// (75) from sysexits(3): "temporary failure, the user is invited to retry".
const MERGE_SH_LOCK_TIMEOUT_EXIT: i32 = 75;

pub struct Args<'a> {
    pub run_id: String,
    /// Override the merge target branch. Falls back to the manifest's
    /// `source_branch`, then to merge.sh's own main/master auto-detection.
    pub source: Option<String>,
    /// Reporting node id; defaults to `n-0001`.
    pub node_id: Option<String>,
    /// Optional §7.3 report payload to submit on a clean merge. When set,
    /// `run merge` stamps it with `via: "explicit-merge"` and submits it —
    /// so an autonomous kind can carry its rich `discussion_items` /
    /// `spinoff_proposals` / `wrap_up_recommendations` in the SAME call
    /// that merges. When absent, a minimal `{success, summary, via}` report
    /// is submitted (enough for a simple spinoff).
    pub report_file: Option<PathBuf>,
    /// Human-reviewer acknowledgement that a `code` run may be merged. A `code`
    /// run is human-reviewed: the reviewer lands the branch after review, never
    /// the coding agent. There is no process-level way to tell the agent's `run
    /// merge` from the human's (both run inside the same claude session), so this
    /// flag is a TRIPWIRE, not proof of a human: it turns the agent's natural
    /// end-of-work `run merge` into a hard STOP rather than a silent self-merge
    /// (issue `interactive-code-run-self-merged`). A no-op for every other kind
    /// (not read, not persisted, not emitted) and skipped on `--dry-run`.
    pub confirm_interactive: bool,
    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,
    /// The resolved merge target, or `null` when left to merge.sh's
    /// main/master auto-detection.
    #[serde(skip_serializing_if = "Option::is_none")]
    source: Option<&'a str>,
    merged: bool,
    /// `node.report` seq, when a terminal report was appended.
    #[serde(skip_serializing_if = "Option::is_none")]
    report_seq: Option<u64>,
    /// Outcome of ensuring a live consumer for the terminal report — the
    /// machine-readable companion to any `warnings` entry, so an agent reads a
    /// `state` instead of regex-parsing prose. Present on a real merge, absent
    /// on `--dry-run`. Additive (no `schema_version` bump).
    #[serde(skip_serializing_if = "Option::is_none")]
    supervisor: Option<ConsumerOutcome>,
    #[serde(skip_serializing_if = "Option::is_none")]
    dry_run: Option<bool>,
}

/// Machine-readable result of [`ensure_report_consumer`]: what state the run's
/// per-run supervisor was left in after the terminal report was appended. The
/// non-silent counterpart to the merge `warnings`.
#[derive(Serialize)]
#[serde(tag = "state", rename_all = "kebab-case")]
enum ConsumerOutcome {
    /// A live supervisor is already consuming the report on its next tick.
    Alive,
    /// The run was already terminal — a supervisor rolled it up and exited, so
    /// nothing remains to consume.
    Terminal,
    /// The run was never supervised (e.g. a skeleton/test run) — there is no
    /// teardown actor to restart, and spawning one would be wrong.
    NotSupervised,
    /// The dead supervisor was restarted; teardown is (re)running. `pid` is the
    /// new supervisor's pid, or `null` when the spawn was not yet confirmed.
    Reattached { pid: Option<u32> },
    /// No live consumer could be ensured; teardown is deferred until the caller
    /// runs `recovery_command`.
    Deferred { recovery_command: String },
}

pub fn run(args: Args<'_>) -> Result<(), 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 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)
        })?;

    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())
        })?;

    // The worktree directory is the cwd the merge backend needs: it derives
    // the branch and the source-side worktree from `git` run there. A node
    // with no materialized worktree (a driver node) cannot be merged.
    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())
    })?;

    // Terminal-state / torn-down-worktree guard (issue `merge-terminal-misleading`).
    //
    // A run that is already done merges its worktree away: the supervisor tears
    // the worktree down and exits when the run rolls up terminal (invariant #5).
    // A later `run merge` would then materialize merge.sh and `cd` into a
    // directory that no longer exists, failing with a misleading
    // `merge_spawn_failed: … No such file or directory` that fingers the temp
    // script instead of the real cause (the run is already finished). Refuse up
    // front with a clear code and NO spawn attempt.
    //
    // The discriminator is deliberately worktree EXISTENCE, not "was there an
    // explicit-merge report" — the latter would (a) wrongly refuse the ONE
    // legitimate merge against a terminal run: the watchdog `agent-died`
    // false-positive path (issue `reducer-adopt-explicit-merge`), where the run
    // is terminal (`Failed`) but the still-alive agent's worktree survives (a
    // blocked handoff PRESERVES it) and its `run merge` must proceed and be
    // adopted; and (b) break the documented crash-safe idempotent retry — if a
    // merge appended its report then crashed before `ensure_report_consumer`,
    // the worktree still exists, so this falls through to the idempotent
    // re-merge + reattach path that completes teardown. Worktree existence
    // cleanly separates both survivors (worktree present → fall through) from
    // the genuinely-finished run (worktree gone → refuse), and it also catches
    // a terminal run torn down WITHOUT an explicit merge (a genuine autonomous
    // failure), which the marker check would have missed. Branch on
    // `manifest.status` (invariant #4), never `lifecycle`.
    //
    // A `Cancelled` run is refused regardless of its worktree, and even under
    // `--dry-run` (so a preview never claims a merge is planned for a run that
    // would hard-fail). The load-bearing reason is the reducer's adoption
    // whitelist — a late `explicit-merge` report is adopted only against a
    // `Failed | Done` node (see `reduce_node_report` / `reducer-adopt-explicit-merge`),
    // never `Cancelled` — so merging a cancelled run would land git state the
    // run state can never reflect, then strand teardown. Refuse instead.
    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));
    }

    // `--dry-run` is a read-only preview that never spawns merge.sh, so it is
    // exempt from the remaining worktree-existence checks.
    if !args.dry_run {
        // try_exists returns Ok(false) only for a definitely-absent path; an
        // Err (e.g. a permission error mid-path) is "cannot tell", and
        // `unwrap_or(true)` treats that as present so we fall through to a real
        // merge attempt (which surfaces the true error) rather than a false
        // "already terminal".
        let worktree_gone = !Path::new(worktree_path).try_exists().unwrap_or(true);
        if worktree_gone {
            // The top-of-function manifest read is unlocked and may predate a
            // supervisor rollup that removed this very worktree. Re-read the
            // status fresh under the shared lock so the terminal-vs-live
            // classification (and thus the error) is correct even under that
            // race (invariant #3 for the read).
            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));
            }
            // A still-live run whose worktree has vanished is a distinct,
            // actionable state — surface it plainly instead of the misleading
            // merge.sh spawn failure. If the run actually finished but its
            // supervisor has not rolled the manifest up yet, `run reattach`
            // completes the transition.
            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));
        }
    }

    // `code`-run merge gate (issue `interactive-code-run-self-merged`).
    //
    // A `code` run exists precisely for the human review gate: the agent works
    // autonomously up to `/wrap-up`, then STOPS and idles while the human reviews,
    // and the human — not the agent — lands the branch. A real bug had a `code`
    // run reach `done` and merge its 5 commits to `main` with no human merge and
    // no review pause, because the agent ran `run merge` on itself as its natural
    // end-of-work step (the report carried the same `via: "explicit-merge"` stamp
    // the human's merge produces — the supervisor/reducer cannot tell them apart,
    // and neither can this verb by process identity: agent and human share one
    // claude session in the same pane).
    //
    // This gate is a TRIPWIRE, not an authorization boundary: any caller that can
    // run `run merge` can also pass `--confirm-interactive`, so it cannot stop a
    // determined agent — the actual safeguard is the brief prohibition injected by
    // the `worktree-code` skill (see its SKILL.template.md). What the gate buys is
    // turning the observed failure — an obedient agent whose natural last step is a
    // bare `run merge` — into a hard STOP with a distinct error code, instead of a
    // silent self-merge. So the refusal is worded to stop a coding agent, and it
    // deliberately does NOT spell out the flag or name the human's merge command:
    // teaching the escape hatch in the very error that blocks it would defeat the
    // tripwire. Scoped to `Kind::Code` (the only kind with this human-review
    // protocol; `orchestrate`'s driver has no worktree to self-merge). Autonomous
    // kinds self-merge as before. `--dry-run` is a read-only preview with no merge
    // and no report, so it is exempt — there is nothing to gate.
    if manifest.kind == octl_core::Kind::Code && !args.confirm_interactive && !args.dry_run {
        return Err(CliError::user(
            "interactive_merge_requires_confirmation",
            format!(
                "run {run_id} is a human-reviewed `{}` run — this branch is landed by the human \
                 reviewer after review, never by the agent. If you are the coding agent: STOP. \
                 Do not merge your own run; finish with `/wrap-up` and idle for the human.",
                manifest.kind.wire_name()
            ),
        )
        .with_invalid_value(&run_id));
    }

    // Resolve the merge target: explicit `--source` wins, else the
    // manifest's source_branch (the integration branch for an orchestrated
    // child, `main` for a code worktree), else None → merge.sh detects
    // main/master itself.
    let effective_source = source.clone().or_else(|| manifest.source_branch.clone());

    // Build the terminal report up front — BEFORE the merge — so a malformed
    // `--report-file` is rejected without having already merged. The report
    // is only submitted after a clean merge; here we just validate its shape
    // and stamp the `via: "explicit-merge"` marker.
    let report = build_report(
        args.report_file.as_deref(),
        branch,
        effective_source.as_deref(),
    )?;

    if args.dry_run {
        let payload = MergePayload {
            run_id: &run_id,
            node_id: node_id.as_str(),
            branch,
            source: effective_source.as_deref(),
            merged: false,
            report_seq: None,
            supervisor: None,
            dry_run: Some(true),
        };
        return emit(&payload, args.spec, args.warnings);
    }

    // Run the merge. A non-zero exit (conflict, dirty tree, lock timeout)
    // surfaces as a CliError and the report is NOT submitted — the node
    // stays live for the agent to recover and retry.
    run_merge_sh(
        Path::new(worktree_path),
        branch,
        effective_source.as_deref(),
    )?;

    // Merge succeeded: submit the terminal report (built above, stamped with
    // `via: "explicit-merge"`) so the supervisor's cleanup gate extends
    // teardown to interactive kinds and any rich `discussion_items` /
    // `spinoff_proposals` reach the parent.
    //
    // Idempotent: a retried `run merge` (e.g. the report append failed but
    // the merge already landed) re-uses the same key and returns the prior
    // seq instead of double-appending. The merge itself is also a clean
    // no-op on retry (the branch is already merged, worktree may be gone).
    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)?;

    // The terminal report is only useful if a supervisor consumes it: the
    // supervisor is the canonical teardown actor (close tmux window, remove
    // worktree, delete branch) AND the roller-up of `manifest.status` (state
    // integrity invariant #5). `run merge` no longer tears anything down itself:
    // the octl-core reducer now ADOPTS a late `via: "explicit-merge"` report even
    // against a node an earlier watchdog `agent-died` false positive already
    // terminalized (issue `reducer-adopt-explicit-merge`). So `last_report` carries
    // the merge marker, `any_node_merged_explicitly` sees it, and the supervisor
    // warrants teardown on every path — restoring single-owner teardown and
    // retiring the inline reclaim of `merge-skips-teardown` /
    // `agent-died-merge-no-teardown-interactive`.
    //
    // The one path where no live supervisor exists is exactly that swallowed case:
    // the watchdog terminal already rolled the run up and its supervisor exited, so
    // there is nothing running to consume our now-adopted report. `ensure_report_
    // consumer` reattaches one — including on a terminal-but-teardown-warranted run
    // — so the reattached supervisor runs the same `cleanup_terminal_nodes` and
    // exits. A bare `merged: true` never returns while teardown has no actor.
    //
    // `result` (seq / idempotent_replay / applied) is intentionally NOT used to
    // gate the reattach: the reattach is driven by durable projection state
    // (`manifest.status` + the adopted merge marker), read fresh under the shared
    // lock, so an idempotent `run merge` retry after a crash between append and
    // reattach still reattaches and completes teardown (the reducer's adoption is
    // durable even though this call's `applied` is false). See the 4-model review
    // (`reducer-adopt-explicit-merge`) — gating on `applied` here was a crash-retry
    // leak. (`result.seq` is still surfaced in the envelope below.)
    let (outcome, warnings) = ensure_report_consumer(&paths, &run_id, args.warnings);

    let payload = MergePayload {
        run_id: &run_id,
        node_id: node_id.as_str(),
        branch,
        source: effective_source.as_deref(),
        merged: true,
        report_seq: Some(result.seq),
        supervisor: Some(outcome),
        dry_run: None,
    };
    emit(&payload, args.spec, &warnings)
}

/// Guarantee the terminal `node.report` just appended has a live consumer, or
/// surface why not — never silent success. Returns the machine-readable
/// [`ConsumerOutcome`] plus the caller's `base` warnings with (at most) one
/// human-readable entry appended.
///
/// The recovery decision reasons across FOUR facts, read together under one
/// shared lock (state-integrity invariant #3): the manifest status, whether
/// teardown is warranted (`any_node_merged_explicitly`, since a `run merge`
/// always makes teardown warranted once its report is adopted), the
/// `supervisor.pid` liveness, and whether a supervisor was EVER started. The
/// discriminator for "reattach or not" is deliberately NOT "is a dead pid file
/// present" — an orphan can also have NO pid file (the `claim_pid_atomic` hint
/// tells an operator to delete a stale `supervisor.pid`, after which a merge
/// would otherwise silently strand the run). The sound rule is:
///
///   reattach  ⟺  no live supervisor  ∧  ever supervised
///                ∧  ( run not yet terminal  ∨  (terminal ∧ warranted) )
///
/// where "ever supervised" = a recorded pid (stale file) OR a `supervisor.started`
/// event in the log, and "warranted" = the adopted merge (or an autonomous kind)
/// means the supervisor will tear the worktree/branch/window down. The terminal
/// clause is what closes the swallowed-report leak (issues `merge-skips-teardown`,
/// `agent-died-merge-no-teardown-interactive`): a watchdog `agent-died` false
/// positive rolled the run up and its supervisor exited BEFORE the merge, so
/// reattaching one is the ONLY way teardown runs — and the reattached supervisor,
/// seeing the now-adopted `via: "explicit-merge"` report, warrants it.
///
/// The reattach is driven purely by DURABLE state (`manifest.status` + the adopted
/// merge marker), never by whether THIS call freshly adopted the report. An
/// earlier revision gated it on `AppendResult::applied`, but the 4-model review of
/// `reducer-adopt-explicit-merge` showed that leaks: if `run merge` adopts, then
/// crashes before reattaching, the retried merge is an idempotent replay
/// (`applied == false`) and would skip the reattach, stranding the worktree
/// forever. Reattaching whenever teardown is warranted-but-unmanned is safe
/// because the supervisor's cleanup is idempotent — a redundant reattach on an
/// already-torn-down run boots, finds nothing to remove, and exits — and
/// `spawn_supervisor`'s `supervisor_already_running` guard prevents a double
/// reattach race.
///
/// A never-materialized skeleton run (e.g. a `--skip-materialize` test run) has
/// no recorded pid and no `supervisor.started`, so it is left untouched
/// (`NotSupervised`) — spawning a supervisor for it would be wrong. This is never
/// a production worktree run: `run create` spawns + confirms a supervisor before
/// returning, so a real run always reads `ever supervised`.
///
/// KNOWN RESIDUAL: a legacy bare-integer `supervisor.pid` whose pid has been
/// recycled by an unrelated live process reads as `alive` (§7.6 identity check
/// cannot fire without a recorded start-time), so this returns `Alive` and skips
/// reattach. Modern pid files (the norm) carry a start-time and are immune.
fn ensure_report_consumer(
    paths: &octl_core::RunPaths,
    run_id: &str,
    base: &[String],
) -> (ConsumerOutcome, Vec<String>) {
    // One shared-locked read of manifest.json + nodes/* + supervisor.pid +
    // events.jsonl: the reattach decision is a multi-projection read, so it must
    // not observe a half-applied set (invariant #3). The event scan is
    // short-circuited when a pid file is already present, so the common
    // (signal-death) path pays only the manifest read + node scan + pid probe.
    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());
        // Teardown is warranted for an autonomous kind unconditionally, or for any
        // kind once an explicit `run merge` report has been adopted onto a node —
        // exactly the supervisor's own `cleanup` gate. Read under the same shared
        // lock as the manifest (the fold scans every node projection).
        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,
        // A locked read failed (corrupt log / I/O). Don't guess the run's health
        // — the merge itself already landed; surface a deferred-teardown warning
        // so the caller can recover rather than assuming a clean close.
        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,
            );
        }
    };

    // A live supervisor will consume the report on its next tick.
    if live.alive {
        return (ConsumerOutcome::Alive, base.to_vec());
    }
    // Never supervised (skeleton run): no teardown actor to restart.
    if !ever_supervised {
        return (ConsumerOutcome::NotSupervised, base.to_vec());
    }
    // Already rolled up by a supervisor that has since exited. When teardown is
    // NOT warranted (a terminal interactive run that ended without an explicit
    // merge — the human owns that window), there is nothing to tear down, so
    // report `Terminal`. But when teardown IS warranted (the adopted merge, or an
    // autonomous kind) and no supervisor is live, the exited supervisor never ran
    // it — fall through to reattach, the ONLY actor that will tear the worktree /
    // branch / window down (the swallowed-report path). Idempotent, so a retry
    // whose teardown already completed simply reattaches, no-ops, and exits.
    if terminal && !warranted {
        return (ConsumerOutcome::Terminal, base.to_vec());
    }

    // Orphaned: was supervised, no live supervisor remains to consume the
    // terminal report — either the run is still non-terminal, or it is terminal
    // but this call freshly adopted a merge that warrants teardown the exited
    // supervisor never saw. Restore the invariant by reattaching.
    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) }
        }
        // A live supervisor appeared between the probe and the spawn attempt:
        // there is a consumer after all, so this is not a warning condition.
        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)
}

/// The branch a node works on. Prefers the explicit `branch` field; a
/// well-formed worktree node always has it.
fn branch_for(node: &Node) -> Option<&str> {
    node.branch.as_deref().filter(|s| !s.is_empty())
}

/// Build (and validate) the terminal §7.3 report to submit on a clean merge.
///
/// With `report_file`, read the agent's payload (so an autonomous kind can
/// carry its `discussion_items` / `spinoff_proposals` /
/// `wrap_up_recommendations` in the same call) and stamp it with the
/// `via: "explicit-merge"` marker, overriding any caller-set `via`. Without
/// one, synthesize a minimal `{success, summary, via}` report — enough for a
/// simple spinoff. Either way the result is validated against the §7.3 schema
/// before it can reach the event log.
fn build_report(
    report_file: Option<&Path>,
    branch: &str,
    source: Option<&str>,
) -> Result<Value, 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 })
    };
    // `run merge` owns the marker: stamp it regardless of what the file held.
    let obj = report.as_object_mut().ok_or_else(|| {
        CliError::user(
            "report_not_object",
            "--report-file payload must be a JSON object",
        )
    })?;
    // A clean merge is by definition a SUCCESS. Reject (rather than silently
    // rewrite) a `--report-file` that claims `success: false` or `cancelled: true`:
    // such a report contradicts the merge that just landed, and — stamped with the
    // explicit-merge marker — it would either mis-terminalize a live node as
    // Failed/Cancelled or, against an already-terminal node, fail the reducer's
    // confirmed-merge adoption gate and strand teardown (4-model review of
    // `reducer-adopt-explicit-merge`). Silently changing the caller's outcome
    // fields would be worse than refusing, so refuse.
    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()),
    );

    validate_report_payload(&report)
        .map_err(|e| CliError::user("schema_violation", e.to_string()))?;
    Ok(report)
}

/// Read and JSON-parse a `--report-file`, enforcing the size cap during the
/// read (TOCTOU-safe, mirroring `node report`'s `read_capped`).
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())
    })
}

/// Resolve the merge backend: `OCTL_MERGE_SH` override (tests) or the
/// embedded script materialized to a temp file with the exec bit set.
/// Returns the temp-file guard so it lives until the command has run.
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))
}

/// Where the materialized merge backend lives — an external override path
/// or an owned temp file that must outlive the command invocation.
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(),
        }
    }
}

/// Invoke the merge backend from inside `worktree_path`, inheriting the
/// environment (notably `$TMUX`/`$TMUX_PANE`, which the backend uses to
/// close the agent's window). On a non-zero exit, the captured stderr
/// becomes the error message and the report is skipped by the caller.
fn run_merge_sh(worktree_path: &Path, branch: &str, source: 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);
    }
    cmd.arg(branch);

    let output = cmd.output().map_err(|e| {
        // The pre-flight guard in `run` refuses a torn-down worktree up front,
        // but the worktree can still vanish in the TOCTOU window between that
        // check and this spawn (the supervisor tearing a just-terminalized run
        // down). A `NotFound` here can be the missing `current_dir` OR a missing
        // executable (e.g. an `OCTL_MERGE_SH` override pointing nowhere, or a
        // missing shebang interpreter), so disambiguate by stat-ing the worktree
        // instead of blindly blaming either one.
        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);
        // merge.sh refuses on preconditions (on main, dirty tree, same
        // branch, lock timeout) and fails on a rebase conflict from
        // `workmux merge`. Both are user-actionable: the agent recovers
        // (commit / resolve / `/complex-rebase`) and retries `run merge`.
        let detail = stderr.trim();
        let detail = if detail.is_empty() {
            stdout.trim()
        } else {
            detail
        };
        let exit_code = output.status.code().unwrap_or(-1);
        // merge.sh reserves EX_TEMPFAIL for "could not acquire the merge lock in
        // time" — a concurrent self-merge into the SAME target branch held the
        // serializing lock past the timeout (issue concurrent-self-merge-race).
        // Surface it under a distinct `merge_in_progress` code so a caller can
        // tell a transient serialization conflict (retry) apart from a genuine
        // dirty tree / conflict (commit / resolve). It is retryable, not a hard
        // failure. See MERGE_SH_LOCK_TIMEOUT_EXIT for why 75 is unambiguous.
        let code = if exit_code == MERGE_SH_LOCK_TIMEOUT_EXIT {
            "merge_in_progress"
        } 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(())
}