//! @arch:layer(kg_store)
//! @arch:role(substrate)
//! @arch:see(.yah/docs/working/yah-task-runs.md)
//!
//! PTY subprocess driver — spawn commands, capture output as append-only
//! chunks, handle SIGTERM/SIGKILL with a grace period, and mark stale
//! `Running` runs as `Lost` when the daemon restarts.
//!
//! ## Tier 2 side-channel (yah-log shims)
//!
//! When `SpawnOpts::log_fd_enabled` is true (the default), the driver creates
//! a named pipe (FIFO) and exports two env vars into the child:
//!
//! - `YAH_TASK_RUN` — the `TaskRunId` as a hyphenated UUID string.
//! - `YAH_LOG_PIPE` — absolute path to the FIFO.
//!
//! The child opens `YAH_LOG_PIPE` for writing and emits JSON-lines. The
//! driver reads those lines in a background thread and stores them as
//! [`EventSource::Shim`] events.
//!
//! **Why FIFO instead of a raw fd?** `portable-pty` calls `close_random_fds()`
//! in its `pre_exec` hook, closing every fd ≥ 3 before exec. A raw-pipe write
//! fd is always ≥ 3 and would be closed before the child could use it. Opening
//! a FIFO by path requires no fd inheritance.
//!
//! Wire format — one JSON object per line:
//! ```json
//! {"level":"info","target":"myapp::module","msg":"text","fields":{"key":"val"}}
//! ```
//! Optional shim-identity keys: `"_lib"` (string), `"_lib_ver"` (string).
//! Unknown keys in `fields` pass through as freeform JSON.
//!
//! The driver holds the write end of the FIFO open until the run lifecycle
//! task completes, which triggers EOF for the receiver thread. The FIFO file
//! is deleted after the receiver thread drains the last line.
//!
//! On non-Unix platforms `YAH_TASK_RUN` and `YAH_LOG_PIPE` are not exported.
//! Shim libraries must treat absent `YAH_TASK_RUN` as "not inside a TaskRun".
//!
//! @yah:ticket(R617-F6, "Reattach-by-run_id replaces Lost-on-disappear for origin=terminal shells")
//! @yah:status(review)
//! @yah:assignee(agent:bundle-anthropic-ashguard)
//! @yah:at(2026-07-24T01:26:41Z)
//! @yah:phase(P3)
//! @yah:parent(R617)
//! @arch:see(.yah/docs/working/W280-durable-terminal-sessions.md)
//! @yah:depends_on(R617-F13)
//! @yah:handoff("DELIVERED. Verified: `cd oss/qed && cargo test -p task-runs --lib` 243/243 (was 237 — 6 new); `cargo test -p kg-daemon --lib shell_vt` 9/9; `cargo test -p yah --lib r617` 9/9; `cargo test -p desktop --lib` 357 pass / 2 fail, both pre-existing and in files this ticket does not touch (agent.rs rules-view expects 12 rows and a peer's approval-rule change makes 19; agent_process reader-finished is a known timing flake).")
//! @yah:handoff("THE TICKET'S OWN FRAMING WAS WRONG ABOUT THE MECHANISM, and the correction is the design. @yah:next said to 're-adopt' a live shell by 'control channel rebuilt, reader thread restarted against the surviving PTY'. That is not possible and never was: you cannot re-open another process's PTY master fd. The real defect is narrower and worse — a driver was tombstoning runs IT DID NOT OWN. `.yah/db/task-runs.turso` has several writers (desktop, the R617-F13 shell host, one CampService per MCP sidecar), and `TaskDriver::new` assumed any leftover `Running` row must be its own predecessor's corpse. So every attach marked some other LIVE process's shell `Lost`, and that shell kept producing output under a status saying it was dead. The fix is therefore 'do not tombstone what you do not own', not 'reattach'. Actual PTY reattach is unnecessary once F13 puts the PTY in a process that outlives the desktop.")
//! @yah:handoff("HOW OWNERSHIP IS KNOWN: new `TaskRunMeta::host_pid` — the pid of the process whose driver spawned the run, NOT the child's. Stamped by `spawn_run` at INSERT, before the child exists, so a crash between insert and spawn still leaves the row attributable. Store column added by the same idempotent `ALTER TABLE ... ADD COLUMN` pattern `origin` used, and `row_to_meta` reads index 15 with `.ok().flatten()` so a DB with no such column reads `None` rather than erroring.")
//! @yah:handoff("THE SEAM IS ORIGIN-AGNOSTIC, per this ticket's gotcha. New `task_runs::StaleRunPolicy` in oss/qed/crates/task-runs/src/driver.rs: `LostOnDisappear` (the default — `TaskDriver::new` and `with_channels` behave exactly as before, so no existing embedder changed) and `AdoptLiveHosts { origins: Vec<String> }`, which spares a leftover run only when its `host_pid` names a process that still exists. The crate decides on OWNERSHIP and takes the origin list as data — it never learns what 'terminal' means. New `TaskDriver::with_config` is the constructor that takes it.")
//! @yah:handoff("yah side: `crates/yah/kg-daemon/src/service.rs::open_task_store` now passes `AdoptLiveHosts { origins: [ORIGIN_TERMINAL] }`. Also replaced the magic string — new `kg_daemon::shell_vt::ORIGIN_TERMINAL` now backs the two live `origin == \"terminal\"` gates in shell_vt.rs plus the policy, so the VT-parsing gate and the tombstone-exemption gate cannot drift apart by a typo. The constant lives on the yah side, NOT in task-runs, precisely to keep the crate generic.")
//! @yah:handoff("Also stamped at app/yah/desktop/src/terminal.rs:519 — the desktop-local PTY path (terminal_open_local's scrollback mint) owns its own PTYs, so those rows carry the desktop's pid. Without it the shell host's driver would tombstone a live desktop-local session on attach, which is the same bug pointing the other way.")
//! @yah:handoff("PID REUSE is the honest weakness and is why the policy is opt-in and origin-narrowed. `kill(pid, 0)` (EPERM counts as alive — the process exists, it is just not ours to signal) can read a recycled pid as the original owner. The failure mode of a false 'alive' is one run left `Running` until something closes it; the false 'dead' this replaces kills a live session's status. Strictly the better direction for an interactive shell, and the exposure is bounded to origins the embedder opted in. Non-unix has no kill(2), so `host_process_alive` reports false there and the platform keeps the old behaviour rather than stranding runs forever.")
//! @yah:handoff("SIX NEW TESTS, each pinned to a failure rather than a code path: a live-owner terminal run survives a new driver (the ticket's whole point); a run whose owner pid was spawned and reaped in-test IS tombstoned (a crashed host must not leave zombie tiles); origin-less and non-matching origins are tombstoned even with a live owner (an in-flight `cargo build` whose driver is gone has nobody left to record its exit); an unattributed row (pre-migration) is tombstoned; `TaskDriver::new` still tombstones unconditionally (no silent behaviour change for existing embedders); and `spawn_run` stamps this process — the policy is worthless if rows arrive unattributed.")
//! @yah:verify("cd oss/qed && cargo test -p task-runs --lib # 243/243, 6 new under driver::tests")
//! @yah:verify("cargo test -p kg-daemon --lib shell_vt # 9/9")
//! @yah:verify("cargo test -p yah --lib r617 # 9/9")
//! @yah:verify("Manual (needs a desktop rebuild): open a shell, run `sleep 300`, quit and relaunch the desktop — the run is still Running, not Lost")
//! @yah:verify("sqlite3 .yah/db/task-runs.turso \"select id, origin, host_pid, status from runs where status='running';\" # every live row names a pid that ps shows")
//! @yah:gotcha("This is an oss/qed crate — changes land in-tree under oss/qed/crates/task-runs and flow outward via scripts/export-oss.sh. The seam was kept origin-agnostic (StaleRunPolicy decides on host_pid, takes origins as data); the one yah-ism, ORIGIN_TERMINAL, lives in crates/yah/kg-daemon/src/shell_vt.rs instead.")
//! @yah:gotcha("`host_pid` is NOT on the wire. rpc::WireRunMeta does not carry it, so a client cannot ask 'is this run's owner alive'. Nothing needs it today — the policy runs entirely daemon-side — but R617-F7 should check whether reattaching tiles want it before adding a second liveness notion of their own.")
//! @yah:gotcha("pid reuse can make a dead owner read alive, leaving a run `Running` with nobody driving it. Bounded on purpose (opt-in + origin-narrowed) and strictly safer than the false-dead it replaces, but it is a real edge: if zombie terminal rows ever accumulate, this is why.")
//! @yah:gotcha("TaskRunMeta gained a required field, so every struct-literal construction site had to be updated (velveteen-exec x4, scryer, task-runs fixtures, kg-daemon fixtures, desktop/terminal.rs x2). A new construction site added by anyone else will fail to compile until they pick a value — which is the intended forcing function: a run with no recorded owner is a run the policy has to tombstone.")
//!
//! @yah:ticket(R617-B9, "Pre-existing: task-runs log_pipe_events_land_in_store never completes (233 pass / 1 fail)")
//! @yah:status(review)
//! @yah:assignee(agent:bundle-anthropic-ashguard)
//! @yah:at(2026-07-22T19:50:25Z)
//! @yah:phase(P1)
//! @yah:parent(R617)
//! @yah:handoff("Root cause: not the FIFO, not the PTY. The whole pipeline completed correctly every time (child wrote the JSON line, receiver drained it, reader hit EOF, child.wait returned 0) — but the lifecycle's terminal `store.update_status` returned `Sql(Busy(\"database is locked\"))` and run_lifecycle swallowed it with `let _ =`, so the run stayed Running forever and the 20s poll deadline blew. A live run has three concurrent turso writers (PTY chunk appends, shim-FIFO event appends, lifecycle status) on independent connections with no busy handling at all.")
//! @yah:handoff("Fix in oss/qed/crates/task-runs/src/store.rs: (1) `conn()` now sets `busy_timeout(5s)` on every connection; (2) new `exec_retry()` wraps writes in an outer exponential-backoff retry on the `Busy`/`BusySnapshot` class, because turso caps its internal backoff and then hands `Busy` back; (3) insert_run / update_status / update_beholder_status / append_chunk / append_event all routed through it.")
//! @yah:handoff("driver.rs run_lifecycle no longer swallows the terminal status write — a genuine failure after retries now prints `[yah task-runs] failed to record terminal status for run <id>`, matching the crate's existing eprintln convention.")
//! @yah:handoff("New regression test store.rs::concurrent_writers_do_not_lose_the_terminal_status — two background tasks hammer append_chunk/append_event while update_status lands. Verified it has teeth: with busy_timeout and the retry disabled it fails 3/3 with the exact `Busy(\"database is locked\")`; with them it passes 5/5.")
//! @yah:verify("cd oss/qed && cargo test -p task-runs --lib — 237 passed / 0 failed (was 235 pass / 1 fail)")
//! @yah:verify("log_pipe_events_land_in_store run 8x sequentially: 8/8 green in ~0.58s each. Before the fix the same loop was 11/12 red at the 20s timeout.")
//!
//! @yah:ticket(R652-T6, "Login shell: when cmd is the resolved shell, exec it directly (not sh -c) with -l")
//! @yah:at(2026-08-02T00:03:08Z)
//! @yah:status(review)
//! @yah:assignee(agent:bundle-ollama-cloud-boulder)
//! @yah:phase(P1)
//! @yah:parent(R652)
//! @yah:handoff("Login shells now exec directly with -l instead of going through sh -c. SpawnOpts (oss/qed/crates/task-runs/src/driver.rs) gained `argv: Option<Vec<String>>`: when set, spawn_run builds the CommandBuilder from that argv verbatim instead of wrapping `cmd` in `sh -c`. camp-service task_run sets it to [resolved_shell, \"-l\"] whenever the request is a shell request.")
//! @yah:handoff("Why an argv escape hatch rather than a `login_shell: bool` flag in the driver: task-runs is an oss/qed crate and has no business knowing what a login shell is. The caller names the exact process; the driver just execs it. This also made R652-T4 a two-line addition rather than a second flag.")
//! @yah:handoff("Three things this fixes beyond .zprofile finally running. (1) `sh -c \"zsh -l\"` left an inert `sh` as the PTY's foreground process group leader, so job control misbehaved and signals went to the wrong process. (2) That same inert sh is what the foreground-pid cwd probe (R652-T2) would have reported for, so T2 could not have worked without this. (3) -l is now a real argv element instead of text inside a shell string, so no quoting layer can eat it.")
//! @yah:handoff("`cmd` is still what lands on TaskRunMeta.command, so a shell run reads back as \"$SHELL\" -- the rail label and the history re-run path both keep working. Beholder argv rewriting is bypassed when argv is set (the attach runs with BeholderSelect::None): the rewritten argv would be discarded on that path, so recording a `rewrite=...` that never happened would be a lie in the run metadata.")
//! @yah:handoff("An empty argv falls back to the sh -c path rather than spawning nothing -- a caller bug should not become an exec of the empty string.")
//! @yah:verify("cd oss/qed && cargo test -p task-runs --lib # 246/246 green (3 new: explicit_argv_execs_the_program_directly, explicit_argv_still_records_the_requested_command, empty_argv_falls_back_to_the_shell_path)")
//! @yah:verify("Manual (needs desktop rebuild): add `echo W289-login-test >> /tmp/w289.log` to ~/.zprofile, open a shell tile, confirm the file gets a line")
//! @yah:gotcha("driver.rs is an oss/qed crate -- this lands in-tree under oss/qed/crates/task-runs and flows outward via scripts/export-oss.sh on the next release. SpawnOpts gained a field, but every in-tree construction site uses ..Default::default(), so nothing else needed touching.")
use std::collections::HashMap;
use std::io::Read;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use portable_pty::{native_pty_system, CommandBuilder, PtySize};
use thiserror::Error;
use tokio::sync::{mpsc, oneshot};
use tokio::task;
use crate::beholders::{registry_with_user_beholders, BeholderSelect};
use crate::store::{RunFilter, StoreError, TaskStore};
use crate::types::{BeholderStatus, Initiator, OutputChunk, RunStatus, Stream, TaskRunId, TaskRunMeta};
const DEFAULT_GRACE: Duration = Duration::from_secs(5);
const READ_BUF_SIZE: usize = 4096;
const SIGTERM: i32 = 15;
const SIGKILL: i32 = 9;
// ─── Error ────────────────────────────────────────────────────────────────────
#[derive(Debug, Error)]
pub enum DriverError {
#[error("store: {0}")]
Store(#[from] StoreError),
#[error("pty: {0}")]
Pty(String),
#[error("run not found: {0}")]
NotFound(String),
#[error("io: {0}")]
Io(#[from] std::io::Error),
}
// ─── SpawnOpts ────────────────────────────────────────────────────────────────
/// Options for [`TaskDriver::spawn_run`].
#[derive(Debug, Clone)]
pub struct SpawnOpts {
pub cwd: PathBuf,
/// Env vars set on the child process (merged on top of the current env).
pub env: Vec<(String, String)>,
pub label: Option<String>,
pub initiator: Initiator,
/// PTY column count. Defaults to 80.
pub pty_cols: u16,
/// PTY row count. Defaults to 24.
pub pty_rows: u16,
/// Enable stdin relay via [`TaskDriver::send_stdin`].
pub stdin_enabled: bool,
/// Pin the run so the GC sweep does not drop its output during warm rolloff.
pub pin: bool,
/// Beholder attachment policy. Defaults to [`BeholderSelect::Auto`].
pub beholder_select: BeholderSelect,
/// `true` when a human-facing terminal tile is attached. Causes `Rewriter`
/// beholders to decline in `Auto` mode so the human sees unmodified output.
pub tty_attached: bool,
/// Create a side-channel FIFO and export `YAH_TASK_RUN` / `YAH_LOG_PIPE`
/// so Tier-2 shim libraries (yah-log-rust, @yah/log) can emit structured
/// events. Has no effect on non-Unix platforms. Defaults to `true`.
pub log_fd_enabled: bool,
/// Provenance tag stored on the run's `TaskRunMeta.origin` (e.g.
/// `Some("terminal")` for an interactive shell). `None` is an ordinary job.
pub origin: Option<String>,
/// Exec this argv directly instead of wrapping `cmd` in `sh -c`.
///
/// The default `sh -c <cmd>` is right for a job — the caller wrote a
/// command line and expects a shell to parse it. It is wrong for an
/// *interactive shell*: `sh -c "zsh -l"` leaves an inert `sh` as the PTY's
/// foreground process group leader, so job control misbehaves, signals go
/// to the wrong process, and anything that reads the foreground pid (a
/// live-cwd probe, say) sees `sh` instead of the shell the operator is
/// typing into. Handing the exact argv here makes the shell itself the
/// child, which is also the only way to pass `-l` as a real argv element
/// so `.zprofile` / `.profile` actually run.
///
/// `cmd` is still what gets recorded on `TaskRunMeta.command`, so the run
/// reads the way the caller asked for it. Beholder argv rewriting is
/// bypassed when this is set: the caller has already decided the exact
/// process to exec, and a recorded `rewrite=…` that didn't happen would be
/// a lie in the run metadata.
pub argv: Option<Vec<String>>,
}
impl Default for SpawnOpts {
fn default() -> Self {
Self {
cwd: std::env::current_dir().unwrap_or_else(|_| PathBuf::from("/")),
env: vec![],
label: None,
initiator: Initiator::Human { camp: "local".to_string() },
pty_cols: 80,
pty_rows: 24,
stdin_enabled: false,
pin: false,
beholder_select: BeholderSelect::Auto,
tty_attached: false,
log_fd_enabled: true,
origin: None,
argv: None,
}
}
}
// ─── Driver channels ─────────────────────────────────────────────────────────
/// Optional side-channels a driver can publish to. Both are fire-and-forget:
/// a closed receiver never stalls or fails a run.
#[derive(Default)]
pub struct DriverChannels {
/// Fires `(run_id, status)` after each run's lifecycle task writes the
/// terminal status. Drives completion listeners (e.g. a triage worker).
pub completion: Option<mpsc::UnboundedSender<(TaskRunId, RunStatus)>>,
/// Mirrors every PTY output chunk as it is captured, *before* any consumer
/// polls the store. Lets a host attach a live view (VT parser, log
/// forwarder) to a run without a read-back loop over the store.
///
/// The driver deliberately stays ignorant of what the tap is for — the
/// chunk carries `run_id`, so the host decides which runs it cares about.
pub output: Option<mpsc::UnboundedSender<OutputChunk>>,
}
// ─── Stale-run policy ────────────────────────────────────────────────────────
/// What a freshly-constructed [`TaskDriver`] does with `Running` rows it finds
/// already in the store.
///
/// The historical rule — tombstone every one of them — bakes in an assumption
/// that stops being true the moment a second process attaches to the same
/// store: that any `Running` row must be a corpse from *this* process's
/// predecessor. When two processes share a store, a driver starting up in one
/// will happily mark the other's live runs `Lost`, and the run keeps producing
/// output under a status that says it is dead.
///
/// The policy is deliberately origin-agnostic in its mechanism — it decides on
/// **who owns the run** ([`TaskRunMeta::host_pid`]) — and takes the origin list
/// as data, so an embedder names the runs it wants exempted without this crate
/// knowing what any of them mean.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub enum StaleRunPolicy {
/// Tombstone every leftover `Running` run as `Lost`.
///
/// Correct, and the default, whenever this process is the only writer:
/// a run whose driver is gone has no one left to notice it exit.
#[default]
LostOnDisappear,
/// Spare runs whose recorded owner process is still alive.
///
/// A leftover run is tombstoned only when its `host_pid` is absent (owner
/// unknown — a row from before the column existed) or names a process that
/// no longer exists. Anything else belongs to a live peer and is left
/// `Running` for that peer to finish.
///
/// `origins` narrows the exemption to runs whose
/// [`TaskRunMeta::origin`] is in the list; empty means every origin
/// qualifies. A run with no origin never matches a non-empty list.
AdoptLiveHosts { origins: Vec<String> },
}
impl StaleRunPolicy {
/// Whether `meta` should be tombstoned `Lost` at driver construction.
fn tombstones(&self, meta: &TaskRunMeta) -> bool {
match self {
StaleRunPolicy::LostOnDisappear => true,
StaleRunPolicy::AdoptLiveHosts { origins } => {
let exempt_origin = origins.is_empty()
|| meta
.origin
.as_deref()
.is_some_and(|o| origins.iter().any(|want| want == o));
if !exempt_origin {
return true;
}
match meta.host_pid {
Some(pid) => !host_process_alive(pid),
None => true,
}
}
}
}
}
/// Is a process with this pid still around?
///
/// `kill(pid, 0)` is the portable liveness probe: it performs the permission
/// check and existence lookup without delivering anything. `EPERM` counts as
/// alive — the process exists, it just is not ours to signal.
///
/// Pid reuse can make a dead owner read as alive. That is why
/// [`StaleRunPolicy::AdoptLiveHosts`] is opt-in and origin-narrowed: the cost
/// of a false "alive" is one run left `Running` until something closes it,
/// which is strictly better for an interactive session than the false "dead"
/// this replaces — which kills a *live* session's status.
#[cfg(unix)]
fn host_process_alive(pid: u32) -> bool {
if pid == 0 {
return false;
}
if pid == std::process::id() {
return true;
}
// SAFETY: `kill` with signal 0 delivers nothing; it only reports whether
// the pid exists and is signallable.
let rc = unsafe { libc::kill(pid as libc::pid_t, 0) };
rc == 0 || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM)
}
/// No `kill(2)` off Unix. Reporting every owner dead keeps the historical
/// Lost-on-disappear behaviour rather than stranding runs `Running` forever.
#[cfg(not(unix))]
fn host_process_alive(_pid: u32) -> bool {
false
}
// ─── Internal run-control handle ─────────────────────────────────────────────
struct RunControl {
kill_tx: mpsc::Sender<KillRequest>,
stdin_tx: Option<mpsc::Sender<Vec<u8>>>,
/// Shared with the lifecycle task, which holds the same `Arc` so the PTY fd
/// outlives `child.wait()`. `MasterPty::resize` takes `&self`, so a mutex is
/// enough to make the `Box<dyn MasterPty + Send>` `Sync` across the two.
master: Arc<Mutex<Box<dyn portable_pty::MasterPty + Send>>>,
}
#[derive(Debug)]
struct KillRequest {
signal: i32,
}
// ─── ShimRecord ───────────────────────────────────────────────────────────────
/// One JSON-line record emitted by a Tier-2 shim to the side-channel FIFO.
///
/// The shim (Rust `yah-log` layer or TS `@yah/log` pino transport) writes one
/// of these per log call. Unknown keys inside `fields` pass through unchanged.
#[cfg(unix)]
#[derive(serde::Deserialize)]
struct ShimRecord {
level: String,
target: String,
msg: String,
#[serde(default)]
fields: serde_json::Value,
/// Shim library name, e.g. `"yah-log-rust"`. Populates
/// [`EventSource::Shim::lib`].
#[serde(rename = "_lib", default)]
lib: Option<String>,
/// Shim library version string.
#[serde(rename = "_lib_ver", default)]
lib_version: Option<String>,
}
// ─── FdCloser ─────────────────────────────────────────────────────────────────
/// RAII wrapper that closes a raw fd on drop.
///
/// Used to hold the write end of the log FIFO open until the lifecycle task
/// completes. Dropping it signals EOF to the receiver thread.
#[cfg(unix)]
struct FdCloser(libc::c_int);
#[cfg(unix)]
impl Drop for FdCloser {
fn drop(&mut self) {
unsafe { libc::close(self.0) };
}
}
// SAFETY: a raw fd number is an integer; closing it from any thread is safe
// provided we never duplicate ownership (enforced by move semantics here).
#[cfg(unix)]
unsafe impl Send for FdCloser {}
// ─── TaskDriver ───────────────────────────────────────────────────────────────
/// Manages in-flight task runs for a single camp.
///
/// Wrap in `Arc` to share across tasks; internal state is mutex-protected.
pub struct TaskDriver {
store: Arc<TaskStore>,
active: Arc<Mutex<HashMap<String, RunControl>>>,
/// Side-channels published to by every run this driver owns.
channels: DriverChannels,
}
impl TaskDriver {
/// Create a driver backed by `store`, with no side-channels.
///
/// Immediately scans the store for `Running` runs left over from a prior
/// daemon process and marks them `Lost` ("Lost-on-disappear").
pub async fn new(store: Arc<TaskStore>) -> Result<Self, DriverError> {
Self::with_channels(store, DriverChannels::default()).await
}
/// Like `new` but wires the optional [`DriverChannels`] side-channels
/// (completion notifications, live output tap).
pub async fn with_channels(
store: Arc<TaskStore>,
channels: DriverChannels,
) -> Result<Self, DriverError> {
Self::with_config(store, channels, StaleRunPolicy::default()).await
}
/// Full constructor: side-channels plus the [`StaleRunPolicy`] applied to
/// `Running` rows already in the store.
///
/// R617-F6 — annotation in this file's header. Splitting the sweep out of
/// the constructor's fixed behaviour is what lets a store be shared: a
/// process that is not the run's owner can now attach without declaring
/// the owner's live work dead.
pub async fn with_config(
store: Arc<TaskStore>,
channels: DriverChannels,
stale_policy: StaleRunPolicy,
) -> Result<Self, DriverError> {
let stale = store
.list_runs(&RunFilter {
status: Some("running".to_string()),
..Default::default()
})
.await?;
for meta in stale {
if !stale_policy.tombstones(&meta) {
continue;
}
store
.update_status(
&meta.id,
&RunStatus::Lost {
reason: "daemon restarted while run was in-flight".to_string(),
},
)
.await?;
}
Ok(Self {
store,
active: Arc::new(Mutex::new(HashMap::new())),
channels,
})
}
/// Spawn `cmd` in a PTY and start capturing its output. Returns immediately
/// with the new [`TaskRunId`].
///
/// A beholder is selected via `opts.beholder_select` (default `Auto`). When
/// a `Rewriter` beholder matches, its `adjust_argv` is applied to the
/// command before spawning and the diff is recorded on `beholder_status`.
/// When `opts.tty_attached` is `true`, `Rewriter` beholders decline in
/// `Auto` mode to preserve human-readable output.
///
/// Output is written to the store as `Stream::Stdout` chunks (the PTY
/// kernel merges stdout and stderr). Signal handling and status updates
/// run in background tasks.
pub async fn spawn_run(&self, cmd: &str, opts: SpawnOpts) -> Result<TaskRunId, DriverError> {
let id = TaskRunId::new();
let started_at = unix_now_secs();
let started_at_ms: u64 = started_at.saturating_mul(1000);
// Attach a beholder (may rewrite argv and produce structured events).
// Resolve user drop-in directory: $YAH_BEHOLDERS_DIR or $HOME/.yah/beholders.
let user_dir = std::env::var_os("YAH_BEHOLDERS_DIR")
.map(std::path::PathBuf::from)
.or_else(|| {
std::env::var_os("HOME")
.map(|h| std::path::PathBuf::from(h).join(".yah/beholders"))
});
let registry = registry_with_user_beholders(user_dir.as_deref());
/* An explicit argv means the caller already chose the exact process
(an interactive login shell, say). Selecting a beholder there would
either do nothing — the rewritten argv is discarded on that path —
or record a rewrite that never happened, so we opt out honestly
instead. */
let select = if opts.argv.is_some() {
&BeholderSelect::None
} else {
&opts.beholder_select
};
let attach = registry.attach(cmd, select, opts.tty_attached);
// Reconstruct the command from argv ONLY when a beholder actually
// rewrote it. `AttachResult.argv` is always populated — it is
// `resolve_argv(cmd)` even when nothing attached — so joining it
// unconditionally ran every run's command through a whitespace
// normalization nobody asked for: runs of spaces collapse and embedded
// newlines become spaces, which is silent corruption for a heredoc or
// any multi-line line. The caller's bytes go to the shell untouched
// unless a rewrite is the whole point.
let effective_cmd = match &attach.status.rewrite_added {
Some(added) if !added.is_empty() && !attach.argv.is_empty() => attach.argv.join(" "),
_ => cmd.to_string(),
};
self.store.insert_run(&TaskRunMeta {
id: id.clone(),
command: cmd.to_string(),
cwd: opts.cwd.clone(),
env: opts.env.clone(),
started_at,
status: RunStatus::Running,
label: opts.label.clone(),
initiator: opts.initiator.clone(),
beholder_status: Some(attach.status),
pinned: opts.pin,
origin: opts.origin.clone(),
/* R617-F6: stamp the OWNER, before the child exists. Written at
insert rather than after spawn so a crash between the two still
leaves the row attributable — an unattributed `Running` row is
exactly what the conservative arm of `StaleRunPolicy` has to
tombstone. */
host_pid: Some(std::process::id()),
}).await?;
// Open PTY pair.
let pty_sys = native_pty_system();
let pair = pty_sys
.openpty(PtySize {
rows: opts.pty_rows,
cols: opts.pty_cols,
pixel_width: 0,
pixel_height: 0,
})
.map_err(|e| DriverError::Pty(e.to_string()))?;
// Clone reader before spawning so the fd is ready immediately.
let pty_reader = pair
.master
.try_clone_reader()
.map_err(|e| DriverError::Pty(e.to_string()))?;
// Optional stdin relay: take the writer before spawning the child.
let stdin_tx: Option<mpsc::Sender<Vec<u8>>> = if opts.stdin_enabled {
let mut writer = pair
.master
.take_writer()
.map_err(|e| DriverError::Pty(e.to_string()))?;
let (tx, mut rx) = mpsc::channel::<Vec<u8>>(64);
task::spawn(async move {
use std::io::Write;
while let Some(bytes) = rx.recv().await {
let _ = writer.write_all(&bytes);
let _ = writer.flush();
}
});
Some(tx)
} else {
None
};
// ── Side-channel log FIFO (Tier 2 / yah-log shims) ──────────────────
//
// Create a named pipe (FIFO) so child processes can write structured
// events without touching stdout/stderr. We export its path via
// YAH_LOG_PIPE; no fd inheritance is involved, so portable-pty's
// close_random_fds() pre_exec hook doesn't interfere.
//
// The parent opens the FIFO twice:
// rfd — O_RDONLY|O_NONBLOCK, then cleared to blocking → read events
// wfd — O_WRONLY (wrapped in FdCloser) → keeps the FIFO alive until
// the lifecycle task drops it (after run completion), producing
// EOF for the receiver thread.
#[cfg(unix)]
let log_fifo: Option<(libc::c_int, FdCloser, std::path::PathBuf)> = if opts.log_fd_enabled {
let fifo_path = std::env::temp_dir().join(format!("yah-log-{}.fifo", id));
let path_cstr = match std::ffi::CString::new(fifo_path.to_string_lossy().as_bytes()) {
Ok(s) => s,
Err(_) => {
// Path contained a nul byte — extremely unlikely; skip FIFO.
return Err(DriverError::Io(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"log FIFO path contained nul byte",
)));
}
};
let mkfifo_ret = unsafe { libc::mkfifo(path_cstr.as_ptr(), 0o600) };
if mkfifo_ret != 0 {
None // FIFO creation failed; continue without side-channel
} else {
// Open read end without blocking (no writer yet).
let rfd = unsafe {
libc::open(path_cstr.as_ptr(), libc::O_RDONLY | libc::O_NONBLOCK)
};
if rfd < 0 {
let _ = unsafe { libc::unlink(path_cstr.as_ptr()) };
None
} else {
// Switch read end to blocking so reads yield proper data.
unsafe { libc::fcntl(rfd, libc::F_SETFL, 0) };
// Open write end — this succeeds immediately because rfd is open.
let wfd = unsafe {
libc::open(path_cstr.as_ptr(), libc::O_WRONLY)
};
if wfd < 0 {
unsafe { libc::close(rfd) };
let _ = unsafe { libc::unlink(path_cstr.as_ptr()) };
None
} else {
Some((rfd, FdCloser(wfd), fifo_path))
}
}
}
} else {
None
};
// Build and spawn the child inside the slave. An explicit argv execs
// that program directly; otherwise the command line goes through `sh`
// so the caller's quoting, pipes and redirections mean what they say.
let mut cb = match opts.argv.as_deref() {
Some([program, args @ ..]) => {
let mut cb = CommandBuilder::new(program);
cb.args(args);
cb
}
// An empty argv is a caller bug, not a request for an empty exec —
// fall back to the shell path rather than spawning nothing.
_ => {
let mut cb = CommandBuilder::new("sh");
cb.args(["-c", &effective_cmd]);
cb
}
};
cb.cwd(&opts.cwd);
for (k, v) in &opts.env {
cb.env(k, v);
}
cb.env("TERM", "xterm-256color");
// Export YAH_TASK_RUN and YAH_LOG_PIPE if the FIFO was created.
#[cfg(unix)]
if let Some((_, _, ref fifo_path)) = log_fifo {
cb.env("YAH_TASK_RUN", id.to_string());
cb.env("YAH_LOG_PIPE", fifo_path.to_string_lossy().as_ref());
}
let child = pair
.slave
.spawn_command(cb)
.map_err(|e| DriverError::Pty(e.to_string()))?;
// Drop the parent's slave handle so EOF propagates once the child exits.
drop(pair.slave);
// Share the master between the lifecycle task (which must outlive
// `child.wait()` so the fd stays open) and `resize_run`.
let master: Arc<Mutex<Box<dyn portable_pty::MasterPty + Send>>> =
Arc::new(Mutex::new(pair.master));
let pid = child.process_id().unwrap_or(0);
// ── FIFO: launch receiver thread; pass write-end holder to lifecycle ──
//
// The receiver thread reads until EOF. EOF arrives when ALL write-end
// holders close: the child's own writers (when it exits) plus the
// FdCloser we hand to the lifecycle task (which drops it after writing
// the terminal RunStatus). Events written before the last close are
// still drained by the receiver thread before it exits.
#[cfg(unix)]
let log_wfd_holder: Option<FdCloser> = if let Some((rfd, wfd, fifo_path)) = log_fifo {
let store_log = Arc::clone(&self.store);
let id_log = id.clone();
let rt = tokio::runtime::Handle::current();
// spawn_blocking: lets the runtime track this thread so the
// Handle::block_on calls inside have a worker to drive futures.
tokio::task::spawn_blocking(move || {
run_log_receiver(rt, store_log, id_log, rfd, fifo_path, started_at_ms);
});
Some(wfd)
} else {
None
};
// Channels.
let (kill_tx, kill_rx) = mpsc::channel::<KillRequest>(4);
let (reader_done_tx, reader_done_rx) = oneshot::channel::<()>();
// Reader thread: PTY output → store chunks → beholder events.
// Runs on a dedicated OS thread because PTY reads are blocking.
{
let store_r = Arc::clone(&self.store);
let id_r = id.clone();
let mut beholder = attach.beholder;
let output_tx = self.channels.output.clone();
let rt = tokio::runtime::Handle::current();
tokio::task::spawn_blocking(move || {
let mut buf = [0u8; READ_BUF_SIZE];
let mut reader = pty_reader;
loop {
match reader.read(&mut buf) {
Ok(0) | Err(_) => break,
Ok(n) => {
let offset = elapsed_ms(started_at_ms);
let append_res = rt.block_on(store_r.append_chunk(
&id_r,
offset,
Stream::Stdout,
&buf[..n],
));
if let Ok(seq) = append_res {
/* Both the tap and the beholder want the same
owned chunk; build it once, and only when
someone is listening. */
let chunk = (output_tx.is_some() || beholder.is_some()).then(|| {
OutputChunk {
run_id: id_r.clone(),
seq,
offset_ms: offset,
stream: Stream::Stdout,
bytes: buf[..n].to_vec(),
}
});
/* Tap first: it feeds live views, where latency
is visible to a human. Send failure means the
host dropped its receiver — never fatal. */
if let (Some(tx), Some(c)) = (&output_tx, &chunk) {
let _ = tx.send(c.clone());
}
let mut detach_beholder = false;
if let (Some(b), Some(chunk)) = (beholder.as_mut(), &chunk) {
for ev in b.parse_chunk(chunk) {
let _ = rt.block_on(store_r.append_event(
&ev.run_id,
ev.offset_ms,
ev.level,
&ev.target,
&ev.msg,
&ev.fields,
ev.anchor.as_ref().map(|a| a.seq),
&ev.source,
));
}
if let Some(reason) = b.unknown_format_reason() {
let new_status = BeholderStatus::unknown_format_with_reason(
b.name(),
reason,
);
let _ = rt.block_on(
store_r.update_beholder_status(&id_r, &new_status),
);
detach_beholder = true;
}
}
if detach_beholder {
beholder = None;
}
}
}
}
}
if let Some(ref mut b) = beholder {
let final_offset = elapsed_ms(started_at_ms);
for ev in b.on_done(&id_r, final_offset) {
let _ = rt.block_on(store_r.append_event(
&ev.run_id,
ev.offset_ms,
ev.level,
&ev.target,
&ev.msg,
&ev.fields,
ev.anchor.as_ref().map(|a| a.seq),
&ev.source,
));
}
if let Some(reason) = b.unknown_format_reason() {
let new_status = BeholderStatus::unknown_format_with_reason(b.name(), reason);
let _ = rt.block_on(store_r.update_beholder_status(&id_r, &new_status));
}
}
let _ = reader_done_tx.send(());
});
}
// Lifecycle task: monitor kill requests, wait for exit, update status.
// The task also holds the log FIFO write-end closer (if any) so that
// EOF propagates to the receiver thread after RunStatus is written.
{
let store_l = Arc::clone(&self.store);
let active_l = Arc::clone(&self.active);
let id_l = id.clone();
let master_l = Arc::clone(&master);
let completion_tx_l = self.channels.completion.clone();
#[cfg(unix)]
let wfd_l = log_wfd_holder;
task::spawn(async move {
run_lifecycle(
store_l,
active_l,
id_l,
pid,
child,
master_l,
kill_rx,
reader_done_rx,
completion_tx_l,
#[cfg(unix)]
wfd_l,
)
.await;
});
}
self.active
.lock()
.unwrap()
.insert(id.to_string(), RunControl { kill_tx, stdin_tx, master });
Ok(id)
}
/// Resize a running task's PTY and deliver `SIGWINCH` to the foreground
/// process group (portable-pty's `resize` does the ioctl, which is what
/// signals the child).
///
/// Returns `DriverError::NotFound` when the run is not active on this
/// driver instance — the same contract as [`TaskDriver::send_stdin`].
pub async fn resize_run(
&self,
id: &TaskRunId,
cols: u16,
rows: u16,
) -> Result<(), DriverError> {
let master = self
.active
.lock()
.unwrap()
.get(&id.to_string())
.map(|c| Arc::clone(&c.master));
match master {
Some(m) => {
let size = PtySize { rows, cols, pixel_width: 0, pixel_height: 0 };
m.lock()
.unwrap()
.resize(size)
.map_err(|e| DriverError::Pty(e.to_string()))
}
None => Err(DriverError::NotFound(id.to_string())),
}
}
/// The pid of the run's *foreground* process — the leader of the process
/// group the PTY currently gives the keyboard to.
///
/// For a shell tile that is the shell itself while it sits at a prompt,
/// and the command the operator is running while one is in flight. That
/// distinction is the whole point: asking the spawned child would report
/// the shell forever, so anything derived from this pid (a live cwd probe,
/// a "what is this pane doing" label) would answer for the wrong process.
///
/// `None` when the run is not active on this driver instance, or when the
/// platform has no notion of a foreground process group.
pub fn foreground_pid(&self, id: &TaskRunId) -> Option<u32> {
let master = self
.active
.lock()
.unwrap()
.get(&id.to_string())
.map(|c| Arc::clone(&c.master))?;
#[cfg(unix)]
{
let pid = master.lock().unwrap().process_group_leader()?;
u32::try_from(pid).ok()
}
#[cfg(not(unix))]
{
let _ = master;
None
}
}
/// Send `signal` to a running task. Defaults to SIGTERM (15).
///
/// For SIGTERM, the driver waits up to 5 seconds for the process to exit
/// before escalating to SIGKILL. Returns `DriverError::NotFound` if the
/// run is not active (already exited or launched on a different driver
/// instance).
pub async fn kill_run(&self, id: &TaskRunId, signal: Option<i32>) -> Result<(), DriverError> {
let kill_tx = self
.active
.lock()
.unwrap()
.get(&id.to_string())
.map(|c| c.kill_tx.clone());
match kill_tx {
Some(tx) => tx
.send(KillRequest { signal: signal.unwrap_or(SIGTERM) })
.await
.map_err(|_| DriverError::NotFound(id.to_string())),
None => Err(DriverError::NotFound(id.to_string())),
}
}
/// Write bytes to the stdin of a running task (requires `stdin_enabled`).
pub async fn send_stdin(&self, id: &TaskRunId, bytes: Vec<u8>) -> Result<(), DriverError> {
let stdin_tx = self
.active
.lock()
.unwrap()
.get(&id.to_string())
.and_then(|c| c.stdin_tx.clone());
match stdin_tx {
Some(tx) => tx
.send(bytes)
.await
.map_err(|_| DriverError::NotFound(id.to_string())),
None => Err(DriverError::NotFound(id.to_string())),
}
}
}
// ─── Log fd receiver ─────────────────────────────────────────────────────────
/// Read JSON-lines from the side-channel FIFO read end and store them as
/// [`EventSource::Shim`] events.
///
/// Runs on a dedicated OS thread; exits when the read end sees EOF. EOF
/// arrives after both the child process AND the lifecycle task have closed
/// their write ends of the FIFO. The FIFO file is deleted on exit.
#[cfg(unix)]
fn run_log_receiver(
rt: tokio::runtime::Handle,
store: Arc<TaskStore>,
run_id: TaskRunId,
read_fd: libc::c_int,
fifo_path: std::path::PathBuf,
started_at_ms: u64,
) {
use std::io::BufRead;
use std::os::unix::io::FromRawFd;
// SAFETY: `read_fd` is a valid, open FIFO fd handed exclusively to this
// thread. `File` takes ownership and closes the fd on drop.
let file = unsafe { std::fs::File::from_raw_fd(read_fd) };
let reader = std::io::BufReader::new(file);
for line in reader.lines() {
let line = match line {
Ok(l) => l,
Err(_) => break,
};
let trimmed = line.trim();
if trimmed.is_empty() {
continue;
}
let rec: ShimRecord = match serde_json::from_str(trimmed) {
Ok(r) => r,
Err(_) => continue, // skip malformed lines silently
};
let level = rec.level.parse::<crate::types::Level>().unwrap_or(crate::types::Level::Info);
let source = crate::types::EventSource::Shim {
lib: rec.lib.unwrap_or_else(|| "unknown".to_string()),
version: rec.lib_version.unwrap_or_else(|| "0.0.0".to_string()),
};
let fields = if rec.fields.is_object() {
rec.fields
} else {
serde_json::Value::Object(Default::default())
};
let offset = elapsed_ms(started_at_ms);
let _ = rt.block_on(store.append_event(
&run_id,
offset,
level,
&rec.target,
&rec.msg,
&fields,
None,
&source,
));
}
// Clean up the FIFO file now that the receiver has drained.
let _ = std::fs::remove_file(&fifo_path);
}
// ─── Lifecycle task ───────────────────────────────────────────────────────────
async fn run_lifecycle(
store: Arc<TaskStore>,
active: Arc<Mutex<HashMap<String, RunControl>>>,
id: TaskRunId,
pid: u32,
child: Box<dyn portable_pty::Child + Send>,
master: Arc<Mutex<Box<dyn portable_pty::MasterPty + Send>>>,
mut kill_rx: mpsc::Receiver<KillRequest>,
reader_done_rx: oneshot::Receiver<()>,
completion_tx: Option<tokio::sync::mpsc::UnboundedSender<(TaskRunId, RunStatus)>>,
// Holds the write end of the log FIFO open until this task completes.
// Dropping it produces EOF for the receiver thread, which happens after
// the terminal RunStatus is written below.
#[cfg(unix)]
_log_wfd: Option<FdCloser>,
) {
// Pin the reader-done future so it can be polled by reference in
// nested select! arms without consuming ownership.
let reader_done = async { reader_done_rx.await.ok(); };
tokio::pin!(reader_done);
let sent_signal: Option<i32>;
tokio::select! {
req = kill_rx.recv() => {
match req {
Some(KillRequest { signal }) => {
send_unix_signal(pid, signal);
if signal == SIGKILL {
sent_signal = Some(SIGKILL);
} else {
// Grace period: give the process a chance to exit cleanly.
tokio::select! {
_ = &mut reader_done => {
// Exited within grace — no SIGKILL needed.
sent_signal = Some(signal);
}
_ = tokio::time::sleep(DEFAULT_GRACE) => {
// Grace expired — escalate.
send_unix_signal(pid, SIGKILL);
sent_signal = Some(SIGKILL);
}
}
}
}
// kill_tx dropped (driver shutting down) — force kill.
None => {
send_unix_signal(pid, SIGKILL);
sent_signal = Some(SIGKILL);
}
}
}
_ = &mut reader_done => {
sent_signal = None;
}
}
// Reap the child (blocking) on a dedicated thread-pool slot.
// Move our master handle in here so the PTY fd outlives the wait. The
// matching `RunControl` (removed from `active` below) holds the other
// `Arc`, so the fd actually closes once both are gone.
let exit_code = task::spawn_blocking(move || {
let mut c = child;
let _m = master; // dropped after wait() returns
c.wait().ok().map(|s| s.exit_code())
})
.await
.ok()
.flatten();
let ended_at = unix_now_secs();
let status = match sent_signal {
Some(sig) => RunStatus::Killed { signal: sig, ended_at },
None => match exit_code {
Some(code) => RunStatus::Done { exit_code: code as i32, ended_at },
None => RunStatus::Lost {
reason: "process exited without an exit code".to_string(),
},
},
};
/* Losing this write is not cosmetic: the run stays `Running` in the store
forever and every reader — tail loops, the terminal UI, the next
daemon's Lost-on-disappear sweep — believes a dead process is alive.
`update_status` already retries through lock contention, so a failure
here is terminal and worth saying out loud. */
if let Err(e) = store.update_status(&id, &status).await {
eprintln!("[yah task-runs] failed to record terminal status for run {id}: {e}");
}
if let Some(ref tx) = completion_tx {
let _ = tx.send((id.clone(), status));
}
active.lock().unwrap().remove(&id.to_string());
}
// ─── Helpers ──────────────────────────────────────────────────────────────────
fn send_unix_signal(pid: u32, signal: i32) {
#[cfg(unix)]
unsafe {
libc::kill(pid as libc::pid_t, signal);
}
// On non-Unix platforms signal delivery is not implemented here.
}
fn unix_now_secs() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
}
fn elapsed_ms(started_at_ms: u64) -> u32 {
let now_ms = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64;
now_ms.saturating_sub(started_at_ms).min(u32::MAX as u64) as u32
}
// ─── Tests ────────────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
use crate::store::ChunkFilter;
async fn open_store(dir: &tempfile::TempDir) -> Arc<TaskStore> {
Arc::new(TaskStore::open(&dir.path().join("tr.turso")).await.unwrap())
}
// ── Lost-on-disappear (pure store, no PTY) ────────────────────────────────
#[tokio::test]
async fn lost_on_disappear_marks_stale_running_runs() {
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
// Simulate a run left in "Running" state by a prior daemon.
let stale_id = TaskRunId::new();
store
.insert_run(&TaskRunMeta {
id: stale_id.clone(),
command: "sleep 9999".to_string(),
cwd: "/tmp".into(),
env: vec![],
started_at: unix_now_secs() - 60,
status: RunStatus::Running,
label: None,
initiator: Initiator::Human { camp: "test".to_string() },
beholder_status: None,
pinned: false,
origin: None,
host_pid: None,
})
.await
.unwrap();
// Creating a new driver must mark stale runs Lost.
let _driver = TaskDriver::new(Arc::clone(&store)).await.unwrap();
let meta = store.get_run(&stale_id).await.unwrap().unwrap();
assert!(
matches!(meta.status, RunStatus::Lost { .. }),
"stale run should be Lost, got {:?}",
meta.status
);
}
// ── Stale-run policy (R617-F6) ───────────────────────────────────────────
/// Plant a `Running` row as if some other process had spawned it.
async fn plant_running(
store: &Arc<TaskStore>,
origin: Option<&str>,
host_pid: Option<u32>,
) -> TaskRunId {
let id = TaskRunId::new();
store
.insert_run(&TaskRunMeta {
id: id.clone(),
command: "sleep 9999".to_string(),
cwd: "/tmp".into(),
env: vec![],
started_at: unix_now_secs() - 60,
status: RunStatus::Running,
label: None,
initiator: Initiator::Human {
camp: "test".to_string(),
},
beholder_status: None,
pinned: false,
origin: origin.map(str::to_string),
host_pid,
})
.await
.unwrap();
id
}
async fn is_lost(store: &Arc<TaskStore>, id: &TaskRunId) -> bool {
matches!(
store.get_run(id).await.unwrap().unwrap().status,
RunStatus::Lost { .. }
)
}
fn adopt_terminal() -> StaleRunPolicy {
StaleRunPolicy::AdoptLiveHosts {
origins: vec!["terminal".to_string()],
}
}
/// The property the whole ticket exists for: attaching to a store must not
/// declare another live process's shell dead.
#[tokio::test]
async fn a_run_owned_by_a_live_host_survives_a_new_driver() {
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
// Our own pid is by definition a live process, and is the cheapest
// honest stand-in for "a peer that is still running".
let id = plant_running(&store, Some("terminal"), Some(std::process::id())).await;
let _driver = TaskDriver::with_config(
Arc::clone(&store),
DriverChannels::default(),
adopt_terminal(),
)
.await
.unwrap();
assert!(
!is_lost(&store, &id).await,
"a terminal run whose owner is alive must stay Running — \
tombstoning it is what made a surviving shell read as dead"
);
}
/// The other half: a genuinely abandoned shell must still be tombstoned,
/// or a crashed host leaves permanent zombie tiles.
#[tokio::test]
async fn a_run_whose_host_is_gone_is_still_tombstoned() {
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
// Reaped in-test, so the pid is real-but-dead rather than guessed.
let dead_pid = {
let child = std::process::Command::new("true").spawn().unwrap();
let pid = child.id();
let mut child = child;
let _ = child.wait();
pid
};
let id = plant_running(&store, Some("terminal"), Some(dead_pid)).await;
let _driver = TaskDriver::with_config(
Arc::clone(&store),
DriverChannels::default(),
adopt_terminal(),
)
.await
.unwrap();
assert!(
is_lost(&store, &id).await,
"pid {dead_pid} was reaped; its run has no owner left and must be Lost"
);
}
/// The exemption is narrowed by origin, so ordinary jobs keep the old rule
/// even when their owner happens to still be alive — an in-flight `cargo
/// build` whose driver is gone has nobody left to record its exit.
#[tokio::test]
async fn a_non_matching_origin_is_tombstoned_even_with_a_live_host() {
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
let job = plant_running(&store, None, Some(std::process::id())).await;
let other = plant_running(&store, Some("gnome"), Some(std::process::id())).await;
let _driver = TaskDriver::with_config(
Arc::clone(&store),
DriverChannels::default(),
adopt_terminal(),
)
.await
.unwrap();
assert!(is_lost(&store, &job).await, "an origin-less job is not exempt");
assert!(
is_lost(&store, &other).await,
"an origin outside the list is not exempt"
);
}
/// A row written before `host_pid` existed reads back `None`. Unknown
/// ownership must fall back to the old behaviour rather than stranding the
/// run `Running` forever.
#[tokio::test]
async fn an_unattributed_run_is_tombstoned() {
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
let id = plant_running(&store, Some("terminal"), None).await;
let _driver = TaskDriver::with_config(
Arc::clone(&store),
DriverChannels::default(),
adopt_terminal(),
)
.await
.unwrap();
assert!(is_lost(&store, &id).await);
}
/// `TaskDriver::new` must not have quietly changed behaviour — every
/// existing embedder still gets Lost-on-disappear.
#[tokio::test]
async fn the_default_policy_is_still_lost_on_disappear() {
assert_eq!(StaleRunPolicy::default(), StaleRunPolicy::LostOnDisappear);
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
let id = plant_running(&store, Some("terminal"), Some(std::process::id())).await;
let _driver = TaskDriver::new(Arc::clone(&store)).await.unwrap();
assert!(
is_lost(&store, &id).await,
"the default must tombstone regardless of origin or owner liveness"
);
}
/// The owner is recorded by `spawn_run` itself, not by the caller — the
/// policy is worthless if rows arrive unattributed.
#[tokio::test]
async fn spawn_run_stamps_this_process_as_the_owner() {
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
let driver = TaskDriver::new(Arc::clone(&store)).await.unwrap();
let id = driver
.spawn_run(
"true",
SpawnOpts {
cwd: "/tmp".into(),
origin: Some("terminal".to_string()),
..Default::default()
},
)
.await
.unwrap();
let meta = store.get_run(&id).await.unwrap().unwrap();
assert_eq!(meta.host_pid, Some(std::process::id()));
}
#[tokio::test]
async fn new_driver_does_not_touch_completed_runs() {
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
let done_id = TaskRunId::new();
store
.insert_run(&TaskRunMeta {
id: done_id.clone(),
command: "true".to_string(),
cwd: "/tmp".into(),
env: vec![],
started_at: unix_now_secs() - 10,
status: RunStatus::Running,
label: None,
initiator: Initiator::Human { camp: "test".to_string() },
beholder_status: None,
pinned: false,
origin: None,
host_pid: None,
})
.await
.unwrap();
store
.update_status(&done_id, &RunStatus::Done { exit_code: 0, ended_at: unix_now_secs() })
.await
.unwrap();
let _driver = TaskDriver::new(Arc::clone(&store)).await.unwrap();
let meta = store.get_run(&done_id).await.unwrap().unwrap();
assert!(
matches!(meta.status, RunStatus::Done { .. }),
"completed run must not be touched"
);
}
// ── PTY spawn + capture ───────────────────────────────────────────────────
#[tokio::test]
async fn spawn_echo_and_read_chunks() {
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
let driver = TaskDriver::new(Arc::clone(&store)).await.unwrap();
let id = driver
.spawn_run(
"echo hello_world",
SpawnOpts { cwd: "/tmp".into(), ..Default::default() },
)
.await
.unwrap();
// Wait for the run to complete (poll status up to 5 s).
let deadline = std::time::Instant::now() + Duration::from_secs(5);
loop {
let meta = store.get_run(&id).await.unwrap().unwrap();
if matches!(meta.status, RunStatus::Done { .. } | RunStatus::Lost { .. }) {
break;
}
if std::time::Instant::now() > deadline {
panic!("run did not complete in time, status={:?}", meta.status);
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
// Chunks must contain "hello_world".
let chunks = store
.get_chunks(&id, &ChunkFilter::default())
.await
.unwrap();
let output: Vec<u8> = chunks.into_iter().flat_map(|c| c.bytes).collect();
let text = String::from_utf8_lossy(&output);
assert!(
text.contains("hello_world"),
"expected 'hello_world' in output, got: {text:?}"
);
let meta = store.get_run(&id).await.unwrap().unwrap();
assert!(
matches!(meta.status, RunStatus::Done { exit_code: 0, .. }),
"expected Done(0), got {:?}",
meta.status
);
}
/// Wait for a run to reach a terminal status, or panic.
async fn await_done(store: &TaskStore, id: &TaskRunId) -> TaskRunMeta {
let deadline = std::time::Instant::now() + Duration::from_secs(5);
loop {
let meta = store.get_run(id).await.unwrap().unwrap();
if matches!(meta.status, RunStatus::Done { .. } | RunStatus::Lost { .. }) {
return meta;
}
if std::time::Instant::now() > deadline {
panic!("run did not complete in time, status={:?}", meta.status);
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
}
async fn output_of(store: &TaskStore, id: &TaskRunId) -> String {
let chunks = store.get_chunks(id, &ChunkFilter::default()).await.unwrap();
let bytes: Vec<u8> = chunks.into_iter().flat_map(|c| c.bytes).collect();
String::from_utf8_lossy(&bytes).into_owned()
}
// ── The caller's bytes reach the shell unchanged (R739-S2) ───────────────
/// `AttachResult.argv` is populated on every run, rewrite or not, so
/// `spawn_run` used to join it back into the command line unconditionally.
/// That put every `task.run` command through a whitespace normalization
/// nobody asked for. A multi-line command is the case where that is not
/// cosmetic: the newline the caller wrote becomes a space, and two
/// commands become one nonsense command.
#[cfg(unix)]
#[tokio::test]
async fn a_multi_line_command_is_not_flattened_into_one_line() {
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
let driver = Arc::new(TaskDriver::new(Arc::clone(&store)).await.unwrap());
// Flattened to one line this is `echo one echo two`, which prints
// "one echo two" — a different answer, not a failure, which is what
// makes the old behaviour dangerous rather than merely wrong.
let id = driver
.spawn_run(
"echo one\necho two",
SpawnOpts { cwd: "/tmp".into(), ..Default::default() },
)
.await
.unwrap();
await_done(&store, &id).await;
let out = output_of(&store, &id).await;
assert!(out.contains("one"), "got: {out:?}");
assert!(
out.contains("two"),
"the second line must have run as its own command; got: {out:?}"
);
assert!(
!out.contains("one echo two"),
"the newline was flattened into a space; got: {out:?}"
);
}
/// `resolve_argv` strips `bunx`/`npx`/`pnpm` so a beholder's `matches` sees
/// the bare tool. That is a *matching* concern; it must never reach the
/// spawn, or the wrapper the caller needed is gone from the command.
#[cfg(unix)]
#[tokio::test]
async fn a_wrapper_the_caller_wrote_is_not_stripped_from_the_spawned_command() {
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
let driver = Arc::new(TaskDriver::new(Arc::clone(&store)).await.unwrap());
// `npx` is almost certainly absent in test environments, and that is
// the point: if the wrapper survived, the shell reports it missing. If
// it were stripped we would be running bare `--version`.
let id = driver
.spawn_run(
"npx r739s2-nonexistent-tool --version",
SpawnOpts { cwd: "/tmp".into(), ..Default::default() },
)
.await
.unwrap();
let meta = await_done(&store, &id).await;
let out = output_of(&store, &id).await;
assert!(
!matches!(meta.status, RunStatus::Done { exit_code: 0, .. }),
"expected a failure, got {:?} with output {out:?}",
meta.status
);
assert!(
!out.contains("--version: "),
"the wrapper was stripped and the shell tried to run the flag; got: {out:?}"
);
}
// ── Direct argv (R652-T6) ────────────────────────────────────────────────
#[tokio::test]
async fn explicit_argv_execs_the_program_directly() {
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
let driver = TaskDriver::new(Arc::clone(&store)).await.unwrap();
/* The distinguishing observation: under `sh -c` the child is `sh` and
`$0` is `sh`; exec'd directly it is the program itself. Printing
`$0` is the cheapest way to see which of the two happened. */
let id = driver
.spawn_run(
"unused-because-argv-wins",
SpawnOpts {
cwd: "/tmp".into(),
argv: Some(vec![
"/bin/sh".into(),
"-c".into(),
"printf 'argv0=%s\\n' \"$0\"".into(),
"direct-exec-marker".into(),
]),
..Default::default()
},
)
.await
.unwrap();
await_done(&store, &id).await;
let text = output_of(&store, &id).await;
assert!(
text.contains("argv0=direct-exec-marker"),
"argv should have been exec'd verbatim, got: {text:?}"
);
}
#[tokio::test]
async fn explicit_argv_still_records_the_requested_command() {
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
let driver = TaskDriver::new(Arc::clone(&store)).await.unwrap();
/* A shell tile asks for "$SHELL" and the daemon resolves it to a real
argv. The run must still read back as what was asked for, or the
rail row and the history re-run both show an implementation
detail. */
let id = driver
.spawn_run(
"$SHELL",
SpawnOpts {
cwd: "/tmp".into(),
argv: Some(vec!["/bin/sh".into(), "-c".into(), "true".into()]),
..Default::default()
},
)
.await
.unwrap();
let meta = await_done(&store, &id).await;
assert_eq!(meta.command, "$SHELL");
assert!(
matches!(meta.status, RunStatus::Done { exit_code: 0, .. }),
"expected Done(0), got {:?}",
meta.status
);
}
#[tokio::test]
async fn empty_argv_falls_back_to_the_shell_path() {
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
let driver = TaskDriver::new(Arc::clone(&store)).await.unwrap();
let id = driver
.spawn_run(
"echo empty_argv_fallback",
SpawnOpts { cwd: "/tmp".into(), argv: Some(vec![]), ..Default::default() },
)
.await
.unwrap();
await_done(&store, &id).await;
let text = output_of(&store, &id).await;
assert!(
text.contains("empty_argv_fallback"),
"empty argv must not spawn nothing, got: {text:?}"
);
}
#[tokio::test]
async fn spawn_failing_command_records_nonzero_exit() {
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
let driver = TaskDriver::new(Arc::clone(&store)).await.unwrap();
let id = driver
.spawn_run(
"exit 42",
SpawnOpts { cwd: "/tmp".into(), ..Default::default() },
)
.await
.unwrap();
let deadline = std::time::Instant::now() + Duration::from_secs(5);
loop {
let meta = store.get_run(&id).await.unwrap().unwrap();
if !matches!(meta.status, RunStatus::Running | RunStatus::Pending) {
match meta.status {
RunStatus::Done { exit_code, .. } => {
assert_ne!(exit_code, 0, "exit 42 should produce a non-zero exit code");
}
other => panic!("unexpected status: {other:?}"),
}
break;
}
if std::time::Instant::now() > deadline {
panic!("run did not complete in time");
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
}
// ── Signal handling ───────────────────────────────────────────────────────
#[cfg(unix)]
#[tokio::test]
async fn kill_with_sigterm_transitions_to_killed() {
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
let driver = Arc::new(TaskDriver::new(Arc::clone(&store)).await.unwrap());
let id = driver
.spawn_run(
"sleep 60",
SpawnOpts { cwd: "/tmp".into(), ..Default::default() },
)
.await
.unwrap();
// Give the process a moment to start.
tokio::time::sleep(Duration::from_millis(100)).await;
driver.kill_run(&id, Some(SIGTERM)).await.unwrap();
let deadline = std::time::Instant::now() + Duration::from_secs(10);
loop {
let meta = store.get_run(&id).await.unwrap().unwrap();
if matches!(meta.status, RunStatus::Killed { .. } | RunStatus::Lost { .. }) {
assert!(
matches!(meta.status, RunStatus::Killed { .. }),
"expected Killed, got {:?}",
meta.status
);
break;
}
if std::time::Instant::now() > deadline {
panic!("run did not become Killed in time, status={:?}", meta.status);
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
}
#[cfg(unix)]
#[tokio::test]
async fn kill_run_returns_not_found_after_exit() {
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
let driver = Arc::new(TaskDriver::new(Arc::clone(&store)).await.unwrap());
let id = driver
.spawn_run(
"echo done",
SpawnOpts { cwd: "/tmp".into(), ..Default::default() },
)
.await
.unwrap();
// Wait for natural exit.
let deadline = std::time::Instant::now() + Duration::from_secs(5);
loop {
let meta = store.get_run(&id).await.unwrap().unwrap();
if !matches!(meta.status, RunStatus::Running | RunStatus::Pending) {
break;
}
if std::time::Instant::now() > deadline {
panic!("run did not complete");
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
// Kill on a completed run should return NotFound.
let result = driver.kill_run(&id, None).await;
assert!(
matches!(result, Err(DriverError::NotFound(_))),
"expected NotFound, got {result:?}"
);
}
// ── Stdin relay ───────────────────────────────────────────────────────────
#[cfg(unix)]
#[tokio::test]
async fn stdin_send_reaches_child() {
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
let driver = Arc::new(TaskDriver::new(Arc::clone(&store)).await.unwrap());
// Shell that reads a line from stdin and echoes it back.
let id = driver
.spawn_run(
"read line && echo got_$line",
SpawnOpts {
cwd: "/tmp".into(),
stdin_enabled: true,
..Default::default()
},
)
.await
.unwrap();
tokio::time::sleep(Duration::from_millis(150)).await;
driver.send_stdin(&id, b"hello\n".to_vec()).await.unwrap();
let deadline = std::time::Instant::now() + Duration::from_secs(5);
loop {
let meta = store.get_run(&id).await.unwrap().unwrap();
if !matches!(meta.status, RunStatus::Running | RunStatus::Pending) {
break;
}
if std::time::Instant::now() > deadline {
panic!("run did not complete after stdin input");
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
let chunks = store.get_chunks(&id, &ChunkFilter::default()).await.unwrap();
let raw: Vec<u8> = chunks.into_iter().flat_map(|c| c.bytes).collect();
let text = String::from_utf8_lossy(&raw);
assert!(
text.contains("got_hello"),
"expected 'got_hello' in output, got: {text:?}"
);
}
/// `resize_run` must change the geometry the *child* sees, not just the
/// master fd — so the assertion reads `stty size` from inside the PTY
/// after the resize rather than inspecting the driver's own state.
#[tokio::test]
async fn resize_run_changes_geometry_the_child_sees() {
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
let driver = Arc::new(TaskDriver::new(Arc::clone(&store)).await.unwrap());
// Wait for a line on stdin, then report the geometry as of that moment.
let id = driver
.spawn_run(
"read line && stty size",
SpawnOpts {
cwd: "/tmp".into(),
stdin_enabled: true,
// Spawn at the default 80x24 so the assertion can't pass by
// accident if the resize is a no-op.
..Default::default()
},
)
.await
.unwrap();
tokio::time::sleep(Duration::from_millis(150)).await;
driver.resize_run(&id, 120, 40).await.unwrap();
driver.send_stdin(&id, b"go\n".to_vec()).await.unwrap();
let deadline = std::time::Instant::now() + Duration::from_secs(5);
loop {
let meta = store.get_run(&id).await.unwrap().unwrap();
if !matches!(meta.status, RunStatus::Running | RunStatus::Pending) {
break;
}
if std::time::Instant::now() > deadline {
panic!("run did not complete after stdin input");
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
let chunks = store.get_chunks(&id, &ChunkFilter::default()).await.unwrap();
let raw: Vec<u8> = chunks.into_iter().flat_map(|c| c.bytes).collect();
let text = String::from_utf8_lossy(&raw);
assert!(
text.contains("40 120"),
"expected resized geometry '40 120' in output, got: {text:?}"
);
}
/// A run that is not active on this driver (finished, or never existed) is
/// `NotFound` rather than a panic — same contract as `send_stdin`.
#[tokio::test]
async fn resize_run_returns_not_found_after_exit() {
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
let driver = Arc::new(TaskDriver::new(Arc::clone(&store)).await.unwrap());
let id = driver
.spawn_run("true", SpawnOpts { cwd: "/tmp".into(), ..Default::default() })
.await
.unwrap();
let deadline = std::time::Instant::now() + Duration::from_secs(5);
loop {
let meta = store.get_run(&id).await.unwrap().unwrap();
if !matches!(meta.status, RunStatus::Running | RunStatus::Pending) {
break;
}
if std::time::Instant::now() > deadline {
panic!("run did not exit");
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
assert!(matches!(
driver.resize_run(&id, 100, 30).await,
Err(DriverError::NotFound(_))
));
}
// ── Tier-2 side-channel log fd ────────────────────────────────────────────
/// Verify that a child writing a JSON-line to `YAH_LOG_PIPE` (via
/// `printf ... >> $YAH_LOG_PIPE`) produces a shim event with the correct
/// fields in the store.
///
/// The child opens the FIFO path for writing — no fd inheritance needed.
#[cfg(unix)]
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn log_pipe_events_land_in_store() {
use crate::store::EventFilter;
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
let driver = Arc::new(TaskDriver::new(Arc::clone(&store)).await.unwrap());
// The shell writes one JSON-line to the FIFO by redirecting printf
// output to the path stored in YAH_LOG_PIPE.
let cmd = r#"printf '{"level":"warn","target":"test.shim","msg":"hello-from-pipe","fields":{"x":42},"_lib":"test-shim","_lib_ver":"0.1.0"}\n' >> "$YAH_LOG_PIPE""#;
let id = driver
.spawn_run(cmd, SpawnOpts { cwd: "/tmp".into(), ..Default::default() })
.await
.unwrap();
// Wait for run completion. Deadline is generous because parallel-test
// load + the rt.block_on hops from the reader/log threads can slow
// child-process scheduling.
let deadline = std::time::Instant::now() + Duration::from_secs(20);
loop {
let meta = store.get_run(&id).await.unwrap().unwrap();
if matches!(meta.status, RunStatus::Done { .. } | RunStatus::Lost { .. }) {
break;
}
if std::time::Instant::now() > deadline {
panic!("run did not complete in time");
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
// The log receiver thread drains after the lifecycle task drops the
// write-end FdCloser; give it a brief moment.
tokio::time::sleep(Duration::from_millis(500)).await;
let events = store.query_events(&id, &EventFilter::default()).await.unwrap();
assert!(
!events.is_empty(),
"expected at least one shim event, got none"
);
let ev = events.iter().find(|e| e.target == "test.shim");
let ev = ev.expect("event with target 'test.shim' not found");
assert_eq!(ev.msg, "hello-from-pipe");
assert_eq!(ev.level, crate::types::Level::Warn);
assert!(
matches!(&ev.source, crate::types::EventSource::Shim { lib, .. } if lib == "test-shim"),
"unexpected source: {:?}",
ev.source
);
assert_eq!(ev.fields.get("x"), Some(&serde_json::json!(42)));
}
/// When `log_fd_enabled` is false, neither `YAH_TASK_RUN` nor
/// `YAH_LOG_PIPE` are exported, and no shim events are written.
#[cfg(unix)]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn log_pipe_disabled_produces_no_events() {
use crate::store::EventFilter;
let dir = tempfile::tempdir().unwrap();
let store = open_store(&dir).await;
let driver = Arc::new(TaskDriver::new(Arc::clone(&store)).await.unwrap());
// Try to write to YAH_LOG_PIPE; the conditional guards against
// the variable being absent, so the command always exits 0.
let cmd = r#"[ -n "$YAH_LOG_PIPE" ] && printf '{"level":"info","target":"t","msg":"m","fields":{}}\n' >> "$YAH_LOG_PIPE" || true"#;
let id = driver
.spawn_run(
cmd,
SpawnOpts { cwd: "/tmp".into(), log_fd_enabled: false, ..Default::default() },
)
.await
.unwrap();
let deadline = std::time::Instant::now() + Duration::from_secs(5);
loop {
let meta = store.get_run(&id).await.unwrap().unwrap();
if matches!(meta.status, RunStatus::Done { .. } | RunStatus::Lost { .. }) {
break;
}
if std::time::Instant::now() > deadline {
panic!("run did not complete");
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
tokio::time::sleep(Duration::from_millis(100)).await;
let events = store.query_events(&id, &EventFilter::default()).await.unwrap();
assert!(
events.is_empty(),
"expected no shim events when log_fd_enabled=false, got {}",
events.len()
);
}
}