orchestratectl 0.1.7

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
//! Spawn a detached `orchestratectl supervise <run-id>` process and
//! wait briefly for its PID file to appear.
//!
//! Shared by `run create` (top-level), `run reattach`, and the parent
//! supervisor's child-spawn loop. All three funnel through the same
//! [`detached_supervise_command`] + [`spawn_and_reap`] pair so the process
//! detachment hardening (the double-fork below) lives in exactly one place.
//!
//! # Why double-fork
//!
//! A supervisor must outlive both the terminal that launched it and the
//! process that spawned it, and must never linger as a zombie. The naive
//! `Command::spawn` leaves the child in the spawner's session/process group
//! (so closing the terminal `SIGHUP`s every supervisor) and as a direct
//! child (so an exited supervisor becomes a zombie until the spawner
//! `wait()`s it — and `kill(pid, 0)` reports a zombie as *alive*, corrupting
//! the PID-staleness check).
//!
//! The fix is a classic double-fork, run inside `pre_exec` (after `fork`,
//! before `exec`):
//!   1. `setsid()` — the child leads a new session with no controlling
//!      terminal, so a terminal `SIGHUP` can never reach it.
//!   2. `fork()` again — the intermediate exits *immediately* (`_exit`,
//!      bypassing Rust destructors), so the grandchild is orphaned and
//!      reparented to init (pid 1). Init reaps the grandchild when it
//!      eventually exits, so no zombie ever accrues on our side.
//!
//! The spawner then `wait()`s the short-lived intermediate (it has already
//! `_exit`ed, so the wait returns at once) to reap *it*. Net: nothing the
//! spawner owns can become a zombie, and the real supervisor is fully
//! detached.
//!
//! Because this runs in `pre_exec` — before the supervisor binary's
//! `main`/tracing-subscriber ever initialize — there is no worker thread to
//! be orphaned by the intermediate's exit (a real hazard if you double-fork
//! *after* the runtime starts). The grandchild builds its tracing stack
//! fresh after `exec`.
//!
//! One consequence: the PID `Command::spawn` returns is the intermediate's,
//! not the grandchild's. The authoritative supervisor PID is the one the
//! supervisor writes into its own `supervisor.pid` during
//! [`crate::supervise::pid_file::claim_pid_atomic`]; lenient callers that need
//! it read it back from that file (see [`await_recorded_pid`]).
//!
//! # Confirming boot — readiness pipe vs. pid-file poll
//!
//! `run create` cannot record a run as started until it knows the supervisor
//! booted. It confirms that with a [readiness pipe](crate::run::supervisor_readiness)
//! threaded through the double-fork: the grandchild writes a readiness byte
//! carrying its pid AFTER `claim_pid_atomic` + init, and [`spawn_for_run`]
//! blocks reading it (a byte → confirmed; EOF → the supervisor died during
//! init; a structured error → the real reason). This has no timeout and no
//! orphan window — replacing the old bounded `supervisor.pid` poll that
//! false-failed a slow-but-healthy boot into `supervisor_spawn_failed` while
//! the grandchild kept running (issue `supervisor-confirm-readiness-pipe`).
//! The lenient callers (`run reattach`, child-spawn) do NOT confirm via the
//! pipe; they read `supervisor.pid` directly ([`await_recorded_pid`] /
//! [`read_live_recorded_pid`]) and tolerate "not yet confirmed".

use std::os::fd::RawFd;
use std::path::Path;
use std::process::Command;
use std::time::{Duration, Instant};

use octl_core::RunPaths;

use crate::error::CliError;
use crate::run::supervisor_readiness::{Readiness, ReadinessPipe, ENV_READINESS_FD};
use crate::supervise::pid_file;

/// Generous backstop for the readiness read: a wedge circuit-breaker, NOT the
/// old confirmation deadline. Default 120s (≈8× the retired 15s pid-file poll)
/// so a merely slow-but-healthy boot is never false-failed; it only bounds a
/// supervisor genuinely stuck during init (e.g. blocked on the run flock).
const READY_WAIT: Duration = Duration::from_secs(120);

/// [`READY_WAIT`] in production; tests point `OCTL_READY_WAIT_MS` at a short
/// value so the wedge-backstop path is exercisable in milliseconds. An
/// unparseable value falls back to the production default.
fn ready_wait() -> Duration {
    std::env::var("OCTL_READY_WAIT_MS")
        .ok()
        .and_then(|v| v.trim().parse::<u64>().ok())
        .map_or(READY_WAIT, Duration::from_millis)
}

/// How long the *lenient* pid-file poll ([`await_recorded_pid`], used by
/// `run reattach`) waits for a freshly-forked supervisor to write its live pid
/// file before giving up and reporting pid 0 ("spawned, pid unconfirmed").
///
/// `run create`'s confirmation path no longer uses this: it uses the readiness
/// pipe ([`spawn_for_run`]), which has no timeout and no orphan window. This
/// deadline governs only the lenient callers that tolerate an unconfirmed pid
/// and rely on `supervisor.pid` as the durable truth.
const PID_FILE_WAIT: Duration = Duration::from_secs(15);
const POLL_TICK: Duration = Duration::from_millis(200);

/// How long [`await_recorded_pid`] waits for the supervisor's pid file.
/// [`PID_FILE_WAIT`] in production; tests point `OCTL_PID_FILE_WAIT_MS` at a
/// short value so the fail-loud confirmation path is exercisable in
/// milliseconds. An unparseable value falls back to the production default.
fn pid_file_wait() -> Duration {
    std::env::var("OCTL_PID_FILE_WAIT_MS")
        .ok()
        .and_then(|v| v.trim().parse::<u64>().ok())
        .map_or(PID_FILE_WAIT, Duration::from_millis)
}

/// The binary a detached supervisor is spawned from. Production always uses the
/// current executable (`orchestratectl supervise <run-id>`); tests override via
/// `OCTL_SUPERVISE_BIN` to point at a stub that never writes a pid file, so the
/// silent-spawn-failure path can be tested deterministically. Mirrors the
/// `OCTL_CREATE_SH` seam. Production callers never set it.
fn supervise_bin() -> Result<std::path::PathBuf, CliError> {
    if let Ok(v) = std::env::var("OCTL_SUPERVISE_BIN") {
        return Ok(std::path::PathBuf::from(v));
    }
    std::env::current_exe().map_err(|e| CliError::system("io_error", format!("current_exe: {e}")))
}

/// Outcome of a supervisor spawn. An enum (not a `{ pid, confirmed }` struct)
/// so the two states are mutually exclusive by construction — there is no way
/// to represent the contradictory "confirmed with pid 0" that reintroduced the
/// original silent-success bug.
pub enum SupervisorSpawn {
    /// The supervisor confirmed boot down the readiness pipe, carrying its own
    /// pid (identical to the value it wrote into `supervisor.pid` under the run
    /// flock during `claim_pid_atomic`).
    Confirmed { pid: u32 },
    /// The supervisor never confirmed boot: the readiness pipe closed without a
    /// ready signal (it died during init), it reported a structured boot error,
    /// or the fork/exec itself failed. `reason` carries the specific cause.
    /// `run create` surfaces this as a loud `supervisor_spawn_failed` envelope
    /// rather than recording a bogus success (issue
    /// `supervisor-spawn-fails-silently-at-run-create`). Only `run create`
    /// constructs/consumes this — the lenient callers read the pid file directly.
    Unconfirmed { reason: String },
}

/// Attach the detach hardening (`setsid` + double-fork) to `cmd`'s child via
/// `pre_exec`. See the module docs for the full rationale.
///
/// `readiness_write_fd`, when set, is the readiness pipe's write end. The parent
/// creates the pipe with `FD_CLOEXEC` on both ends (so a concurrent `exec` on
/// another thread cannot leak it); this closure clears CLOEXEC on that fd inside
/// the forked child — right before `exec` — so exactly the intended grandchild
/// inherits it. `fcntl` is async-signal-safe, and the captured `Option<RawFd>`
/// is `Copy` (no allocation, no `Drop` between fork and exec).
fn apply_detach(cmd: &mut Command, readiness_write_fd: Option<RawFd>) {
    use std::os::unix::process::CommandExt;
    // SAFETY: the closure is async-signal-safe — it calls only `fcntl`,
    // `setsid`, `fork`, and `_exit` (all on the POSIX async-signal-safe list)
    // and does no allocation, locking, or other Rust-runtime work between
    // `fork` and `exec`.
    unsafe {
        cmd.pre_exec(move || {
            // Uncloak the readiness write end for the exec that follows. Clearing
            // all fd flags (only FD_CLOEXEC is defined) makes it survive exec
            // into the grandchild supervisor.
            if let Some(fd) = readiness_write_fd {
                if libc::fcntl(fd, libc::F_SETFD, 0) == -1 {
                    return Err(std::io::Error::last_os_error());
                }
            }
            // New session: detach from the controlling terminal and the
            // spawner's process group so a terminal SIGHUP cannot reach us.
            if libc::setsid() == -1 {
                return Err(std::io::Error::last_os_error());
            }
            // Double-fork: the intermediate exits now so the grandchild is
            // reparented to init (pid 1), which reaps it on exit. Use `_exit`
            // (not `return`/`exit`) so no Rust destructors or atexit hooks run
            // in this forked-but-not-exec'd process.
            match libc::fork() {
                -1 => Err(std::io::Error::last_os_error()),
                0 => Ok(()),         // grandchild: continue to exec the supervisor
                _ => libc::_exit(0), // intermediate: vanish, orphaning the grandchild
            }
        });
    }
}

/// Build a detach-hardened `supervise <run-id>` command with stdout/stderr
/// redirected to `log_path`. Callers may append extra args (e.g. `--once`)
/// before handing it to [`spawn_and_reap`].
///
/// `readiness_write_fd` is `Some` only for `run create`'s confirmation path
/// ([`spawn_for_run`]), which also sets [`ENV_READINESS_FD`] to that fd number;
/// the lenient callers pass `None`.
pub fn detached_supervise_command(
    run_id: &str,
    log_path: &Path,
    readiness_write_fd: Option<RawFd>,
) -> Result<Command, CliError> {
    let stderr_file = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(log_path)
        .map_err(|e| CliError::system("io_error", format!("open {}: {}", log_path.display(), e)))?;
    let stderr_clone = stderr_file
        .try_clone()
        .map_err(|e| CliError::system("io_error", format!("dup fd: {e}")))?;
    let exe = supervise_bin()?;
    let mut cmd = Command::new(exe);
    cmd.arg("supervise")
        .arg(run_id)
        // A detached daemon must not keep the launching terminal's stdin: an
        // inherited TTY fd 0 can deliver SIGTTIN or let a stray read consume
        // terminal input. setsid drops the controlling-terminal relationship
        // but not the fd itself.
        .stdin(std::process::Stdio::null())
        .stdout(stderr_file)
        .stderr(stderr_clone)
        // Clear any inherited readiness-fd hint. Only `run create`'s
        // confirmation path ([`spawn_for_run`]) sets `OCTL_READINESS_FD`, and it
        // does so on its OWN command AFTER this builder. Without this clear, a
        // running supervisor (which itself inherited the variable from its
        // parent `run create`) would leak it to every child supervisor it forks,
        // and each child would write a readiness signal to an fd it never
        // inherited.
        .env_remove(ENV_READINESS_FD);
    apply_detach(&mut cmd, readiness_write_fd);
    Ok(cmd)
}

/// Spawn a detach-hardened supervisor `cmd` and reap the short-lived
/// double-fork intermediate so it never lingers as a zombie. The real
/// supervisor (the grandchild) is already reparented to init by the time
/// this returns.
pub fn spawn_and_reap(cmd: &mut Command, run_id: &str) -> Result<(), CliError> {
    let mut child = cmd
        .spawn()
        .map_err(|e| CliError::system("spawn_failed", format!("spawn supervise {run_id}: {e}")))?;
    // The intermediate `_exit`s immediately after forking the grandchild, so
    // this wait reaps it without blocking on the actual supervisor. A wait
    // error (e.g. ECHILD if it was already reaped) is non-fatal — the
    // grandchild is independent — but surface it rather than swallow it.
    if let Err(e) = child.wait() {
        tracing::warn!(
            target: "orchestratectl::supervise",
            run = %run_id,
            error = %e,
            "failed to reap double-fork intermediate (grandchild unaffected)"
        );
    }
    Ok(())
}

/// Read `<run-dir>/supervisor.pid` ONCE and return the recorded pid iff it is
/// a live process whose start-time still matches the record (§7.6 identity
/// check). Non-blocking. Used where the caller must not stall — e.g. the
/// parent supervisor's tick — and is content with "pid not yet confirmed"
/// (the child writes its own pid file as the durable source of truth).
pub fn read_live_recorded_pid(paths: &RunPaths) -> Option<u32> {
    let (pid, start_time) = pid_file::read_pid_record(&paths.supervisor_pid())?;
    pid_file::pid_live_with_identity(pid, start_time).then_some(pid)
}

/// Poll `<run-dir>/supervisor.pid` for up to [`PID_FILE_WAIT`] and return the
/// live, identity-verified supervisor PID it records. `None` if none appears
/// in time — with double-fork we have no usable spawned PID to fall back to
/// (the intermediate we reaped is gone), so callers decide how to degrade. In
/// practice the supervisor writes its pid file under the run flock within
/// milliseconds of `exec`, so the deadline is reached only if the supervisor
/// failed to boot.
///
/// Identity matters: a stale pid file from a prior generation whose pid has
/// been recycled by an unrelated live process must NOT be accepted as "our"
/// supervisor — hence `read_pid_record` + `pid_live_with_identity`, not a bare
/// liveness probe.
pub fn await_recorded_pid(paths: &RunPaths) -> Option<u32> {
    let deadline = Instant::now() + pid_file_wait();
    loop {
        if let Some(pid) = read_live_recorded_pid(paths) {
            return Some(pid);
        }
        if Instant::now() >= deadline {
            return None;
        }
        std::thread::sleep(POLL_TICK);
    }
}

/// Append a diagnostic line to the supervisor's stderr log so a spawn that
/// never got far enough to boot the tracing subscriber still leaves a trace on
/// disk (issue `supervisor-spawn-fails-silently-at-run-create`, suggested-fix
/// #2 "always write supervisor.stderr.log … capture the fork/exec failure
/// reason"). Best-effort: a log-write failure must never mask the spawn error
/// the caller is already returning.
fn append_spawn_diag(log_path: &Path, msg: &str) {
    use std::io::Write as _;
    if let Ok(mut f) = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(log_path)
    {
        let _ = writeln!(f, "[orchestratectl run create] {msg}");
    }
}

/// Fork+exec a fully-detached supervisor with stdout/stderr redirected to
/// `<run-dir>/supervisor.stderr.log`, then confirm its boot over a
/// [readiness pipe](crate::run::supervisor_readiness) — no timeout, no orphan
/// window.
///
/// The stderr log is opened (created, possibly empty) *before* the fork by
/// [`detached_supervise_command`], so a trace file always exists on disk from
/// the moment of spawn. A fork/exec failure, or a supervisor that never
/// confirms boot, is additionally recorded into that log via
/// [`append_spawn_diag`] — otherwise a silent spawn failure leaves zero trace
/// to diagnose from (the original bug signature).
///
/// Confirmation mechanism (issue `supervisor-confirm-readiness-pipe`): the
/// parent creates a pipe whose write end the grandchild inherits across
/// `exec`. The grandchild writes a readiness signal carrying its pid AFTER it
/// has claimed `supervisor.pid` and finished init, then closes the write end.
/// The parent closes its own write-end copy and reads the read end, bounded by
/// a generous wedge backstop ([`ready_wait`]):
///   - a `ready` signal → [`SupervisorSpawn::Confirmed`] with the supervisor pid;
///   - EOF with no signal → the supervisor died during init (fate-sharing) →
///     [`SupervisorSpawn::Unconfirmed`];
///   - a structured error signal → `Unconfirmed` carrying the real reason;
///   - deadline elapsed → `Unconfirmed` (supervisor wedged, e.g. on the run
///     flock — alive but not progressing).
///
/// Confirmation is edge-triggered: the read returns the moment the grandchild
/// signals OR the write end closes, so a slow-but-healthy boot is confirmed
/// whenever it finishes and a genuinely dead supervisor is detected at once —
/// the ambiguity of the old bounded pid-file poll is gone. The backstop only
/// bounds a true hang (a purely unbounded read would freeze `run create`
/// forever behind a stuck `claim_pid_atomic` flock). `run create` turns every
/// `Unconfirmed` into a loud `supervisor_spawn_failed`; lenient callers never
/// use this path.
pub fn spawn_for_run(paths: &RunPaths, run_id: &str) -> Result<SupervisorSpawn, CliError> {
    let log_path = paths.root.join("supervisor.stderr.log");

    // Readiness pipe first, so its write-end fd can be uncloaked in `pre_exec`
    // (CLOEXEC cleared only for the intended grandchild — see `apply_detach`).
    let mut pipe = match ReadinessPipe::new() {
        Ok(p) => p,
        Err(e) => {
            // Falling back to a pid-file poll here would reintroduce the very
            // timeout ambiguity this fix removes; a pipe() failure is a real,
            // rare resource exhaustion worth surfacing loudly instead.
            let reason = format!("failed to create supervisor readiness pipe: {e}");
            append_spawn_diag(&log_path, &reason);
            return Ok(SupervisorSpawn::Unconfirmed { reason });
        }
    };
    let write_fd = pipe.write_fd();
    let mut cmd = detached_supervise_command(run_id, &log_path, Some(write_fd))?;
    cmd.env(ENV_READINESS_FD, write_fd.to_string());

    if let Err(e) = spawn_and_reap(&mut cmd, run_id) {
        append_spawn_diag(
            &log_path,
            &format!("fork/exec of supervisor failed: {}", e.message),
        );
        return Err(e);
    }

    // The parent must stop being a writer, or `await_ready` never observes EOF
    // on the grandchild's death. After this, the only remaining write-end copy
    // is the grandchild's (the double-fork intermediate already `_exit`ed and
    // was reaped by `spawn_and_reap`).
    pipe.close_write();

    match pipe.await_ready(ready_wait()) {
        Readiness::Ready { pid } => Ok(SupervisorSpawn::Confirmed { pid }),
        Readiness::Died => {
            let reason = "supervisor died during init: the readiness pipe closed \
                          without a ready signal (see supervisor.stderr.log)"
                .to_string();
            append_spawn_diag(&log_path, &reason);
            Ok(SupervisorSpawn::Unconfirmed { reason })
        }
        Readiness::Timeout => {
            let reason = format!(
                "supervisor did not confirm boot within {:?}: it is wedged during init \
                 (e.g. blocked acquiring the run lock) — check for a stuck supervisor holding \
                 the run flock, then inspect supervisor.stderr.log",
                ready_wait()
            );
            append_spawn_diag(&log_path, &reason);
            Ok(SupervisorSpawn::Unconfirmed { reason })
        }
        Readiness::Error { code, message } => {
            let reason = if message.is_empty() {
                format!("supervisor reported a boot error: {code}")
            } else {
                format!("supervisor reported a boot error: {code}: {message}")
            };
            append_spawn_diag(&log_path, &reason);
            Ok(SupervisorSpawn::Unconfirmed { reason })
        }
        Readiness::Malformed(raw) => {
            let reason = format!(
                "supervisor sent a malformed/truncated readiness signal ({raw:?}); \
                 treating boot as failed (see supervisor.stderr.log)"
            );
            append_spawn_diag(&log_path, &reason);
            Ok(SupervisorSpawn::Unconfirmed { reason })
        }
    }
}