diff --git a/distro/skills/berd-monitor/SKILL.md b/distro/skills/berd-monitor/SKILL.md new file mode 100644 index 000000000..fc199b26a --- /dev/null +++ b/distro/skills/berd-monitor/SKILL.md @@ -0,0 +1,52 @@ +--- +name: berd-monitor +description: >- + Run a long-lived command outside the current Berd turn and wake the owning + session with actionable stdout. Use for builds, tests, reviews, deployments, + polling, watchers, and other external waits. +metadata: + berdBundled: true +--- + +# Berd Monitor + +Use `berd-monitor` when a command may outlive the current turn. It detaches the +producer from Berd and the agent harness, buffers output across delivery +failures, and sends complete stdout lines to the exact originating session. + +Run short, bounded commands normally. Do not hold a turn open with `sleep`, +polling, log tailing, or a foreground timeout. + +## Start + +```bash +berd-monitor run \ + --state-key \ + --label '' \ + --instructions '' \ + -- [args...] +``` + +`AGENT_SESSION_ID` selects the current session. Use `--session-id` only for +another positively identified session; never infer a session from the working +directory. The command prints the detached monitor PID. After checking the +monitor's `watcher.log` under the printed state directory when diagnosis is +needed, continue other work or end the turn—never wait on the detached PID. + +The producer's stdout is the event API: + +- emit concise, newline-terminated milestones and flush promptly; +- put verbose output and diagnostics in durable logs or stderr; +- avoid NUL bytes and emit a final summary when practical. + +Default `--if-running steer` adds an event to an active run or starts a new +turn. Use `--if-running queue` only when the active run must not be steered. +Pending events retry without being dropped, and a trailing partial line is +delivered when the producer exits. + +Delivered messages are visibly labeled as coming from `berd-monitor`. Stop a +monitor with: + +```bash +berd-monitor stop --state-key +``` diff --git a/distro/skills/berd-orchestrator/SKILL.md b/distro/skills/berd-orchestrator/SKILL.md new file mode 100644 index 000000000..38d4f07bf --- /dev/null +++ b/distro/skills/berd-orchestrator/SKILL.md @@ -0,0 +1,29 @@ +--- +name: berd-orchestrator +description: >- + Coordinate work across Berd sessions while keeping one conversation + available to the user. Use for a long-lived orchestration session. +metadata: + berdBundled: true +--- + +# Berd Orchestrator + +Keep this session available for conversation. Use `berdctl session list`, +`get`, `create`, and `send` to delegate substantial work to other sessions +instead of doing it here. + +Prefer an existing session that owns the relevant context. Create one when +work has no owner, and give it a complete task. Do not approve, merge, or +archive work without the user's direction. + +When creating or sending to another session, pass +`--from ''` so the receiving transcript explains +where the message came from. + +Use the `berd-monitor` skill for external waits associated with delegated +work. End the turn after starting a monitor; do not poll in the foreground. + +Stay quiet while work is progressing. When attention is needed, either +continue an obvious next step or bring the user one result or decision with a +`[session link](berd://session/)`. diff --git a/justfile b/justfile index ca853fac1..350803bac 100644 --- a/justfile +++ b/justfile @@ -498,16 +498,18 @@ dev: export VITE_APP_VERSION="$BERD_APP_VERSION_RICH" echo "Using app version: ${BERD_APP_VERSION} (${BERD_APP_VERSION_RICH})" - # tauri dev only builds the root package; the berdctl CLI workspace - # member needs an explicit build, resolved at runtime via BERDCTL_BIN - # because tauri.dev.conf.json blanks externalBin. + # tauri dev only builds the root package; the agent-facing CLI workspace + # members need explicit builds because tauri.dev.conf.json blanks externalBin. BERDCTL_FEATURES=() [[ "${VITE_FEEDBACK:-0}" == "1" ]] && BERDCTL_FEATURES+=(--features block-feedback) # ${arr[@]+...} guards the empty-array expansion, which bash 3.2 (stock # macOS) treats as an unbound variable under `set -u`. (cd src-tauri && cargo build -p berdctl ${BERDCTL_FEATURES[@]+"${BERDCTL_FEATURES[@]}"}) + (cd src-tauri && cargo build -p berd-monitor) export BERDCTL_BIN="${CARGO_TARGET_DIR}/debug/berdctl" + export BERD_MONITOR_BIN="${CARGO_TARGET_DIR}/debug/berd-monitor" echo "Using berdctl CLI: ${BERDCTL_BIN}" + echo "Using berd-monitor CLI: ${BERD_MONITOR_BIN}" if [[ "${VITE_AGENT_TOOLS:-0}" == "1" ]]; then ./scripts/prepare-bb-cli-resource.sh diff --git a/scripts/prepare-berdctl-sidecar.sh b/scripts/prepare-berdctl-sidecar.sh index e91e8a447..0f9e4838e 100755 --- a/scripts/prepare-berdctl-sidecar.sh +++ b/scripts/prepare-berdctl-sidecar.sh @@ -1,5 +1,5 @@ #!/usr/bin/env bash -# Build and stage the berdctl CLI for Tauri's externalBin bundling. +# Build and stage the berdctl and berd-monitor CLIs for Tauri externalBin bundling. # # Tauri expects external binaries to be present at build time with the target # triple appended to the configured stem. For config @@ -13,8 +13,9 @@ usage() { cat <<'USAGE' Usage: scripts/prepare-berdctl-sidecar.sh [target-triple] -Builds the berdctl workspace crate in release mode and copies the binary -into src-tauri/binaries with the target triple suffix required by Tauri. +Builds the berdctl and berd-monitor workspace crates in release mode and +copies both binaries into src-tauri/binaries with the target triple suffix +required by Tauri. The triple defaults to the rustc host. Pass it explicitly (or set BERDCTL_TRIPLE) when the Tauri build itself uses an explicit --target, so @@ -29,9 +30,9 @@ if [[ "${1:-}" == "-h" || "${1:-}" == "--help" ]]; then fi EXPLICIT_TRIPLE="${1:-${BERDCTL_TRIPLE:-}}" -CARGO_ARGS=(build -p berdctl --release) +CARGO_ARGS=(build -p berdctl -p berd-monitor --release) if [[ "${VITE_FEEDBACK:-0}" == "1" ]]; then - CARGO_ARGS+=(--features block-feedback) + CARGO_ARGS+=(--features berdctl/block-feedback) fi if [[ -n "$EXPLICIT_TRIPLE" ]]; then TRIPLE="$EXPLICIT_TRIPLE" @@ -75,3 +76,19 @@ mkdir -p "$OUT_DIR" cp "$BUILT" "$OUT" chmod +x "$OUT" echo "Staged berdctl sidecar: $OUT" + +if [[ -n "$EXPLICIT_TRIPLE" ]]; then + MONITOR_BUILT="$TARGET_DIR/$TRIPLE/release/berd-monitor" +else + MONITOR_BUILT="$TARGET_DIR/release/berd-monitor" +fi + +if [[ ! -x "$MONITOR_BUILT" ]]; then + echo "Built berd-monitor binary not found at: $MONITOR_BUILT" >&2 + exit 1 +fi + +MONITOR_OUT="$OUT_DIR/berd-monitor-$TRIPLE" +cp "$MONITOR_BUILT" "$MONITOR_OUT" +chmod +x "$MONITOR_OUT" +echo "Staged berd-monitor sidecar: $MONITOR_OUT" diff --git a/scripts/windows/Dev-Windows.ps1 b/scripts/windows/Dev-Windows.ps1 index 63a4d0149..12edb6db4 100644 --- a/scripts/windows/Dev-Windows.ps1 +++ b/scripts/windows/Dev-Windows.ps1 @@ -124,6 +124,13 @@ if (-not (Test-Path $env:BERDCTL_BIN -PathType Leaf)) { } Write-WindowsDevInfo "Using berdctl CLI: $env:BERDCTL_BIN" +Invoke-CheckedCommand -FilePath "cargo" -ArgumentList @("build", "-p", "berd-monitor") -WorkingDirectory (Join-Path (Get-BerdRepoRoot) "src-tauri") -Label "cargo build berd-monitor" +$env:BERD_MONITOR_BIN = Join-Path (Join-Path $env:CARGO_TARGET_DIR "debug") "berd-monitor.exe" +if (-not (Test-Path $env:BERD_MONITOR_BIN -PathType Leaf)) { + throw "Expected berd-monitor.exe at $env:BERD_MONITOR_BIN after cargo build." +} +Write-WindowsDevInfo "Using berd-monitor CLI: $env:BERD_MONITOR_BIN" + if ([string]::IsNullOrWhiteSpace($env:GOOSE_BIN)) { $env:GOOSE_BUILD_PROFILE = "debug" $result = Invoke-EnsureLocalGoose -Action Check diff --git a/scripts/windows/Stage-Sidecar-Windows.ps1 b/scripts/windows/Stage-Sidecar-Windows.ps1 index e7542af02..bf8225564 100644 --- a/scripts/windows/Stage-Sidecar-Windows.ps1 +++ b/scripts/windows/Stage-Sidecar-Windows.ps1 @@ -71,9 +71,9 @@ Write-WindowsDevInfo "Staged Goose sidecar: $staged" $tauriTargetDir = Get-TauriCargoTargetDir $env:CARGO_TARGET_DIR = $tauriTargetDir $hostTriple = Get-RustHostTriple -$cargoArgs = @("build", "-p", "berdctl", "--release") +$cargoArgs = @("build", "-p", "berdctl", "-p", "berd-monitor", "--release") if ($env:VITE_FEEDBACK -eq "1") { - $cargoArgs += @("--features", "block-feedback") + $cargoArgs += @("--features", "berdctl/block-feedback") } if (-not [string]::IsNullOrWhiteSpace($hostTriple) -and $Triple -ne $hostTriple) { $cargoArgs += @("--target", $Triple) @@ -82,10 +82,15 @@ if (-not [string]::IsNullOrWhiteSpace($hostTriple) -and $Triple -ne $hostTriple) $berdctlReleaseDir = Join-Path $tauriTargetDir "release" } Invoke-CheckedCommand -FilePath "cargo" -ArgumentList $cargoArgs ` - -WorkingDirectory (Join-Path (Get-BerdRepoRoot) "src-tauri") -Label "cargo build -p berdctl --release" + -WorkingDirectory (Join-Path (Get-BerdRepoRoot) "src-tauri") -Label "cargo build -p berdctl -p berd-monitor --release" $berdctlSource = Join-Path $berdctlReleaseDir (Get-WindowsExeName "berdctl") $staged = Stage-WindowsSidecar -SourcePath $berdctlSource -Triple $Triple -Stem "berdctl" -BinDir $binDir Write-WindowsDevInfo "Staged berdctl sidecar: $staged" +# ── berd-monitor ───────────────────────────────────────────── +$monitorSource = Join-Path $berdctlReleaseDir (Get-WindowsExeName "berd-monitor") +$staged = Stage-WindowsSidecar -SourcePath $monitorSource -Triple $Triple -Stem "berd-monitor" -BinDir $binDir +Write-WindowsDevInfo "Staged berd-monitor sidecar: $staged" + # Catch is deliberately not staged on Windows (see header). Write-WindowsDevInfo "Skipping Catch sidecar: unsupported on Windows (excluded from externalBin)." diff --git a/scripts/windows/Test-WindowsDev.ps1 b/scripts/windows/Test-WindowsDev.ps1 index 5f6c0982b..e58f0b7c7 100644 --- a/scripts/windows/Test-WindowsDev.ps1 +++ b/scripts/windows/Test-WindowsDev.ps1 @@ -439,6 +439,7 @@ try { $windowsExternalBin = @(Get-ObjectValue (Get-ObjectValue $windowsConf "bundle") "externalBin") Assert-Equal "Windows externalBin stages goosed" ($windowsExternalBin -contains "binaries/goosed") $true Assert-Equal "Windows externalBin stages berdctl" ($windowsExternalBin -contains "binaries/berdctl") $true + Assert-Equal "Windows externalBin stages berd-monitor" ($windowsExternalBin -contains "binaries/berd-monitor") $true Assert-Equal "Windows externalBin excludes catch" ($windowsExternalBin -contains "binaries/catch") $false # Tauri merges platform overlays into the base config with json_patch (RFC @@ -454,6 +455,7 @@ try { $mergedExternalBin = if ($null -ne $windowsExternalBin) { $windowsExternalBin } else { $baseExternalBin } Assert-Equal "merged Windows externalBin stages goosed" ($mergedExternalBin -contains "binaries/goosed") $true Assert-Equal "merged Windows externalBin stages berdctl" ($mergedExternalBin -contains "binaries/berdctl") $true + Assert-Equal "merged Windows externalBin stages berd-monitor" ($mergedExternalBin -contains "binaries/berd-monitor") $true Assert-Equal "merged Windows externalBin drops catch" ($mergedExternalBin -contains "binaries/catch") $false # ── Windows bundle recipes route through native staging ────── diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index 11ef77893..be42f1984 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -583,6 +583,16 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a8241f3ebb85c056b509d4327ad0358fbbba6ffb340bf388f26350aeda225b1" +[[package]] +name = "berd-monitor" +version = "0.6.2" +dependencies = [ + "clap", + "fs2", + "libc", + "windows-sys 0.59.0", +] + [[package]] name = "berd-voice" version = "0.1.0" @@ -951,6 +961,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "473c7e07f409a8d772161724aa8db6a765a2532a70f9667eeb7b49d3d02fbdca" dependencies = [ "clap_builder", + "clap_derive", ] [[package]] @@ -966,6 +977,18 @@ dependencies = [ "terminal_size", ] +[[package]] +name = "clap_derive" +version = "4.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d012d2b9d65aca7f18f4d9878a045bc17899bba951561ba5ec3c2ba1eed9a061" +dependencies = [ + "heck 0.5.0", + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "clap_lex" version = "1.1.0" @@ -2038,6 +2061,16 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213" +dependencies = [ + "libc", + "winapi", +] + [[package]] name = "fs_extra" version = "1.3.0" diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 0d87e6415..3e72803f1 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -14,7 +14,12 @@ crate-type = ["staticlib", "cdylib", "rlib"] # --features ...` only works for workspace members. app-test-driver # stays excluded (a plain path dependency, as before this workspace existed). [workspace] -members = ["crates/berd-voice", "crates/berdctl", "plugins/berdctl"] +members = [ + "crates/berd-monitor", + "crates/berd-voice", + "crates/berdctl", + "plugins/berdctl", +] exclude = ["plugins/app-test-driver"] [build-dependencies] diff --git a/src-tauri/crates/berd-monitor/Cargo.toml b/src-tauri/crates/berd-monitor/Cargo.toml new file mode 100644 index 000000000..a15db1b44 --- /dev/null +++ b/src-tauri/crates/berd-monitor/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "berd-monitor" +version = "0.6.2" +description = "Detached command monitor for Berd sessions" +authors = ["Block, Inc."] +edition = "2021" + +[dependencies] +clap = { version = "4", features = ["derive", "env"] } +fs2 = "0.4" + +[target.'cfg(unix)'.dependencies] +libc = "0.2" + +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.59", features = [ + "Win32_Foundation", + "Win32_Security", + "Win32_Storage_FileSystem", + "Win32_System_Diagnostics_ToolHelp", + "Win32_System_JobObjects", + "Win32_System_Threading", +] } diff --git a/src-tauri/crates/berd-monitor/src/main.rs b/src-tauri/crates/berd-monitor/src/main.rs new file mode 100644 index 000000000..73094c758 --- /dev/null +++ b/src-tauri/crates/berd-monitor/src/main.rs @@ -0,0 +1,2060 @@ +use clap::{Parser, Subcommand, ValueEnum}; +use fs2::FileExt; +use std::env; +use std::ffi::OsString; +use std::fs::{self, File, OpenOptions}; +use std::io::{self, Read, Write}; +use std::path::{Path, PathBuf}; +use std::process::{Command, ExitCode, Stdio}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::mpsc::{self, RecvTimeoutError}; +use std::thread; +use std::time::{Duration, Instant, SystemTime}; + +const FOREGROUND_ENV: &str = "BERD_MONITOR_FOREGROUND"; +const LAUNCH_TOKEN_ENV: &str = "BERD_MONITOR_LAUNCH_TOKEN"; +const RETRY_INTERVAL: Duration = Duration::from_secs(15); +const BATCH_WINDOW: Duration = Duration::from_millis(250); +const MAX_DELIVERY_BYTES: usize = 40_000; +const MAX_PROMPT_CODE_UNITS: usize = 49_000; +const MAX_INSTRUCTIONS_CODE_UNITS: usize = 4_000; +const MAX_LABEL_CODE_UNITS: usize = 120; +const MAX_LOCK_CANDIDATES: usize = 8; +const DELIVERY_POLL_INTERVAL: Duration = Duration::from_millis(50); +const DELIVERY_TIMEOUT: Duration = Duration::from_secs(15); +const LAUNCH_TIMEOUT: Duration = Duration::from_secs(10); +const LAUNCH_TERMINATION_GRACE: Duration = Duration::from_secs(1); +const PENDING_HEADER_PREFIX: &str = "BERD_MONITOR_PENDING_V1 "; + +static PENDING_GENERATION_COUNTER: AtomicU64 = AtomicU64::new(0); + +#[derive(Clone, Copy, Debug, ValueEnum)] +enum RunningMode { + Steer, + Queue, +} + +impl RunningMode { + fn as_str(self) -> &'static str { + match self { + Self::Steer => "steer", + Self::Queue => "queue", + } + } +} + +#[derive(Debug, Parser)] +#[command( + name = "berd-monitor", + about = "Run a long-lived command and wake its owning Berd session" +)] +struct Cli { + #[command(subcommand)] + command: MonitorCommand, +} + +#[derive(Debug, Subcommand)] +enum MonitorCommand { + /// Start a detached line-oriented command monitor. + Run { + /// Stable key used to identify and stop this monitor. + #[arg(long)] + state_key: String, + /// Concise source name included with delivered events. + #[arg(long)] + label: String, + /// Guidance appended to every delivered event. + #[arg(long, default_value = "")] + instructions: String, + /// Berd session that owns the monitor. + #[arg(long, env = "AGENT_SESSION_ID")] + session_id: String, + /// Whether events steer a running turn or queue behind it. + #[arg(long, value_enum, default_value_t = RunningMode::Steer)] + if_running: RunningMode, + /// Producer command and arguments, following `--`. + #[arg(last = true, required = true, allow_hyphen_values = true)] + command: Vec, + }, + /// Request that a running monitor stop its producer. + Stop { + /// Stable key passed when the monitor was started. + #[arg(long)] + state_key: String, + /// Berd session that owns the monitor. + #[arg(long, env = "AGENT_SESSION_ID")] + session_id: String, + }, +} + +struct StatePaths { + root: PathBuf, + log: PathBuf, + pending: PathBuf, + owner: PathBuf, +} + +struct PendingState { + generation: String, + active_len: usize, + bytes: Vec, +} + +enum ProducerOutputEvent { + Data(Vec), + ReadError(io::Error), +} + +impl PendingState { + fn empty() -> Self { + Self { + generation: next_pending_generation(), + active_len: 0, + bytes: Vec::new(), + } + } + + fn delivery_id(&self, paths: &StatePaths) -> String { + let state_hash = stable_hash(paths.root.to_string_lossy().as_bytes()); + format!("berd-monitor-{state_hash:016x}-{}", self.generation) + } + + fn rotate(&mut self) { + self.generation = next_pending_generation(); + self.active_len = 0; + } +} + +fn next_pending_generation() -> String { + let timestamp = SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos(); + let sequence = PENDING_GENERATION_COUNTER.fetch_add(1, Ordering::Relaxed); + format!("{}-{timestamp}-{sequence}", std::process::id()) +} + +impl StatePaths { + fn for_key(key: &str, session_id: &str) -> Self { + let identity = format!("{session_id}\0{key}"); + let suffix = stable_hash(identity.as_bytes()); + let safe = key + .chars() + .take(80) + .map(|character| { + if character.is_ascii_alphanumeric() || matches!(character, '.' | '_' | '-') { + character + } else { + '-' + } + }) + .collect::(); + let root = state_base_dir().join(format!("{safe}-{suffix:016x}")); + Self { + log: root.join("watcher.log"), + pending: root.join("pending.txt"), + owner: root.join("owner.pid"), + root, + } + } + + fn launch_status(&self, token: &str) -> PathBuf { + self.root.join(format!("launch-{token}.status")) + } + + fn stop_for(&self, owner_token: &str) -> PathBuf { + self.root.join(format!( + "stop-{:016x}.requested", + stable_hash(owner_token.as_bytes()) + )) + } +} + +fn state_base_dir() -> PathBuf { + if let Some(explicit) = env::var_os("BERD_MONITOR_STATE_DIR") { + return PathBuf::from(explicit); + } + #[cfg(target_os = "windows")] + if let Some(local) = env::var_os("LOCALAPPDATA") { + return PathBuf::from(local).join("Berd").join("monitor"); + } + #[cfg(target_os = "macos")] + if let Some(home) = env::var_os("HOME") { + return PathBuf::from(home) + .join("Library") + .join("Caches") + .join("Berd") + .join("monitor"); + } + #[cfg(all(unix, not(target_os = "macos")))] + { + if let Some(runtime) = env::var_os("XDG_RUNTIME_DIR") { + return PathBuf::from(runtime).join("berd-monitor"); + } + if let Some(state) = env::var_os("XDG_STATE_HOME") { + return PathBuf::from(state).join("berd").join("monitor"); + } + if let Some(home) = env::var_os("HOME") { + return PathBuf::from(home) + .join(".local") + .join("state") + .join("berd") + .join("monitor"); + } + } + env::temp_dir().join(format!("berd-monitor-{}", current_user_suffix())) +} + +#[cfg(unix)] +fn current_user_suffix() -> u32 { + unsafe { libc::geteuid() } +} + +#[cfg(windows)] +fn current_user_suffix() -> u32 { + std::process::id() +} + +fn stable_hash(bytes: &[u8]) -> u64 { + bytes.iter().fold(0xcbf29ce484222325, |hash, byte| { + (hash ^ u64::from(*byte)).wrapping_mul(0x100000001b3) + }) +} + +fn main() -> ExitCode { + match run_cli() { + Ok(()) => ExitCode::SUCCESS, + Err(error) => { + eprintln!("berd-monitor: {error}"); + ExitCode::FAILURE + } + } +} + +fn run_cli() -> Result<(), String> { + let cli = Cli::parse(); + match cli.command { + MonitorCommand::Run { + state_key, + label, + instructions, + session_id, + if_running, + command, + } => { + validate_single_line("--label", &label)?; + if label.encode_utf16().count() > MAX_LABEL_CODE_UNITS { + return Err(format!( + "--label must not exceed {MAX_LABEL_CODE_UNITS} characters" + )); + } + if instructions.encode_utf16().count() > MAX_INSTRUCTIONS_CODE_UNITS { + return Err(format!( + "--instructions must not exceed {MAX_INSTRUCTIONS_CODE_UNITS} characters" + )); + } + if env::var_os(FOREGROUND_ENV).is_some() { + run_foreground( + &state_key, + &label, + &instructions, + &session_id, + if_running, + &command, + env::var(LAUNCH_TOKEN_ENV).ok().as_deref(), + ) + .map_err(|error| error.to_string()) + } else { + spawn_detached(&state_key, &session_id) + } + } + MonitorCommand::Stop { + state_key, + session_id, + } => request_stop(&state_key, &session_id).map_err(|error| error.to_string()), + } +} + +fn validate_single_line(flag: &str, value: &str) -> Result<(), String> { + if value.trim().is_empty() { + return Err(format!("{flag} must not be empty")); + } + if value.contains('\n') || value.contains('\r') { + return Err(format!("{flag} must be a single line")); + } + Ok(()) +} + +fn spawn_detached(state_key: &str, session_id: &str) -> Result<(), String> { + let executable = + env::current_exe().map_err(|error| format!("resolve current executable: {error}"))?; + let token = format!( + "{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + ); + let paths = StatePaths::for_key(state_key, session_id); + let status_path = paths.launch_status(&token); + let mut child = Command::new(executable); + child + .args(env::args_os().skip(1)) + .env(FOREGROUND_ENV, "1") + .env(LAUNCH_TOKEN_ENV, &token) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + configure_detached(&mut child); + let mut process = child + .spawn() + .map_err(|error| format!("start detached monitor: {error}"))?; + let stop_path = paths.stop_for(&token); + wait_for_launch( + &mut process, + &paths, + &status_path, + &stop_path, + LAUNCH_TIMEOUT, + ) +} + +fn wait_for_launch( + process: &mut std::process::Child, + paths: &StatePaths, + status_path: &Path, + stop_path: &Path, + timeout: Duration, +) -> Result<(), String> { + let deadline = Instant::now() + timeout; + loop { + if let Ok(status) = fs::read_to_string(status_path) { + let _ = fs::remove_file(status_path); + if status.trim() == "ready" { + println!("{} {}", process.id(), paths.root.display()); + return Ok(()); + } + return Err(status + .strip_prefix("error: ") + .unwrap_or(status.trim()) + .to_owned()); + } + if Instant::now() >= deadline { + terminate_timed_out_launch(process, status_path, stop_path); + return Err(format!( + "monitor did not become ready within {} seconds; inspect {}", + timeout.as_secs_f64(), + paths.log.display() + )); + } + if let Ok(Some(status)) = process.try_wait() { + let _ = fs::remove_file(status_path); + return Err(format!("monitor exited before becoming ready ({status})")); + } + thread::sleep(Duration::from_millis(50)); + } +} + +fn terminate_timed_out_launch( + process: &mut std::process::Child, + status_path: &Path, + stop_path: &Path, +) { + let _ = fs::write(stop_path, b"stop\n"); + let deadline = Instant::now() + LAUNCH_TERMINATION_GRACE; + while Instant::now() < deadline { + match process.try_wait() { + Ok(Some(_)) => break, + Ok(None) => thread::sleep(Duration::from_millis(50)), + Err(_) => break, + } + } + if process.try_wait().ok().flatten().is_none() { + let _ = process.kill(); + let _ = process.wait(); + } + let _ = fs::remove_file(status_path); + let _ = fs::remove_file(stop_path); +} + +#[cfg(unix)] +fn configure_detached(command: &mut Command) { + use std::os::unix::process::CommandExt; + unsafe { + command.pre_exec(|| { + if libc::setsid() == -1 { + return Err(io::Error::last_os_error()); + } + Ok(()) + }); + } +} + +#[cfg(unix)] +fn configure_producer(command: &mut Command) { + use std::os::unix::process::CommandExt; + unsafe { + command.pre_exec(|| { + if libc::setpgid(0, 0) == -1 { + return Err(io::Error::last_os_error()); + } + Ok(()) + }); + } +} + +#[cfg(windows)] +fn configure_producer(command: &mut Command) { + use std::os::windows::process::CommandExt; + const CREATE_NEW_PROCESS_GROUP: u32 = 0x0000_0200; + const CREATE_SUSPENDED: u32 = 0x0000_0004; + command.creation_flags(CREATE_NEW_PROCESS_GROUP | CREATE_SUSPENDED); +} + +#[cfg(unix)] +fn resume_producer(_child: &std::process::Child) -> io::Result<()> { + Ok(()) +} + +#[cfg(windows)] +fn resume_producer(child: &std::process::Child) -> io::Result<()> { + use std::mem::{size_of, zeroed}; + use windows_sys::Win32::Foundation::{CloseHandle, INVALID_HANDLE_VALUE}; + use windows_sys::Win32::System::Diagnostics::ToolHelp::{ + CreateToolhelp32Snapshot, Thread32First, Thread32Next, TH32CS_SNAPTHREAD, THREADENTRY32, + }; + use windows_sys::Win32::System::Threading::{OpenThread, ResumeThread, THREAD_SUSPEND_RESUME}; + + let snapshot = unsafe { CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD, 0) }; + if snapshot == INVALID_HANDLE_VALUE { + return Err(io::Error::last_os_error()); + } + let mut entry: THREADENTRY32 = unsafe { zeroed() }; + entry.dwSize = size_of::() as u32; + let mut found = unsafe { Thread32First(snapshot, &mut entry) } != 0; + while found && entry.th32OwnerProcessID != child.id() { + found = unsafe { Thread32Next(snapshot, &mut entry) } != 0; + } + unsafe { + CloseHandle(snapshot); + } + if !found { + return Err(io::Error::new( + io::ErrorKind::NotFound, + "could not find the suspended producer thread", + )); + } + let thread = unsafe { OpenThread(THREAD_SUSPEND_RESUME, 0, entry.th32ThreadID) }; + if thread.is_null() { + return Err(io::Error::last_os_error()); + } + let resumed = unsafe { ResumeThread(thread) }; + let resume_error = (resumed == u32::MAX).then(io::Error::last_os_error); + unsafe { + CloseHandle(thread); + } + if let Some(error) = resume_error { + return Err(error); + } + Ok(()) +} + +#[cfg(unix)] +struct ProducerTree { + process_group: i32, +} + +#[cfg(unix)] +fn attach_producer_tree(child: &std::process::Child) -> io::Result { + Ok(ProducerTree { + process_group: child.id() as i32, + }) +} + +#[cfg(unix)] +fn terminate_producer_tree(child: &mut std::process::Child, tree: &ProducerTree) { + unsafe { + libc::kill(-tree.process_group, libc::SIGKILL); + } + let _ = child.kill(); +} + +#[cfg(windows)] +struct ProducerTree { + job: windows_sys::Win32::Foundation::HANDLE, +} + +#[cfg(windows)] +fn attach_producer_tree(child: &std::process::Child) -> io::Result { + use std::mem::{size_of, zeroed}; + use std::os::windows::io::AsRawHandle; + use windows_sys::Win32::System::JobObjects::{ + AssignProcessToJobObject, CreateJobObjectW, JobObjectExtendedLimitInformation, + SetInformationJobObject, JOBOBJECT_EXTENDED_LIMIT_INFORMATION, + JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE, + }; + + let job = unsafe { CreateJobObjectW(std::ptr::null(), std::ptr::null()) }; + if job.is_null() { + return Err(io::Error::last_os_error()); + } + let mut limits: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = unsafe { zeroed() }; + limits.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE; + let configured = unsafe { + SetInformationJobObject( + job, + JobObjectExtendedLimitInformation, + (&raw const limits).cast(), + size_of::() as u32, + ) + }; + let assigned = configured != 0 + && unsafe { AssignProcessToJobObject(job, child.as_raw_handle().cast()) } != 0; + if !assigned { + let error = io::Error::last_os_error(); + unsafe { + windows_sys::Win32::Foundation::CloseHandle(job); + } + return Err(error); + } + Ok(ProducerTree { job }) +} + +#[cfg(windows)] +fn terminate_producer_tree(child: &mut std::process::Child, tree: &ProducerTree) { + unsafe { + windows_sys::Win32::System::JobObjects::TerminateJobObject(tree.job, 1); + } + let _ = child.kill(); +} + +#[cfg(windows)] +impl Drop for ProducerTree { + fn drop(&mut self) { + unsafe { + windows_sys::Win32::Foundation::CloseHandle(self.job); + } + } +} + +struct ProducerProcess { + child: std::process::Child, + tree: Option, +} + +impl ProducerProcess { + fn terminate_and_wait(&mut self) { + self.terminate_and_wait_with(terminate_producer_tree); + } + + fn terminate_and_wait_with( + &mut self, + terminate: impl FnOnce(&mut std::process::Child, &ProducerTree), + ) { + let Some(tree) = self.tree.take() else { + return; + }; + terminate(&mut self.child, &tree); + let _ = self.child.wait(); + } +} + +impl Drop for ProducerProcess { + fn drop(&mut self) { + self.terminate_and_wait(); + } +} + +fn ensure_private_directory(path: &Path) -> io::Result<()> { + fs::create_dir_all(path)?; + #[cfg(unix)] + { + use std::os::unix::fs::{MetadataExt, PermissionsExt}; + let metadata = fs::symlink_metadata(path)?; + if metadata.file_type().is_symlink() || metadata.uid() != unsafe { libc::geteuid() } { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + format!("refusing unsafe monitor state directory {}", path.display()), + )); + } + fs::set_permissions(path, fs::Permissions::from_mode(0o700))?; + } + Ok(()) +} + +fn claim_owner(paths: &StatePaths, owner_token: &str) -> io::Result { + let mut owner = OpenOptions::new() + .create(true) + .truncate(false) + .read(true) + .write(true) + .open(&paths.owner)?; + owner.try_lock_exclusive().map_err(|error| { + io::Error::new( + error.kind(), + format!("a monitor already owns {}: {error}", paths.root.display()), + ) + })?; + owner.set_len(0)?; + writeln!(owner, "{owner_token}")?; + owner.flush()?; + Ok(owner) +} + +fn owner_token(paths: &StatePaths) -> io::Result { + let token = fs::read_to_string(&paths.owner)?.trim().to_owned(); + if token.is_empty() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "monitor owner token is empty", + )); + } + Ok(token) +} + +fn stop_requested(paths: &StatePaths) -> bool { + owner_token(paths) + .map(|token| paths.stop_for(&token).exists()) + .unwrap_or(false) +} + +fn remove_active_stop(paths: &StatePaths) -> io::Result<()> { + let stop_path = paths.stop_for(&owner_token(paths)?); + match fs::remove_file(stop_path) { + Ok(()) => Ok(()), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(error), + } +} + +fn write_launch_status(paths: &StatePaths, token: Option<&str>, status: &str) { + if let Some(token) = token { + let _ = atomic_write(&paths.launch_status(token), status.as_bytes()); + } +} + +#[cfg(windows)] +fn configure_detached(command: &mut Command) { + use std::os::windows::process::CommandExt; + const CREATE_NEW_PROCESS_GROUP: u32 = 0x0000_0200; + const DETACHED_PROCESS: u32 = 0x0000_0008; + command.creation_flags(CREATE_NEW_PROCESS_GROUP | DETACHED_PROCESS); +} + +fn request_stop(state_key: &str, session_id: &str) -> io::Result<()> { + let paths = StatePaths::for_key(state_key, session_id); + let owner = OpenOptions::new() + .read(true) + .write(true) + .open(&paths.owner) + .map_err(|error| { + if error.kind() == io::ErrorKind::NotFound { + io::Error::new( + io::ErrorKind::NotFound, + format!("no monitor found for state key {state_key:?}"), + ) + } else { + error + } + })?; + if owner.try_lock_exclusive().is_ok() { + return Err(io::Error::new( + io::ErrorKind::NotFound, + format!("no monitor found for state key {state_key:?}"), + )); + } + let token = owner_token(&paths)?; + fs::write(paths.stop_for(&token), b"stop\n")?; + println!("stop requested for {}", paths.root.display()); + Ok(()) +} + +fn run_foreground( + state_key: &str, + label: &str, + instructions: &str, + session_id: &str, + if_running: RunningMode, + producer_command: &[OsString], + launch_token: Option<&str>, +) -> io::Result<()> { + run_foreground_after_claim( + state_key, + label, + instructions, + session_id, + if_running, + producer_command, + launch_token, + || {}, + ) +} + +fn run_foreground_after_claim( + state_key: &str, + label: &str, + instructions: &str, + session_id: &str, + if_running: RunningMode, + producer_command: &[OsString], + launch_token: Option<&str>, + after_claim: impl FnOnce(), +) -> io::Result<()> { + let paths = StatePaths::for_key(state_key, session_id); + ensure_private_directory(&paths.root)?; + let owner_token = launch_token + .map(str::to_owned) + .unwrap_or_else(next_pending_generation); + let _owner = claim_owner(&paths, &owner_token).inspect_err(|error| { + write_launch_status(&paths, launch_token, &format!("error: {error}")); + })?; + after_claim(); + if paths.stop_for(&owner_token).exists() { + log_line(&paths, "stop requested before producer start")?; + write_launch_status(&paths, launch_token, "ready"); + remove_active_stop(&paths)?; + return Ok(()); + } + let result = (|| { + run_producer( + &paths, + label, + instructions, + session_id, + if_running, + producer_command, + launch_token, + ) + })(); + if let Err(error) = &result { + let _ = log_line(&paths, &format!("monitor failed: {error}")); + write_launch_status(&paths, launch_token, &format!("error: {error}")); + } + result +} + +fn run_producer( + paths: &StatePaths, + label: &str, + instructions: &str, + session_id: &str, + if_running: RunningMode, + producer_command: &[OsString], + launch_token: Option<&str>, +) -> io::Result<()> { + let diagnostics = OpenOptions::new() + .create(true) + .append(true) + .open(&paths.log)?; + log_line( + paths, + &format!("starting producer: {}", render_command(producer_command)), + )?; + let mut pending = read_pending(&paths.pending)?; + + let mut command = Command::new(&producer_command[0]); + command + .args(&producer_command[1..]) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::from(diagnostics.try_clone()?)); + configure_producer(&mut command); + let mut child = command.spawn()?; + let producer_tree = match attach_producer_tree(&child) { + Ok(tree) => tree, + Err(error) => { + let _ = child.kill(); + let _ = child.wait(); + return Err(io::Error::new( + error.kind(), + format!("attach producer process tree: {error}"), + )); + } + }; + let mut producer = ProducerProcess { + child, + tree: Some(producer_tree), + }; + if let Err(error) = resume_producer(&producer.child) { + return Err(io::Error::new( + error.kind(), + format!("resume producer process: {error}"), + )); + } + write_launch_status(paths, launch_token, "ready"); + let stdout = producer.child.stdout.take().expect("stdout was piped"); + let (sender, receiver) = mpsc::sync_channel::(1024); + thread::spawn(move || forward_producer_output(stdout, sender)); + + let mut batch = Vec::new(); + let mut batch_deadline: Option = None; + let mut retry_deadline = Instant::now(); + let mut producer_status = None; + let mut receiver_closed = false; + let mut termination_requested = false; + let mut capture_failure = None; + + loop { + if stop_requested(paths) && !termination_requested { + log_line(paths, "stop requested")?; + producer.terminate_and_wait(); + termination_requested = true; + } + + let now = Instant::now(); + if batch_deadline.is_some_and(|deadline| now >= deadline) { + append_batch(paths, &mut pending, &mut batch)?; + batch_deadline = None; + flush_pending( + paths, + &mut pending, + label, + instructions, + session_id, + if_running, + )?; + retry_deadline = now + RETRY_INTERVAL; + } + if now >= retry_deadline { + flush_pending( + paths, + &mut pending, + label, + instructions, + session_id, + if_running, + )?; + retry_deadline = now + RETRY_INTERVAL; + } + + if producer_status.is_none() { + producer_status = producer.child.try_wait()?; + } + if producer_status.is_some() && receiver_closed { + break; + } + + let timeout = batch_deadline + .map(|deadline| deadline.saturating_duration_since(Instant::now())) + .unwrap_or(Duration::from_millis(100)) + .min(Duration::from_millis(100)); + match receiver.recv_timeout(timeout) { + Ok(ProducerOutputEvent::Data(line)) => { + if !batch.is_empty() && batch.len() + line.len() > MAX_DELIVERY_BYTES { + append_batch(paths, &mut pending, &mut batch)?; + flush_pending( + paths, + &mut pending, + label, + instructions, + session_id, + if_running, + )?; + retry_deadline = Instant::now() + RETRY_INTERVAL; + } + batch.extend_from_slice(&line); + if batch.len() >= MAX_DELIVERY_BYTES { + append_batch(paths, &mut pending, &mut batch)?; + flush_pending( + paths, + &mut pending, + label, + instructions, + session_id, + if_running, + )?; + retry_deadline = Instant::now() + RETRY_INTERVAL; + batch_deadline = None; + } else { + batch_deadline.get_or_insert(Instant::now() + BATCH_WINDOW); + } + } + Ok(ProducerOutputEvent::ReadError(error)) => { + capture_failure = Some(record_capture_failure(paths, &mut producer, error)?); + termination_requested = true; + } + Err(RecvTimeoutError::Timeout) => {} + Err(RecvTimeoutError::Disconnected) => receiver_closed = true, + } + } + + producer.terminate_and_wait(); + append_batch(paths, &mut pending, &mut batch)?; + let status = producer_status.expect("producer has exited"); + let summary = capture_failure + .as_ref() + .map(|(_, message)| format!("[monitor] {message}\n")) + .unwrap_or_else(|| format!("[monitor] producer exited with status {status}\n")); + append_pending(&paths.pending, &mut pending, summary.as_bytes())?; + flush_pending( + paths, + &mut pending, + label, + instructions, + session_id, + if_running, + )?; + + while !pending.bytes.is_empty() && !stop_requested(paths) { + thread::sleep(RETRY_INTERVAL); + flush_pending( + paths, + &mut pending, + label, + instructions, + session_id, + if_running, + )?; + } + if !pending.bytes.is_empty() && stop_requested(paths) { + pending.bytes.clear(); + pending.rotate(); + persist_pending(&paths.pending, &pending)?; + log_line(paths, "discarded undelivered output after explicit stop")?; + } + remove_active_stop(paths)?; + if let Some((kind, message)) = capture_failure { + return Err(io::Error::new(kind, message)); + } + log_line(paths, &format!("producer exited with status {status}"))?; + Ok(()) +} + +fn record_capture_failure( + paths: &StatePaths, + producer: &mut ProducerProcess, + error: io::Error, +) -> io::Result<(io::ErrorKind, String)> { + let message = format!("producer stdout capture failed: {error}"); + log_line(paths, &message)?; + producer.terminate_and_wait(); + Ok((error.kind(), message)) +} + +fn forward_producer_output(mut reader: R, sender: mpsc::SyncSender) { + let mut read_buffer = [0_u8; 8 * 1024]; + let mut record = Vec::with_capacity(MAX_DELIVERY_BYTES); + loop { + let read = match reader.read(&mut read_buffer) { + Ok(0) => break, + Ok(read) => read, + Err(error) => { + if !send_output_record(&sender, &mut record) { + return; + } + let _ = sender.send(ProducerOutputEvent::ReadError(error)); + return; + } + }; + for byte in &read_buffer[..read] { + record.push(*byte); + if *byte == b'\n' || record.len() == MAX_DELIVERY_BYTES { + if sender + .send(ProducerOutputEvent::Data(std::mem::take(&mut record))) + .is_err() + { + return; + } + record = Vec::with_capacity(MAX_DELIVERY_BYTES); + } + } + } + let _ = send_output_record(&sender, &mut record); +} + +fn send_output_record( + sender: &mpsc::SyncSender, + record: &mut Vec, +) -> bool { + if !record.is_empty() { + if !record.ends_with(b"\n") { + record.push(b'\n'); + } + return sender + .send(ProducerOutputEvent::Data(std::mem::take(record))) + .is_ok(); + } + true +} + +fn append_batch( + paths: &StatePaths, + pending: &mut PendingState, + batch: &mut Vec, +) -> io::Result<()> { + if batch.is_empty() { + return Ok(()); + } + append_pending(&paths.pending, pending, batch)?; + batch.clear(); + Ok(()) +} + +fn flush_pending( + paths: &StatePaths, + pending: &mut PendingState, + label: &str, + instructions: &str, + session_id: &str, + if_running: RunningMode, +) -> io::Result<()> { + while !pending.bytes.is_empty() { + if pending.active_len == 0 { + pending.active_len = pending_chunk_end(&pending.bytes, MAX_DELIVERY_BYTES); + persist_pending(&paths.pending, pending)?; + } + let end = pending.active_len; + let prompt = build_delivery_prompt(label, &pending.bytes[..end], instructions)?; + let delivery_id = pending.delivery_id(paths); + if !deliver_to_session(paths, session_id, &prompt, if_running, &delivery_id) { + log_line(paths, "delivery failed; buffered output will be retried")?; + return Ok(()); + } + pending.bytes.drain(..end); + pending.rotate(); + persist_pending(&paths.pending, pending)?; + log_line(paths, "delivered one event batch")?; + } + Ok(()) +} + +fn build_delivery_prompt(label: &str, bytes: &[u8], instructions: &str) -> io::Result { + let text = String::from_utf8_lossy(bytes).replace('\0', "\u{fffd}"); + let mut prompt = format!( + "[monitor: {label} | pid {}]\n{}", + std::process::id(), + text.trim_end() + ); + if !instructions.is_empty() { + prompt.push_str("\n\n"); + prompt.push_str(instructions); + } + if prompt.encode_utf16().count() > MAX_PROMPT_CODE_UNITS { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "monitor delivery prompt exceeded its internal size bound", + )); + } + Ok(prompt) +} + +fn deliver_to_session( + paths: &StatePaths, + session_id: &str, + prompt: &str, + if_running: RunningMode, + delivery_id: &str, +) -> bool { + deliver_with_candidates( + paths, + session_id, + prompt, + if_running, + delivery_id, + lock_candidates(), + berdctl_candidates(), + DELIVERY_TIMEOUT, + ) +} + +fn deliver_with_candidates( + paths: &StatePaths, + session_id: &str, + prompt: &str, + if_running: RunningMode, + delivery_id: &str, + locks: Vec, + binaries: Vec, + timeout: Duration, +) -> bool { + for lock in locks { + for binary in &binaries { + if stop_requested(paths) { + return false; + } + let mut child = match Command::new(binary) + .arg("--lock-path") + .arg(&lock) + .arg("--timeout-ms") + .arg("10000") + .arg("session") + .arg("send") + .arg("--session-id") + .arg(session_id) + .arg("--prompt") + .arg(prompt) + .arg("--if-running") + .arg(if_running.as_str()) + .arg("--delivery-id") + .arg(delivery_id) + .arg("--from") + .arg("berd-monitor") + .arg("--json") + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + { + Ok(child) => child, + Err(_) => continue, + }; + let deadline = Instant::now() + timeout; + loop { + if stop_requested(paths) { + let _ = child.kill(); + let _ = child.wait(); + return false; + } + if Instant::now() >= deadline { + let _ = child.kill(); + let _ = child.wait(); + break; + } + match child.try_wait() { + Ok(Some(status)) if status.success() => return true, + Ok(Some(_)) | Err(_) => break, + Ok(None) => thread::sleep(DELIVERY_POLL_INTERVAL), + } + } + } + } + false +} + +fn lock_candidates() -> Vec { + let Some(explicit) = env::var_os("BERDCTL_LOCK").map(PathBuf::from) else { + return Vec::new(); + }; + lock_candidates_for(&explicit) +} + +fn lock_candidates_for(explicit: &Path) -> Vec { + let mut candidates = vec![explicit.to_path_buf()]; + if let Some(parent) = explicit.parent() { + let mut siblings = fs::read_dir(parent) + .into_iter() + .flatten() + .flatten() + .filter_map(|entry| { + let name = entry.file_name(); + let name = name.to_string_lossy(); + if name.starts_with("control-") && name.ends_with(".json") { + let modified = entry + .metadata() + .ok()? + .modified() + .ok() + .unwrap_or(SystemTime::UNIX_EPOCH); + Some((modified, entry.path())) + } else { + None + } + }) + .collect::>(); + siblings.sort_by(|left, right| right.0.cmp(&left.0)); + for (_, sibling) in siblings + .into_iter() + .take(MAX_LOCK_CANDIDATES.saturating_sub(1)) + { + if sibling != explicit { + candidates.push(sibling); + } + } + } + candidates +} + +fn berdctl_candidates() -> Vec { + let mut candidates = Vec::new(); + if let Some(explicit) = env::var_os("BERDCTL_BIN") { + candidates.push(explicit); + } + let default = OsString::from(if cfg!(windows) { + "berdctl.exe" + } else { + "berdctl" + }); + if !candidates.contains(&default) { + candidates.push(default); + } + candidates +} + +fn pending_chunk_end(pending: &[u8], limit: usize) -> usize { + if pending.len() <= limit { + return pending.len(); + } + let mut hard_end = limit; + if let Err(error) = std::str::from_utf8(&pending[..hard_end]) { + if error.error_len().is_none() { + hard_end = error.valid_up_to(); + } + } + let end = pending[..hard_end] + .iter() + .rposition(|byte| *byte == b'\n') + .map(|index| index + 1) + .unwrap_or(hard_end); + end.max(1) +} + +fn read_optional(path: &Path) -> io::Result> { + match File::open(path) { + Ok(mut file) => { + let mut contents = Vec::new(); + file.read_to_end(&mut contents)?; + Ok(contents) + } + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(Vec::new()), + Err(error) => Err(error), + } +} + +fn read_pending(path: &Path) -> io::Result { + let contents = read_optional(path)?; + if contents.is_empty() { + return Ok(PendingState::empty()); + } + if !contents.starts_with(PENDING_HEADER_PREFIX.as_bytes()) { + return Ok(PendingState { + generation: next_pending_generation(), + active_len: 0, + bytes: contents, + }); + } + let header_end = contents + .iter() + .position(|byte| *byte == b'\n') + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "invalid pending header"))?; + let header = std::str::from_utf8(&contents[PENDING_HEADER_PREFIX.len()..header_end]) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "invalid pending header"))?; + let (generation, active_len) = header + .split_once(' ') + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "invalid pending header"))?; + let active_len = active_len + .parse::() + .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "invalid pending length"))?; + let bytes = contents[(header_end + 1)..].to_vec(); + if generation.is_empty() || active_len > bytes.len() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "invalid pending delivery state", + )); + } + Ok(PendingState { + generation: generation.to_owned(), + active_len, + bytes, + }) +} + +fn persist_pending(path: &Path, pending: &PendingState) -> io::Result<()> { + if pending.bytes.is_empty() { + return match fs::remove_file(path) { + Ok(()) => Ok(()), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(error), + }; + } + let mut contents = format!( + "{PENDING_HEADER_PREFIX}{} {}\n", + pending.generation, pending.active_len + ) + .into_bytes(); + contents.extend_from_slice(&pending.bytes); + atomic_write(path, &contents) +} + +fn append_pending(path: &Path, pending: &mut PendingState, data: &[u8]) -> io::Result<()> { + if data.is_empty() { + return Ok(()); + } + let existed = path.exists(); + pending.bytes.extend_from_slice(data); + if existed { + append_file(path, data) + } else { + persist_pending(path, pending) + } +} + +fn append_file(path: &Path, data: &[u8]) -> io::Result<()> { + OpenOptions::new() + .create(true) + .append(true) + .open(path)? + .write_all(data) +} + +fn atomic_write(path: &Path, data: &[u8]) -> io::Result<()> { + atomic_write_with(path, data, replace_file_atomically) +} + +fn atomic_write_with( + path: &Path, + data: &[u8], + replace: impl FnOnce(&Path, &Path) -> io::Result<()>, +) -> io::Result<()> { + let temporary = path.with_extension(format!("tmp.{}", std::process::id())); + let mut file = OpenOptions::new() + .create(true) + .truncate(true) + .write(true) + .open(&temporary)?; + file.write_all(data)?; + file.sync_all()?; + drop(file); + + if let Err(error) = replace(&temporary, path) { + let _ = fs::remove_file(&temporary); + return Err(error); + } + + #[cfg(unix)] + if let Some(parent) = path.parent() { + File::open(parent)?.sync_all()?; + } + Ok(()) +} + +#[cfg(unix)] +fn replace_file_atomically(source: &Path, destination: &Path) -> io::Result<()> { + fs::rename(source, destination) +} + +#[cfg(windows)] +fn replace_file_atomically(source: &Path, destination: &Path) -> io::Result<()> { + use std::os::windows::ffi::OsStrExt; + use windows_sys::Win32::Storage::FileSystem::{ + MoveFileExW, MOVEFILE_REPLACE_EXISTING, MOVEFILE_WRITE_THROUGH, + }; + + let source = source + .as_os_str() + .encode_wide() + .chain(std::iter::once(0)) + .collect::>(); + let destination = destination + .as_os_str() + .encode_wide() + .chain(std::iter::once(0)) + .collect::>(); + let result = unsafe { + MoveFileExW( + source.as_ptr(), + destination.as_ptr(), + MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH, + ) + }; + if result == 0 { + Err(io::Error::last_os_error()) + } else { + Ok(()) + } +} + +fn log_line(paths: &StatePaths, message: &str) -> io::Result<()> { + writeln!( + OpenOptions::new() + .create(true) + .append(true) + .open(&paths.log)?, + "{message}" + ) +} + +fn render_command(command: &[OsString]) -> String { + command + .iter() + .map(|part| part.to_string_lossy()) + .collect::>() + .join(" ") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[cfg(unix)] + #[test] + fn launch_timeout_terminates_and_reaps_the_child() { + let key = format!( + "launch-timeout-test-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + let paths = StatePaths::for_key(&key, "test-session"); + ensure_private_directory(&paths.root).unwrap(); + let status_path = paths.launch_status("never-ready"); + let stop_path = paths.stop_for("never-ready"); + let mut child = Command::new("sh") + .args(["-c", "exec sleep 30"]) + .spawn() + .unwrap(); + let pid = child.id(); + + let launch_timeout = Duration::from_millis(100); + let launch_result = + wait_for_launch(&mut child, &paths, &status_path, &stop_path, launch_timeout); + let error = launch_result.unwrap_err(); + + assert!(error.contains("did not become ready")); + assert!(!test_process_exists(pid)); + assert!(!status_path.exists()); + assert!(!stop_path.exists()); + fs::remove_dir_all(paths.root).unwrap(); + } + + #[test] + fn failed_atomic_replace_preserves_pending_output() { + let root = env::temp_dir().join(format!( + "berd-monitor-atomic-write-test-{}", + std::process::id() + )); + fs::create_dir_all(&root).unwrap(); + let pending = root.join("pending"); + fs::write(&pending, b"still pending").unwrap(); + + let error = atomic_write_with(&pending, b"replacement", |_, _| { + Err(io::Error::other("injected replacement failure")) + }) + .unwrap_err(); + + assert_eq!(error.kind(), io::ErrorKind::Other); + assert_eq!(fs::read(&pending).unwrap(), b"still pending"); + assert!(!pending + .with_extension(format!("tmp.{}", std::process::id())) + .exists()); + fs::remove_dir_all(root).unwrap(); + } + + #[test] + fn pending_delivery_id_survives_restart_and_rotates_after_ack() { + let key = format!( + "pending-id-test-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + let paths = StatePaths::for_key(&key, "test-session"); + ensure_private_directory(&paths.root).unwrap(); + let mut pending = PendingState::empty(); + append_pending(&paths.pending, &mut pending, b"same event\n").unwrap(); + pending.active_len = pending.bytes.len(); + persist_pending(&paths.pending, &pending).unwrap(); + let first_id = pending.delivery_id(&paths); + + let mut restarted = read_pending(&paths.pending).unwrap(); + assert_eq!(restarted.delivery_id(&paths), first_id); + assert_eq!(restarted.active_len, restarted.bytes.len()); + + restarted.bytes.clear(); + restarted.rotate(); + persist_pending(&paths.pending, &restarted).unwrap(); + append_pending(&paths.pending, &mut restarted, b"same event\n").unwrap(); + restarted.active_len = restarted.bytes.len(); + persist_pending(&paths.pending, &restarted).unwrap(); + + assert_ne!(restarted.delivery_id(&paths), first_id); + fs::remove_dir_all(paths.root).unwrap(); + } + + #[cfg(unix)] + fn test_process_exists(pid: u32) -> bool { + let result = unsafe { libc::kill(pid as i32, 0) }; + result == 0 || io::Error::last_os_error().raw_os_error() == Some(libc::EPERM) + } + + #[cfg(unix)] + #[test] + fn producer_guard_cleans_up_the_process_group_on_error() { + use std::io::{BufRead, BufReader}; + + let mut observed_pids = None; + let result: io::Result<()> = (|| { + let mut command = Command::new("sh"); + command + .args(["-c", "sleep 30 & child=$!; echo $child; wait"]) + .stdout(Stdio::piped()); + configure_producer(&mut command); + let child = command.spawn()?; + let tree = attach_producer_tree(&child)?; + let mut producer = ProducerProcess { + child, + tree: Some(tree), + }; + resume_producer(&producer.child)?; + let direct_pid = producer.child.id(); + let mut descendant = String::new(); + let stdout = producer.child.stdout.take().unwrap(); + let mut reader = BufReader::new(stdout); + reader.read_line(&mut descendant)?; + let descendant_pid = descendant.trim().parse::().unwrap(); + observed_pids = Some((direct_pid, descendant_pid)); + Err(io::Error::other("injected post-spawn failure")) + })(); + + assert_eq!(result.unwrap_err().kind(), io::ErrorKind::Other); + let (direct_pid, descendant_pid) = observed_pids.unwrap(); + let deadline = Instant::now() + Duration::from_secs(2); + while (test_process_exists(direct_pid) || test_process_exists(descendant_pid)) + && Instant::now() < deadline + { + thread::sleep(Duration::from_millis(25)); + } + assert!(!test_process_exists(direct_pid)); + assert!(!test_process_exists(descendant_pid)); + } + + #[cfg(unix)] + #[test] + fn explicit_producer_cleanup_disarms_drop() { + let mut command = Command::new("sh"); + command.args(["-c", "exec sleep 30"]); + configure_producer(&mut command); + let child = command.spawn().unwrap(); + let tree = attach_producer_tree(&child).unwrap(); + let mut producer = ProducerProcess { + child, + tree: Some(tree), + }; + resume_producer(&producer.child).unwrap(); + let mut termination_calls = 0; + + producer.terminate_and_wait_with(|child, tree| { + termination_calls += 1; + terminate_producer_tree(child, tree); + }); + producer.terminate_and_wait_with(|_, _| termination_calls += 1); + drop(producer); + + assert_eq!(termination_calls, 1); + } + + #[test] + fn chunk_prefers_complete_lines() { + let input = b"first\nsecond\nthird\n"; + assert_eq!(pending_chunk_end(input, 13), 13); + assert_eq!(&input[..pending_chunk_end(input, 10)], b"first\n"); + } + + #[test] + fn chunk_never_exceeds_limit_for_a_long_line() { + let input = vec![b'x'; MAX_DELIVERY_BYTES + 10_000]; + assert_eq!( + pending_chunk_end(&input, MAX_DELIVERY_BYTES), + MAX_DELIVERY_BYTES + ); + } + + #[test] + fn producer_output_splits_unterminated_records_before_they_can_grow_unbounded() { + let input = vec![b'x'; MAX_DELIVERY_BYTES * 3 + 17]; + let (sender, receiver) = mpsc::sync_channel(8); + + forward_producer_output(io::Cursor::new(&input), sender); + let records = receiver + .into_iter() + .map(|event| match event { + ProducerOutputEvent::Data(record) => record, + ProducerOutputEvent::ReadError(error) => { + panic!("unexpected capture failure: {error}") + } + }) + .collect::>(); + + assert_eq!(records.len(), 4); + assert!(records + .iter() + .all(|record| record.len() <= MAX_DELIVERY_BYTES)); + let mut output = records.concat(); + assert_eq!(output.pop(), Some(b'\n')); + assert_eq!(output, input); + } + + #[test] + fn producer_output_preserves_partial_record_before_read_error() { + struct BytesThenError(bool); + + impl Read for BytesThenError { + fn read(&mut self, buffer: &mut [u8]) -> io::Result { + if self.0 { + return Err(io::Error::other("injected stdout read failure")); + } + self.0 = true; + let bytes = b"partial output"; + buffer[..bytes.len()].copy_from_slice(bytes); + Ok(bytes.len()) + } + } + + let (sender, receiver) = mpsc::sync_channel(2); + forward_producer_output(BytesThenError(false), sender); + let events = receiver.into_iter().collect::>(); + + assert_eq!(events.len(), 2); + assert!(matches!( + &events[0], + ProducerOutputEvent::Data(bytes) if bytes == b"partial output\n" + )); + assert!(matches!( + &events[1], + ProducerOutputEvent::ReadError(error) + if error.to_string() == "injected stdout read failure" + )); + } + + #[cfg(unix)] + #[test] + fn capture_failure_is_recorded_and_cleans_up_the_process_group() { + use std::io::{BufRead, BufReader}; + + let key = format!( + "capture-failure-test-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + let paths = StatePaths::for_key(&key, "test-session"); + ensure_private_directory(&paths.root).unwrap(); + let mut command = Command::new("sh"); + command + .args(["-c", "sleep 30 & child=$!; echo $child; wait"]) + .stdout(Stdio::piped()); + configure_producer(&mut command); + let child = command.spawn().unwrap(); + let tree = attach_producer_tree(&child).unwrap(); + let mut producer = ProducerProcess { + child, + tree: Some(tree), + }; + resume_producer(&producer.child).unwrap(); + let direct_pid = producer.child.id(); + let mut descendant = String::new(); + BufReader::new(producer.child.stdout.take().unwrap()) + .read_line(&mut descendant) + .unwrap(); + let descendant_pid = descendant.trim().parse::().unwrap(); + + let failure = record_capture_failure( + &paths, + &mut producer, + io::Error::other("injected stdout read failure"), + ) + .unwrap(); + + assert_eq!(failure.0, io::ErrorKind::Other); + assert!(failure.1.contains("injected stdout read failure")); + assert!(fs::read_to_string(&paths.log) + .unwrap() + .contains("producer stdout capture failed")); + let deadline = Instant::now() + Duration::from_secs(2); + while (test_process_exists(direct_pid) || test_process_exists(descendant_pid)) + && Instant::now() < deadline + { + thread::sleep(Duration::from_millis(25)); + } + assert!(!test_process_exists(direct_pid)); + assert!(!test_process_exists(descendant_pid)); + fs::remove_dir_all(paths.root).unwrap(); + } + + #[test] + fn delivery_prompt_replaces_embedded_nul_bytes_without_growing() { + let prompt = build_delivery_prompt("test", b"before\0after\n", "continue").unwrap(); + + assert!(!prompt.contains('\0')); + assert!(prompt.contains("before\u{fffd}after")); + assert!(prompt.contains("continue")); + } + + #[cfg(unix)] + #[test] + fn nul_output_is_delivered_once_and_does_not_block_later_output() { + use std::os::unix::fs::PermissionsExt; + + let key = format!( + "delivery-nul-test-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + let session_id = "test-session"; + let paths = StatePaths::for_key(&key, session_id); + ensure_private_directory(&paths.root).unwrap(); + let capture = paths.root.join("delivered.txt"); + let fake_berdctl = paths.root.join("fake-berdctl"); + fs::write( + &fake_berdctl, + format!( + "#!/bin/sh\nprintf '%s\\n' \"$@\" >> '{}'\n", + capture.display() + ), + ) + .unwrap(); + fs::set_permissions(&fake_berdctl, fs::Permissions::from_mode(0o700)).unwrap(); + let binary = fake_berdctl.into_os_string(); + + let first = build_delivery_prompt("test", b"before\0after\n", "").unwrap(); + assert!(deliver_with_candidates( + &paths, + session_id, + &first, + RunningMode::Steer, + "nul-delivery", + vec![PathBuf::from("stale")], + vec![binary.clone()], + Duration::from_secs(2), + )); + let later = build_delivery_prompt("test", b"later output\n", "").unwrap(); + assert!(deliver_with_candidates( + &paths, + session_id, + &later, + RunningMode::Steer, + "later-delivery", + vec![PathBuf::from("stale")], + vec![binary], + Duration::from_secs(2), + )); + + let delivered = fs::read_to_string(capture).unwrap(); + assert_eq!(delivered.matches("before\u{fffd}after").count(), 1); + assert!(delivered.contains("later output")); + fs::remove_dir_all(paths.root).unwrap(); + } + + #[cfg(unix)] + #[test] + fn stop_interrupts_a_nonresponsive_delivery_probe() { + use std::os::unix::fs::PermissionsExt; + + let key = format!( + "delivery-stop-test-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + let session_id = "test-session"; + let paths = StatePaths::for_key(&key, session_id); + ensure_private_directory(&paths.root).unwrap(); + let owner = claim_owner(&paths, "delivery-stop-owner").unwrap(); + let fake_berdctl = paths.root.join("fake-berdctl"); + fs::write(&fake_berdctl, "#!/bin/sh\nexec sleep 30\n").unwrap(); + fs::set_permissions(&fake_berdctl, fs::Permissions::from_mode(0o700)).unwrap(); + + let worker_key = key.clone(); + let worker_binary = fake_berdctl.into_os_string(); + let started = Instant::now(); + let worker = thread::spawn(move || { + let worker_paths = StatePaths::for_key(&worker_key, session_id); + deliver_with_candidates( + &worker_paths, + session_id, + "test", + RunningMode::Steer, + "test-delivery", + vec![PathBuf::from("stale-a"), PathBuf::from("stale-b")], + vec![worker_binary], + DELIVERY_TIMEOUT, + ) + }); + thread::sleep(Duration::from_millis(150)); + request_stop(&key, session_id).unwrap(); + + assert!(!worker.join().unwrap()); + assert!(started.elapsed() < Duration::from_secs(2)); + drop(owner); + fs::remove_dir_all(&paths.root).unwrap(); + } + + #[cfg(unix)] + #[test] + fn delivery_timeout_terminates_and_reaps_the_probe() { + use std::os::unix::fs::PermissionsExt; + + let key = format!( + "delivery-timeout-test-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + let session_id = "test-session"; + let paths = StatePaths::for_key(&key, session_id); + ensure_private_directory(&paths.root).unwrap(); + let pid_file = paths.root.join("probe.pid"); + let fake_berdctl = paths.root.join("fake-berdctl"); + fs::write( + &fake_berdctl, + format!( + "#!/bin/sh\nprintf '%s\\n' \"$$\" > '{}'\nexec sleep 30\n", + pid_file.display() + ), + ) + .unwrap(); + fs::set_permissions(&fake_berdctl, fs::Permissions::from_mode(0o700)).unwrap(); + + let started = Instant::now(); + assert!(!deliver_with_candidates( + &paths, + session_id, + "test", + RunningMode::Steer, + "test-delivery", + vec![PathBuf::from("stale")], + vec![fake_berdctl.into_os_string()], + Duration::from_millis(500), + )); + + let pid = fs::read_to_string(pid_file) + .unwrap() + .trim() + .parse::() + .unwrap(); + assert!(started.elapsed() < Duration::from_secs(2)); + assert!(!test_process_exists(pid)); + fs::remove_dir_all(paths.root).unwrap(); + } + + #[test] + fn lock_candidates_keep_the_explicit_lock_and_bound_siblings() { + let root = env::temp_dir().join(format!( + "berd-monitor-locks-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + fs::create_dir_all(&root).unwrap(); + let explicit = root.join("control-explicit.json"); + fs::write(&explicit, "{}").unwrap(); + for index in 0..(MAX_LOCK_CANDIDATES * 2) { + fs::write(root.join(format!("control-{index}.json")), "{}").unwrap(); + } + + let candidates = lock_candidates_for(&explicit); + + assert_eq!(candidates.first(), Some(&explicit)); + assert_eq!(candidates.len(), MAX_LOCK_CANDIDATES); + fs::remove_dir_all(root).unwrap(); + } + + #[test] + fn state_key_path_is_stable_and_safe() { + let first = StatePaths::for_key("pr/123 checks", "session-a"); + let second = StatePaths::for_key("pr/123 checks", "session-a"); + assert_eq!(first.root, second.root); + assert!(first.root.to_string_lossy().contains("pr-123-checks-")); + assert_ne!( + first.root, + StatePaths::for_key("pr/123 checks", "session-b").root + ); + } + + #[test] + fn active_owner_rejects_a_duplicate_monitor() { + let key = format!( + "owner-test-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + let paths = StatePaths::for_key(&key, "test-session"); + ensure_private_directory(&paths.root).unwrap(); + let owner = claim_owner(&paths, "owner-a").unwrap(); + assert_eq!( + claim_owner(&paths, "owner-b").unwrap_err().kind(), + io::ErrorKind::WouldBlock + ); + drop(owner); + claim_owner(&paths, "owner-b").unwrap(); + fs::remove_dir_all(&paths.root).unwrap(); + } + + #[test] + fn simultaneous_owner_claims_have_one_winner() { + use std::sync::{Arc, Barrier}; + + let key = format!( + "owner-race-test-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + let paths = StatePaths::for_key(&key, "test-session"); + ensure_private_directory(&paths.root).unwrap(); + let barrier = Arc::new(Barrier::new(3)); + let mut workers = Vec::new(); + for _ in 0..2 { + let barrier = Arc::clone(&barrier); + let key = key.clone(); + workers.push(thread::spawn(move || { + barrier.wait(); + let owner = claim_owner(&StatePaths::for_key(&key, "test-session"), "race-owner"); + if owner.is_ok() { + thread::sleep(Duration::from_millis(100)); + } + owner.is_ok() + })); + } + barrier.wait(); + let winners = workers + .into_iter() + .map(|worker| worker.join().unwrap()) + .filter(|won| *won) + .count(); + assert_eq!(winners, 1); + fs::remove_dir_all(&paths.root).unwrap(); + } + + #[test] + fn missing_producer_is_recorded_as_a_launch_failure() { + let key = format!( + "missing-producer-test-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + let paths = StatePaths::for_key(&key, "test-session"); + let error = run_foreground( + &key, + "test", + "", + "test-session", + RunningMode::Steer, + &[OsString::from("berd-monitor-command-that-does-not-exist")], + None, + ) + .unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::NotFound); + assert!(fs::read_to_string(&paths.log) + .unwrap() + .contains("monitor failed")); + fs::remove_dir_all(&paths.root).unwrap(); + } + + #[cfg(unix)] + #[test] + fn stop_after_owner_claim_prevents_producer_start() { + let key = format!( + "startup-stop-race-test-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + let session_id = "test-session"; + let paths = StatePaths::for_key(&key, session_id); + let producer_started = paths.root.join("producer-started"); + let command = format!("printf started > '{}'", producer_started.display()); + + run_foreground_after_claim( + &key, + "test", + "", + session_id, + RunningMode::Steer, + &[ + OsString::from("sh"), + OsString::from("-c"), + OsString::from(command), + ], + Some("startup-stop-owner"), + || request_stop(&key, session_id).unwrap(), + ) + .unwrap(); + + assert!(!producer_started.exists()); + assert!(!paths.stop_for("startup-stop-owner").exists()); + assert!(fs::read_to_string(&paths.log) + .unwrap() + .contains("stop requested before producer start")); + fs::remove_dir_all(&paths.root).unwrap(); + } + + #[cfg(unix)] + #[test] + fn stop_terminates_the_producer_process_group_and_discards_pending() { + let key = format!( + "process-tree-test-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + let session_id = "test-session"; + let paths = StatePaths::for_key(&key, session_id); + let worker_key = key.clone(); + let worker = thread::spawn(move || { + run_foreground( + &worker_key, + "test", + "", + session_id, + RunningMode::Steer, + &[ + OsString::from("sh"), + OsString::from("-c"), + OsString::from("sleep 30 & child=$!; echo $child; wait"), + ], + None, + ) + }); + let deadline = Instant::now() + Duration::from_secs(5); + while !paths.pending.is_file() && Instant::now() < deadline { + thread::sleep(Duration::from_millis(25)); + } + let child_pid = loop { + if let Ok(pending) = read_pending(&paths.pending) { + let pending_text = String::from_utf8_lossy(&pending.bytes); + let parsed_pid = pending_text.trim().parse::(); + if let Ok(pid) = parsed_pid { + break pid; + } + } + assert!( + Instant::now() < deadline, + "producer output was not buffered" + ); + thread::sleep(Duration::from_millis(25)); + }; + request_stop(&key, session_id).unwrap(); + worker.join().unwrap().unwrap(); + assert!(read_optional(&paths.pending).unwrap().is_empty()); + let child_gone_deadline = Instant::now() + Duration::from_secs(2); + while test_process_exists(child_pid as u32) && Instant::now() < child_gone_deadline { + thread::sleep(Duration::from_millis(25)); + } + assert!(!test_process_exists(child_pid as u32)); + fs::remove_dir_all(&paths.root).unwrap(); + } + + #[test] + fn running_mode_matches_berdctl_values() { + assert_eq!(RunningMode::Steer.as_str(), "steer"); + assert_eq!(RunningMode::Queue.as_str(), "queue"); + } +} diff --git a/src-tauri/crates/berdctl/api-surface-feedback.json b/src-tauri/crates/berdctl/api-surface-feedback.json index e3e404ac5..e5807acb7 100644 --- a/src-tauri/crates/berdctl/api-surface-feedback.json +++ b/src-tauri/crates/berdctl/api-surface-feedback.json @@ -6,7 +6,7 @@ "description": "Manage the user's chat sessions: create (fire-and-forget, on any installed agent harness), send, open, list, get, rename, move, move to group, clear project, fork, archive.", "actions": { "create": { - "description": "Create a new chat session on any installed agent harness and send the prompt in it. Fire-and-forget: returns the session id immediately and the session runs in the background without changing what the user sees; the user can open it themselves. Only check on it later (action \"get\") if the user asks.", + "description": "Create a new chat session on any installed agent harness and send the prompt in it. Fire-and-forget: returns the session id immediately and the session runs in the background without changing what the user sees; the user can open it themselves. Use --from to give the delegating session or tool a concise visible label on the initial message. Only check on it later (action \"get\") if the user asks.", "fields": [ { "name": "prompt", @@ -47,6 +47,14 @@ "description": "Branch/worktree name when the project's startup mode is branch or worktree; required for those modes.", "min": 1, "max": 200 + }, + { + "name": "from", + "required": false, + "kind": "string", + "description": "Optional visible sender label for the initial message (1-120 chars).", + "min": 1, + "max": 120 } ], "schema": { @@ -80,6 +88,13 @@ "type": "string", "minLength": 1, "maxLength": 200 + }, + "from": { + "description": "Optional visible sender label for the initial message (1-120 chars).", + "type": "string", + "minLength": 1, + "maxLength": 120, + "pattern": "^[^\\r\\n]*$" } }, "required": ["prompt"], @@ -87,7 +102,7 @@ } }, "send": { - "description": "Send a prompt into an existing chat session without opening or focusing it. Idle sends are fire-and-forget and visibly add a user message marked as sent by Berd from another session. Running sessions are refused by default; use --if-running steer to add context to the active run, or --if-running queue to send one follow-up after the current run finishes.", + "description": "Send a prompt into an existing chat session without opening or focusing it. Idle sends are fire-and-forget and visibly add a user message marked as sent by Berd from another session. Running sessions are refused by default; use --if-running steer to add context to the active run, or --if-running queue to send one follow-up after the current run finishes. Use --from to give the sending session or tool a concise visible label in the transcript. Use --delivery-id when retries must create at most one user turn.", "fields": [ { "name": "session_id", @@ -117,6 +132,22 @@ "kind": "string", "description": "What to do if the target session is running: refuse, steer, or queue.", "values": ["refuse", "steer", "queue"] + }, + { + "name": "from", + "required": false, + "kind": "string", + "description": "Optional visible sender label for this message (1-120 chars).", + "min": 1, + "max": 120 + }, + { + "name": "delivery_id", + "required": false, + "kind": "string", + "description": "Optional idempotency id; a repeated id for this session is accepted without creating another user turn (1-200 chars).", + "min": 1, + "max": 200 } ], "schema": { @@ -144,6 +175,20 @@ "description": "What to do if the target session is running: refuse, steer, or queue.", "type": "string", "enum": ["refuse", "steer", "queue"] + }, + "from": { + "description": "Optional visible sender label for this message (1-120 chars).", + "type": "string", + "minLength": 1, + "maxLength": 120, + "pattern": "^[^\\r\\n]*$" + }, + "delivery_id": { + "description": "Optional idempotency id; a repeated id for this session is accepted without creating another user turn (1-200 chars).", + "type": "string", + "minLength": 1, + "maxLength": 200, + "pattern": "^[^\\r\\n]*$" } }, "required": ["session_id", "prompt"], diff --git a/src-tauri/crates/berdctl/api-surface.json b/src-tauri/crates/berdctl/api-surface.json index 18c5163b0..9aaf40f05 100644 --- a/src-tauri/crates/berdctl/api-surface.json +++ b/src-tauri/crates/berdctl/api-surface.json @@ -6,7 +6,7 @@ "description": "Manage the user's chat sessions: create (fire-and-forget, on any installed agent harness), send, open, list, get, rename, move, move to group, clear project, fork, archive.", "actions": { "create": { - "description": "Create a new chat session on any installed agent harness and send the prompt in it. Fire-and-forget: returns the session id immediately and the session runs in the background without changing what the user sees; the user can open it themselves. Only check on it later (action \"get\") if the user asks.", + "description": "Create a new chat session on any installed agent harness and send the prompt in it. Fire-and-forget: returns the session id immediately and the session runs in the background without changing what the user sees; the user can open it themselves. Use --from to give the delegating session or tool a concise visible label on the initial message. Only check on it later (action \"get\") if the user asks.", "fields": [ { "name": "prompt", @@ -47,6 +47,14 @@ "description": "Branch/worktree name when the project's startup mode is branch or worktree; required for those modes.", "min": 1, "max": 200 + }, + { + "name": "from", + "required": false, + "kind": "string", + "description": "Optional visible sender label for the initial message (1-120 chars).", + "min": 1, + "max": 120 } ], "schema": { @@ -80,6 +88,13 @@ "type": "string", "minLength": 1, "maxLength": 200 + }, + "from": { + "description": "Optional visible sender label for the initial message (1-120 chars).", + "type": "string", + "minLength": 1, + "maxLength": 120, + "pattern": "^[^\\r\\n]*$" } }, "required": ["prompt"], @@ -87,7 +102,7 @@ } }, "send": { - "description": "Send a prompt into an existing chat session without opening or focusing it. Idle sends are fire-and-forget and visibly add a user message marked as sent by Berd from another session. Running sessions are refused by default; use --if-running steer to add context to the active run, or --if-running queue to send one follow-up after the current run finishes.", + "description": "Send a prompt into an existing chat session without opening or focusing it. Idle sends are fire-and-forget and visibly add a user message marked as sent by Berd from another session. Running sessions are refused by default; use --if-running steer to add context to the active run, or --if-running queue to send one follow-up after the current run finishes. Use --from to give the sending session or tool a concise visible label in the transcript. Use --delivery-id when retries must create at most one user turn.", "fields": [ { "name": "session_id", @@ -117,6 +132,22 @@ "kind": "string", "description": "What to do if the target session is running: refuse, steer, or queue.", "values": ["refuse", "steer", "queue"] + }, + { + "name": "from", + "required": false, + "kind": "string", + "description": "Optional visible sender label for this message (1-120 chars).", + "min": 1, + "max": 120 + }, + { + "name": "delivery_id", + "required": false, + "kind": "string", + "description": "Optional idempotency id; a repeated id for this session is accepted without creating another user turn (1-200 chars).", + "min": 1, + "max": 200 } ], "schema": { @@ -144,6 +175,20 @@ "description": "What to do if the target session is running: refuse, steer, or queue.", "type": "string", "enum": ["refuse", "steer", "queue"] + }, + "from": { + "description": "Optional visible sender label for this message (1-120 chars).", + "type": "string", + "minLength": 1, + "maxLength": 120, + "pattern": "^[^\\r\\n]*$" + }, + "delivery_id": { + "description": "Optional idempotency id; a repeated id for this session is accepted without creating another user turn (1-200 chars).", + "type": "string", + "minLength": 1, + "maxLength": 200, + "pattern": "^[^\\r\\n]*$" } }, "required": ["session_id", "prompt"], diff --git a/src-tauri/crates/berdctl/cli-surface-feedback.json b/src-tauri/crates/berdctl/cli-surface-feedback.json index 1bd8962b5..1f199f737 100644 --- a/src-tauri/crates/berdctl/cli-surface-feedback.json +++ b/src-tauri/crates/berdctl/cli-surface-feedback.json @@ -8,12 +8,12 @@ "create": { "action": "create", "about": "Create a new chat session and send a prompt in it (fire-and-forget)", - "afterHelp": "Examples:\n berdctl session create --prompt \"Triage the failing nightly build\" \\\n --harness-id claude-acp --json\n berdctl session create --prompt \"Implement the fix\" \\\n --project-id --startup-name my-feature\n\nResult:\n {\"session_id\": \"...\", \"title\": \"...\", \"harness_id\": \"...\",\n \"send_status\": \"dispatched\"}\n The session runs in the background; the user's view does not change. Check\n progress later with `berdctl session get --session-id `." + "afterHelp": "Examples:\n berdctl session create --prompt \"Triage the failing nightly build\" \\\n --harness-id claude-acp --from \"the release orchestrator\" --json\n berdctl session create --prompt \"Implement the fix\" \\\n --project-id --startup-name my-feature\n\nResult:\n {\"session_id\": \"...\", \"title\": \"...\", \"harness_id\": \"...\",\n \"send_status\": \"dispatched\"}\n The session runs in the background; the user's view does not change. Check\n progress later with `berdctl session get --session-id `." }, "send": { "action": "send", "about": "Send a prompt into an existing chat session", - "afterHelp": "Example:\n berdctl session send --session-id \\\n --prompt \"Check the latest CI failure\" --if-running queue --json\n\nResult:\n {\"session_id\": \"...\", \"send_status\": \"dispatched\"|\"steered\"|\"queued\"}\n The user's current view does not change." + "afterHelp": "Example:\n berdctl session send --session-id \\\n --prompt \"Check the latest CI failure\" --if-running queue \\\n --from \"the Berd session handling CI\" --delivery-id --json\n\nResult:\n {\"session_id\": \"...\", \"send_status\": \"dispatched\"|\"steered\"|\"queued\"|\"deduplicated\"}\n The user's current view does not change." }, "open": { "action": "open", diff --git a/src-tauri/crates/berdctl/cli-surface.json b/src-tauri/crates/berdctl/cli-surface.json index c15c55bbe..a096970e9 100644 --- a/src-tauri/crates/berdctl/cli-surface.json +++ b/src-tauri/crates/berdctl/cli-surface.json @@ -8,12 +8,12 @@ "create": { "action": "create", "about": "Create a new chat session and send a prompt in it (fire-and-forget)", - "afterHelp": "Examples:\n berdctl session create --prompt \"Triage the failing nightly build\" \\\n --harness-id claude-acp --json\n berdctl session create --prompt \"Implement the fix\" \\\n --project-id --startup-name my-feature\n\nResult:\n {\"session_id\": \"...\", \"title\": \"...\", \"harness_id\": \"...\",\n \"send_status\": \"dispatched\"}\n The session runs in the background; the user's view does not change. Check\n progress later with `berdctl session get --session-id `." + "afterHelp": "Examples:\n berdctl session create --prompt \"Triage the failing nightly build\" \\\n --harness-id claude-acp --from \"the release orchestrator\" --json\n berdctl session create --prompt \"Implement the fix\" \\\n --project-id --startup-name my-feature\n\nResult:\n {\"session_id\": \"...\", \"title\": \"...\", \"harness_id\": \"...\",\n \"send_status\": \"dispatched\"}\n The session runs in the background; the user's view does not change. Check\n progress later with `berdctl session get --session-id `." }, "send": { "action": "send", "about": "Send a prompt into an existing chat session", - "afterHelp": "Example:\n berdctl session send --session-id \\\n --prompt \"Check the latest CI failure\" --if-running queue --json\n\nResult:\n {\"session_id\": \"...\", \"send_status\": \"dispatched\"|\"steered\"|\"queued\"}\n The user's current view does not change." + "afterHelp": "Example:\n berdctl session send --session-id \\\n --prompt \"Check the latest CI failure\" --if-running queue \\\n --from \"the Berd session handling CI\" --delivery-id --json\n\nResult:\n {\"session_id\": \"...\", \"send_status\": \"dispatched\"|\"steered\"|\"queued\"|\"deduplicated\"}\n The user's current view does not change." }, "open": { "action": "open", diff --git a/src-tauri/crates/berdctl/src/main.rs b/src-tauri/crates/berdctl/src/main.rs index 5cc943846..da8e9a981 100644 --- a/src-tauri/crates/berdctl/src/main.rs +++ b/src-tauri/crates/berdctl/src/main.rs @@ -622,7 +622,9 @@ mod tests { const EXPECTED_SESSION_CREATE_HELP: &str = r#"Create a new chat session on any installed agent harness and send the prompt in it. Fire-and-forget: returns the session id immediately and the session runs in the background without changing what the user sees; the user can open it -themselves. Only check on it later (action "get") if the user asks. +themselves. Use --from to give the delegating session or tool a concise visible +label on the initial message. Only check on it later (action "get") if the user +asks. Usage: berdctl session create [OPTIONS] --prompt @@ -647,6 +649,9 @@ Options: Branch/worktree name when the project's startup mode is branch or worktree; required for those modes. + --from + Optional visible sender label for the initial message (1-120 chars). + --json Print the raw JSON result on a single line (default: pretty-printed JSON) @@ -660,7 +665,7 @@ Options: Examples: berdctl session create --prompt "Triage the failing nightly build" \ - --harness-id claude-acp --json + --harness-id claude-acp --from "the release orchestrator" --json berdctl session create --prompt "Implement the fix" \ --project-id --startup-name my-feature diff --git a/src-tauri/src/services/acp/goose_serve.rs b/src-tauri/src/services/acp/goose_serve.rs index fb34f649a..0905b457b 100644 --- a/src-tauri/src/services/acp/goose_serve.rs +++ b/src-tauri/src/services/acp/goose_serve.rs @@ -830,6 +830,7 @@ fn resolve_berdctl_spawn_paths( prepend_dirs: &mut Vec, ) -> BerdctlSpawnPaths { let berdctl_bin = resolve_berdctl_bin(); + let berd_monitor_bin = resolve_berd_monitor_bin(); let app_data_dir = match app_handle.path().app_data_dir() { Ok(app_data_dir) => Some(app_data_dir), Err(error) => { @@ -839,14 +840,25 @@ fn resolve_berdctl_spawn_paths( None } }; - if let (Some(cli_path), Some(app_data_dir)) = (berdctl_bin.as_deref(), app_data_dir.as_deref()) - { + if let Some(app_data_dir) = app_data_dir.as_deref() { let shim_dir = app_data_dir.join("bin"); - match create_berdctl_shim(&shim_dir, cli_path) { - // After the distro bin dir so that dir keeps its - // pinned PATH-front position. - Ok(()) => prepend_dirs.push(shim_dir), - Err(error) => log::warn!("Skipping berdctl PATH shim: {error}"), + let mut installed_any = false; + if let Some(cli_path) = berdctl_bin.as_deref() { + match create_berdctl_shim(&shim_dir, cli_path) { + Ok(()) => installed_any = true, + Err(error) => log::warn!("Skipping berdctl PATH shim: {error}"), + } + } + if let Some(cli_path) = berd_monitor_bin.as_deref() { + match create_berd_monitor_shim(&shim_dir, cli_path) { + Ok(()) => installed_any = true, + Err(error) => log::warn!("Skipping berd-monitor PATH shim: {error}"), + } + } + if installed_any { + // After the distro bin dir so that dir keeps its pinned + // PATH-front position. + prepend_dirs.push(shim_dir); } } BerdctlSpawnPaths { @@ -890,9 +902,19 @@ fn apply_berdctl_env( /// replacing an in-use copied `.exe` can fail. #[cfg(feature = "berdctl")] fn create_berdctl_shim(shim_dir: &Path, cli_path: &Path) -> Result<(), String> { + create_cli_shim(shim_dir, cli_path, berdctl_shim_name()) +} + +#[cfg(feature = "berdctl")] +fn create_berd_monitor_shim(shim_dir: &Path, cli_path: &Path) -> Result<(), String> { + create_cli_shim(shim_dir, cli_path, berd_monitor_shim_name()) +} + +#[cfg(feature = "berdctl")] +fn create_cli_shim(shim_dir: &Path, cli_path: &Path, shim_name: &str) -> Result<(), String> { if !cli_path.exists() { return Err(format!( - "berdctl binary not found at {}", + "agent CLI binary not found at {}", cli_path.display() )); } @@ -900,7 +922,7 @@ fn create_berdctl_shim(shim_dir: &Path, cli_path: &Path) -> Result<(), String> { std::fs::create_dir_all(shim_dir) .map_err(|error| format!("failed to create {}: {error}", shim_dir.display()))?; - let link = shim_dir.join(berdctl_shim_name()); + let link = shim_dir.join(shim_name); // `remove_file` deletes a symlink itself rather than its target. match std::fs::remove_file(&link) { Ok(()) => {} @@ -912,7 +934,7 @@ fn create_berdctl_shim(shim_dir: &Path, cli_path: &Path) -> Result<(), String> { )); } } - create_berdctl_shim_file(cli_path, &link) + create_cli_shim_file(cli_path, &link) } /// Mirrors `get_goose_command`: explicit env override (exported by `just @@ -930,6 +952,18 @@ fn resolve_berdctl_bin() -> Option { Some(exe.parent()?.join(berdctl_binary_name())) } +#[cfg(feature = "berdctl")] +fn resolve_berd_monitor_bin() -> Option { + if let Ok(override_path) = std::env::var("BERD_MONITOR_BIN") { + if !override_path.is_empty() { + return Some(PathBuf::from(override_path)); + } + } + + let exe = std::env::current_exe().ok()?; + Some(exe.parent()?.join(berd_monitor_binary_name())) +} + #[cfg(feature = "berdctl")] fn berdctl_binary_name() -> &'static str { if cfg!(windows) { @@ -939,6 +973,15 @@ fn berdctl_binary_name() -> &'static str { } } +#[cfg(feature = "berdctl")] +fn berd_monitor_binary_name() -> &'static str { + if cfg!(windows) { + "berd-monitor.exe" + } else { + "berd-monitor" + } +} + #[cfg(feature = "berdctl")] fn berdctl_shim_name() -> &'static str { if cfg!(windows) { @@ -948,8 +991,17 @@ fn berdctl_shim_name() -> &'static str { } } +#[cfg(feature = "berdctl")] +fn berd_monitor_shim_name() -> &'static str { + if cfg!(windows) { + "berd-monitor.cmd" + } else { + berd_monitor_binary_name() + } +} + #[cfg(all(feature = "berdctl", unix))] -fn create_berdctl_shim_file(cli_path: &Path, link: &Path) -> Result<(), String> { +fn create_cli_shim_file(cli_path: &Path, link: &Path) -> Result<(), String> { std::os::unix::fs::symlink(cli_path, link).map_err(|error| { format!( "failed to symlink {} -> {}: {error}", @@ -960,7 +1012,7 @@ fn create_berdctl_shim_file(cli_path: &Path, link: &Path) -> Result<(), String> } #[cfg(all(feature = "berdctl", windows))] -fn create_berdctl_shim_file(cli_path: &Path, link: &Path) -> Result<(), String> { +fn create_cli_shim_file(cli_path: &Path, link: &Path) -> Result<(), String> { let content = format!("@echo off\r\n\"{}\" %*\r\n", cli_path.to_string_lossy()); std::fs::write(link, content).map_err(|error| { format!( diff --git a/src-tauri/tauri.conf.json b/src-tauri/tauri.conf.json index 335c6ecf7..5ddf6940a 100644 --- a/src-tauri/tauri.conf.json +++ b/src-tauri/tauri.conf.json @@ -83,7 +83,12 @@ "../resources/berd-sounds-5.mp3": "berd-sounds-5.mp3", "../resources/berd-sounds-6.mp3": "berd-sounds-6.mp3" }, - "externalBin": ["binaries/goosed", "binaries/berdctl", "binaries/catch"], + "externalBin": [ + "binaries/goosed", + "binaries/berdctl", + "binaries/berd-monitor", + "binaries/catch" + ], "linux": { "deb": { "depends": ["libvulkan1"] diff --git a/src-tauri/tauri.windows.conf.json b/src-tauri/tauri.windows.conf.json index 16f16c208..ae47a37b4 100644 --- a/src-tauri/tauri.windows.conf.json +++ b/src-tauri/tauri.windows.conf.json @@ -17,7 +17,11 @@ }, "bundle": { "targets": ["nsis"], - "externalBin": ["binaries/goosed", "binaries/berdctl"], + "externalBin": [ + "binaries/goosed", + "binaries/berdctl", + "binaries/berd-monitor" + ], "windows": { "webviewInstallMode": { "type": "downloadBootstrapper", diff --git a/src/features/berdctl/__tests__/bridge/useBerdctlQueuedMessageDrain.test.tsx b/src/features/berdctl/__tests__/bridge/useBerdctlQueuedMessageDrain.test.tsx index 96ae51ada..520f3c16b 100644 --- a/src/features/berdctl/__tests__/bridge/useBerdctlQueuedMessageDrain.test.tsx +++ b/src/features/berdctl/__tests__/bridge/useBerdctlQueuedMessageDrain.test.tsx @@ -13,6 +13,7 @@ import { useSessionWindowStore } from "@/features/chat/stores/sessionWindowStore import { useChatSessionStore } from "@/features/chat/stores/chatSessionStore"; import * as queuePersistence from "@/features/chat/stores/queuePersistence"; import { useBerdctlQueuedMessageDrain } from "@/features/berdctl/bridge/useBerdctlQueuedMessageDrain"; +import { createUserMessage } from "@/shared/types/messages"; const mocks = vi.hoisted(() => ({ sendPromptToExistingSessionInBackground: vi.fn(), @@ -558,6 +559,123 @@ describe("useBerdctlQueuedMessageDrain", () => { ).toBeUndefined(); }); + it("preserves a queued delivery id without a sender label", async () => { + const sendOptions = { + userMessageMetadata: { + origin: "berdctl_cross_session" as const, + berdDeliveryId: "monitor-event-1", + }, + acpGooseMetadata: { + origin: "berdctl_cross_session" as const, + berdDeliveryId: "monitor-event-1", + }, + }; + const chatStore = useChatStore.getState(); + chatStore.setChatState("session-1", "streaming"); + chatStore.enqueueTransportReadyMessage("session-1", { + persona: { kind: "inherit" }, + text: "queued delivery", + sendOptions, + }); + render(); + + act(() => { + useChatStore.getState().setChatState("session-1", "idle"); + }); + + await waitFor(() => { + expect( + mocks.sendPromptToExistingSessionInBackground, + ).toHaveBeenCalledWith( + "session-1", + "queued delivery", + expect.any(Function), + { + returnOnDispatch: true, + sendOptions, + validateHydratedTranscript: expect.any(Function), + }, + ); + }); + }); + + it("dismisses a stale queued delivery already accepted in the transcript", async () => { + const accepted = createUserMessage("queued delivery"); + accepted.metadata = { + origin: "berdctl_cross_session", + berdDeliveryId: "monitor-event-1", + }; + const chatStore = useChatStore.getState(); + chatStore.addMessage("session-1", accepted); + chatStore.enqueueTransportReadyMessage("session-1", { + persona: { kind: "inherit" }, + text: "queued delivery", + sendOptions: { + userMessageMetadata: { + origin: "berdctl_cross_session" as const, + berdDeliveryId: "monitor-event-1", + }, + }, + }); + + render(); + + await waitFor(() => { + expect( + useChatStore.getState().queuedMessageBySession["session-1"], + ).toBeUndefined(); + }); + expect( + mocks.sendPromptToExistingSessionInBackground, + ).not.toHaveBeenCalled(); + expect(useChatStore.getState().messagesBySession["session-1"]).toHaveLength( + 1, + ); + }); + + it("dismisses a stale queued delivery accepted during cold hydration", async () => { + mocks.sendPromptToExistingSessionInBackground.mockImplementationOnce( + async ( + _sessionId: string, + _prompt: string, + _beforeUserMessageCommitted: () => void, + options?: { validateHydratedTranscript?: () => void }, + ) => { + const accepted = createUserMessage("queued delivery"); + accepted.metadata = { + origin: "berdctl_cross_session", + berdDeliveryId: "monitor-event-1", + }; + useChatStore.getState().addMessage("session-1", accepted); + options?.validateHydratedTranscript?.(); + }, + ); + useChatStore.getState().enqueueTransportReadyMessage("session-1", { + persona: { kind: "inherit" }, + text: "queued delivery", + sendOptions: { + userMessageMetadata: { + origin: "berdctl_cross_session" as const, + berdDeliveryId: "monitor-event-1", + }, + }, + }); + + render(); + + await waitFor(() => { + expect( + useChatStore.getState().queuedMessageBySession["session-1"], + ).toBeUndefined(); + }); + expect( + mocks.sendPromptToExistingSessionInBackground, + ).toHaveBeenCalledOnce(); + expect(useChatStore.getState().messagesBySession["session-1"]).toHaveLength( + 1, + ); + }); + it("drains consecutive berdctl records in FIFO order while idle", async () => { const chatStore = useChatStore.getState(); chatStore.enqueueTransportReadyMessage("session-1", { diff --git a/src/features/berdctl/__tests__/commands/commands.test.ts b/src/features/berdctl/__tests__/commands/commands.test.ts index 6f6333232..8d1246d90 100644 --- a/src/features/berdctl/__tests__/commands/commands.test.ts +++ b/src/features/berdctl/__tests__/commands/commands.test.ts @@ -707,6 +707,7 @@ describe("sessions.create", () => { prompt: "what is 1+1", agent_id: "agent-7", model_id: "model-9", + from: "the test orchestrator", }, ctx, ); @@ -733,7 +734,19 @@ describe("sessions.create", () => { useChatStore.getState().queuedMessageBySession["session-new"]; expect(queued?.[0]).toMatchObject({ kind: "transport-ready", - payload: { text: "what is 1+1" }, + payload: { + text: "what is 1+1", + sendOptions: { + userMessageMetadata: { + origin: "berdctl_cross_session", + berdSenderLabel: "the test orchestrator", + }, + acpGooseMetadata: { + origin: "berdctl_cross_session", + berdSenderLabel: "the test orchestrator", + }, + }, + }, }); expect(controller.openSession).not.toHaveBeenCalled(); }); @@ -1683,6 +1696,212 @@ describe("sessions.send", () => { ), ).toEqual(["next prompt", "another prompt"]); }); + + it("preserves a visible sender label on queued prompts", async () => { + mockSessionFound(); + useChatStore.getState().setChatState("session-1", "streaming"); + + const result = await dispatchCommand( + "sessions", + { + action: "send", + session_id: "session-1", + prompt: "[monitor: checks] complete", + if_running: "queue", + from: "the Berd session handling berd-monitor implementation", + delivery_id: "monitor-event-1", + }, + ctx, + ); + + expect(result).toEqual({ session_id: "session-1", send_status: "queued" }); + expect( + useChatStore.getState().queuedMessageBySession["session-1"]?.[0]?.payload + .sendOptions, + ).toEqual({ + userMessageMetadata: { + origin: "berdctl_cross_session", + berdSenderLabel: + "the Berd session handling berd-monitor implementation", + berdDeliveryId: "monitor-event-1", + }, + acpGooseMetadata: { + origin: "berdctl_cross_session", + berdSenderLabel: + "the Berd session handling berd-monitor implementation", + berdDeliveryId: "monitor-event-1", + }, + }); + }); + + it("accepts a repeated delivery id without queueing another user turn", async () => { + mockSessionFound(); + useChatStore.getState().setChatState("session-1", "streaming"); + + const first = await dispatchCommand( + "sessions", + { + action: "send", + session_id: "session-1", + prompt: "monitor event", + if_running: "queue", + delivery_id: "monitor-event-1", + }, + ctx, + ); + const duplicate = await dispatchCommand( + "sessions", + { + action: "send", + session_id: "session-1", + prompt: "monitor event retried", + if_running: "queue", + delivery_id: "monitor-event-1", + }, + ctx, + ); + + expect(first).toEqual({ session_id: "session-1", send_status: "queued" }); + expect(duplicate).toEqual({ + session_id: "session-1", + send_status: "deduplicated", + }); + expect( + useChatStore.getState().queuedMessageBySession["session-1"], + ).toHaveLength(1); + }); + + it("serializes concurrent admission of the same delivery id", async () => { + const session = makeAcpSession({ + sessionId: "session-1", + providerId: "codex-acp", + }); + let releaseLoads: (() => void) | undefined; + const loadsReleased = new Promise((resolve) => { + releaseLoads = resolve; + }); + let loadCount = 0; + mocks.acpGetSessionInfo.mockImplementation(async () => { + loadCount += 1; + await loadsReleased; + return session; + }); + + const send = (prompt: string) => + dispatchCommand( + "sessions", + { + action: "send", + session_id: "session-1", + prompt, + if_running: "queue", + delivery_id: "monitor-event-1", + }, + ctx, + ); + const first = send("monitor event"); + const retry = send("monitor event retried concurrently"); + await vi.waitFor(() => expect(loadCount).toBe(2)); + releaseLoads?.(); + + await expect(Promise.all([first, retry])).resolves.toEqual([ + { session_id: "session-1", send_status: "dispatched" }, + { session_id: "session-1", send_status: "deduplicated" }, + ]); + expect( + (useChatStore.getState().messagesBySession["session-1"] ?? []).length + + (useChatStore.getState().queuedMessageBySession["session-1"]?.length ?? + 0), + ).toBe(1); + }); + + it("deduplicates a delivery id restored in the transcript before target guards", async () => { + mockSessionFound(); + const accepted = createUserMessage("monitor event"); + accepted.metadata = { + origin: "berdctl_cross_session", + berdDeliveryId: "monitor-event-1", + }; + useChatStore.getState().addMessage("session-1", accepted); + useChatStore.getState().setChatState("session-1", "streaming"); + useSessionWindowStore + .getState() + .setSnapshot([{ sessionId: "session-1", windowLabel: "session" }]); + + const duplicate = await dispatchCommand( + "sessions", + { + action: "send", + session_id: "session-1", + prompt: "monitor event retried after restart", + delivery_id: "monitor-event-1", + }, + ctx, + ); + + expect(duplicate).toEqual({ + session_id: "session-1", + send_status: "deduplicated", + }); + expect(mocks.acpSendMessage).not.toHaveBeenCalled(); + expect(mocks.acpSteerMessage).not.toHaveBeenCalled(); + }); + + it("deduplicates a delivery id restored while hydrating a cold transcript", async () => { + mockSessionFound({ providerId: "codex-acp" }); + mocks.loadSessionMessages.mockImplementationOnce(async () => { + const accepted = createUserMessage("monitor event"); + accepted.metadata = { + origin: "berdctl_cross_session", + berdDeliveryId: "monitor-event-1", + }; + useChatStore.getState().addMessage("session-1", accepted); + return true; + }); + + const duplicate = await dispatchCommand( + "sessions", + { + action: "send", + session_id: "session-1", + prompt: "monitor event retried after restart", + delivery_id: "monitor-event-1", + }, + ctx, + ); + + expect(duplicate).toEqual({ + session_id: "session-1", + send_status: "deduplicated", + }); + expect(useChatStore.getState().messagesBySession["session-1"]).toHaveLength( + 1, + ); + expect( + useChatStore.getState().queuedMessageBySession["session-1"], + ).toBeUndefined(); + expect(mocks.acpSendMessage).not.toHaveBeenCalled(); + expect(mocks.acpSteerMessage).not.toHaveBeenCalled(); + }); + + it("rejects multiline sender labels before dispatch", async () => { + const error = await expectCommandError( + dispatchCommand( + "sessions", + { + action: "send", + session_id: "session-1", + prompt: "monitor update", + if_running: "queue", + from: "first line\nsecond line", + }, + ctx, + ), + "invalid_args", + ); + + expect(error.message).toContain("single line"); + }); }); describe("sessions.open", () => { diff --git a/src/features/berdctl/bridge/useBerdctlQueuedMessageDrain.ts b/src/features/berdctl/bridge/useBerdctlQueuedMessageDrain.ts index 739e5dffc..468a54b16 100644 --- a/src/features/berdctl/bridge/useBerdctlQueuedMessageDrain.ts +++ b/src/features/berdctl/bridge/useBerdctlQueuedMessageDrain.ts @@ -17,6 +17,8 @@ import { import { useChatSessionStore } from "@/features/chat/stores/chatSessionStore"; import { useSessionWindowStore } from "@/features/chat/stores/sessionWindowStore"; import { + BerdctlDeliveryAlreadyAcceptedError, + hasAcceptedBerdctlDeliveryInTranscript, isBerdctlCrossSessionQueuedMessage, sendPromptToExistingSessionInBackground, } from "@/features/berdctl/commands/runtime/sessionSend"; @@ -98,6 +100,17 @@ function drainQueuedMessage(queuedSessionId: string, ownerId: string): void { return; } + const deliveryId = + queuedMessage.payload.sendOptions?.userMessageMetadata?.berdDeliveryId; + if ( + deliveryId && + hasAcceptedBerdctlDeliveryInTranscript(queuedSessionId, deliveryId) + ) { + dismissQueuedMessageIfCurrent(queuedSessionId, queuedMessage); + queueMicrotask(() => drainQueuedMessage(queuedSessionId, ownerId)); + return; + } + drainingSessionIds.add(queuedSessionId); const send = sendPromptToExistingSessionInBackground( queuedSessionId, @@ -110,7 +123,28 @@ function drainQueuedMessage(queuedSessionId: string, ownerId: string): void { ); assertQueuedSessionReady(state.getSessionRuntime(queuedSessionId)); }, - { returnOnDispatch: true }, + { + returnOnDispatch: true, + ...(queuedMessage.payload.sendOptions?.userMessageMetadata + ?.berdSenderLabel || + queuedMessage.payload.sendOptions?.userMessageMetadata?.berdDeliveryId + ? { sendOptions: queuedMessage.payload.sendOptions } + : {}), + ...(deliveryId + ? { + validateHydratedTranscript: () => { + if ( + hasAcceptedBerdctlDeliveryInTranscript( + queuedSessionId, + deliveryId, + ) + ) { + throw new BerdctlDeliveryAlreadyAcceptedError(); + } + }, + } + : {}), + }, ); let sendSucceeded = false; let shouldResumeDrain = false; @@ -118,19 +152,14 @@ function drainQueuedMessage(queuedSessionId: string, ownerId: string): void { void send .then(() => { sendSucceeded = true; - const latestQueuedMessage = - useChatStore.getState().queuedMessageBySession[queuedSessionId]?.[0]; - if ( - latestQueuedMessage?.recordId === queuedMessage.recordId && - latestQueuedMessage.payload === queuedMessage.payload && - !latestQueuedMessage.editing - ) { - useChatStore - .getState() - .dismissQueuedMessage(queuedSessionId, queuedMessage.recordId); - } + dismissQueuedMessageIfCurrent(queuedSessionId, queuedMessage); }) .catch((error) => { + if (error instanceof BerdctlDeliveryAlreadyAcceptedError) { + dismissQueuedMessageIfCurrent(queuedSessionId, queuedMessage); + shouldResumeDrain = true; + return; + } if (error instanceof SessionDispatchContentionError) { waitingForContention = true; const waiter: ContentionWaiter = { @@ -174,6 +203,25 @@ function drainQueuedMessage(queuedSessionId: string, ownerId: string): void { }); } +function dismissQueuedMessageIfCurrent( + sessionId: string, + queuedMessage: QueuedMessageRecord, +): boolean { + const latestQueuedMessage = + useChatStore.getState().queuedMessageBySession[sessionId]?.[0]; + if ( + latestQueuedMessage?.recordId !== queuedMessage.recordId || + latestQueuedMessage.payload !== queuedMessage.payload || + latestQueuedMessage.editing + ) { + return false; + } + useChatStore + .getState() + .dismissQueuedMessage(sessionId, queuedMessage.recordId); + return true; +} + function getQueuedSessionIds( queuedMessageBySession: Record, queuedSessionId?: string, diff --git a/src/features/berdctl/commands/impl/createSession.ts b/src/features/berdctl/commands/impl/createSession.ts index ffdee39e9..dfebbf343 100644 --- a/src/features/berdctl/commands/impl/createSession.ts +++ b/src/features/berdctl/commands/impl/createSession.ts @@ -41,6 +41,16 @@ const createSessionSchema = z .describe( "Branch/worktree name when the project's startup mode is branch or worktree; required for those modes.", ), + from: z + .string() + .trim() + .min(1) + .max(120) + .regex(/^[^\r\n]*$/, "Sender label must be a single line.") + .optional() + .describe( + "Optional visible sender label for the initial message (1-120 chars).", + ), }) .strict(); @@ -65,10 +75,12 @@ export const createSessionCommand = defineCommand({ "Create a new chat session on any installed agent harness and send the prompt in it. " + "Fire-and-forget: returns the session id immediately and the session runs in the " + "background without changing what the user sees; the user can open it themselves. " + + "Use --from to give the delegating session or tool a concise visible label on " + + "the initial message. " + 'Only check on it later (action "get") if the user asks.', helpFooter: `Examples: berdctl session create --prompt "Triage the failing nightly build" \\ - --harness-id claude-acp --json + --harness-id claude-acp --from "the release orchestrator" --json berdctl session create --prompt "Implement the fix" \\ --project-id --startup-name my-feature @@ -211,7 +223,9 @@ Result: persona: persona ? { kind: "persona", id: persona.id, name: persona.displayName } : { kind: "inherit" }, - sendOptions: berdctlCrossSessionSendOptions(), + sendOptions: berdctlCrossSessionSendOptions({ + senderLabel: args.from, + }), }), { project, queueReady: true }, ); diff --git a/src/features/berdctl/commands/impl/sendSession.ts b/src/features/berdctl/commands/impl/sendSession.ts index e48755cea..998bac0e0 100644 --- a/src/features/berdctl/commands/impl/sendSession.ts +++ b/src/features/berdctl/commands/impl/sendSession.ts @@ -39,12 +39,32 @@ const sendSessionSchema = z .describe( "What to do if the target session is running: refuse, steer, or queue.", ), + from: z + .string() + .trim() + .min(1) + .max(120) + .regex(/^[^\r\n]*$/, "Sender label must be a single line.") + .optional() + .describe( + "Optional visible sender label for this message (1-120 chars).", + ), + delivery_id: z + .string() + .trim() + .min(1) + .max(200) + .regex(/^[^\r\n]*$/, "Delivery id must be a single line.") + .optional() + .describe( + "Optional idempotency id; a repeated id for this session is accepted without creating another user turn (1-200 chars).", + ), }) .strict(); interface SendSessionResult { session_id: string; - send_status: "dispatched" | "steered" | "queued"; + send_status: "dispatched" | "steered" | "queued" | "deduplicated"; } function runningTargetMessage(sessionId: string): string { @@ -61,13 +81,16 @@ export const sendSessionCommand = defineCommand({ "Idle sends are fire-and-forget and visibly add a user message marked as sent " + "by Berd from another session. Running sessions are refused by default; use " + "--if-running steer to add context to the active run, or --if-running queue " + - "to send one follow-up after the current run finishes.", + "to send one follow-up after the current run finishes. Use --from to give " + + "the sending session or tool a concise visible label in the transcript. " + + "Use --delivery-id when retries must create at most one user turn.", helpFooter: `Example: berdctl session send --session-id \\ - --prompt "Check the latest CI failure" --if-running queue --json + --prompt "Check the latest CI failure" --if-running queue \\ + --from "the Berd session handling CI" --delivery-id --json Result: - {"session_id": "...", "send_status": "dispatched"|"steered"|"queued"} + {"session_id": "...", "send_status": "dispatched"|"steered"|"queued"|"deduplicated"} The user's current view does not change.`, schema: sendSessionSchema, bridgeTimeoutMs: 60_000, @@ -78,6 +101,9 @@ Result: { findProjectOrThrow }, { berdctlCrossSessionSendOptions, + BerdctlDeliveryAlreadyAcceptedError, + hasAcceptedBerdctlDelivery, + reserveBerdctlDelivery, sendPromptToExistingSessionInBackground, SessionDispatchContentionError, SessionDispatchUnresolvedError, @@ -88,157 +114,188 @@ Result: import("../runtime/projects"), import("../runtime/sessionSend"), ]); + const sendOptions = berdctlCrossSessionSendOptions({ + senderLabel: args.from, + deliveryId: args.delivery_id, + }); await loadSessionForBerdctl(args.session_id); const session = requireSession(args.session_id); - if (useSessionWindowStore.getState().isOpenInWindow(args.session_id)) { - throw new CommandError( - "target_session_running", - `Refusing to send to session "${args.session_id}" while it is open in a separate window; close that window first or ask the user.`, - ); + const releaseDeliveryReservation = args.delivery_id + ? reserveBerdctlDelivery(args.session_id, args.delivery_id) + : undefined; + if (args.delivery_id && !releaseDeliveryReservation) { + return { session_id: session.id, send_status: "deduplicated" }; } - const chatStore = useChatStore.getState(); - const runtime = chatStore.getSessionRuntime(args.session_id); - if ( - args.startup_name && - ((chatStore.queuedMessageBySession[args.session_id]?.length ?? 0) > 0 || - session.messageCount > 0) - ) { - throw new CommandError( - "invalid_args", - "startup_name is only valid for the first send when workspace setup is available.", - ); - } - if (!isQueuedSessionReady(runtime)) { - switch (args.if_running) { - case "refuse": - throw new CommandError( - "target_session_running", - runningTargetMessage(args.session_id), - ); + try { + if (useSessionWindowStore.getState().isOpenInWindow(args.session_id)) { + throw new CommandError( + "target_session_running", + `Refusing to send to session "${args.session_id}" while it is open in a separate window; close that window first or ask the user.`, + ); + } - case "steer": - if (runtime.isRunCancellationPending) { + const chatStore = useChatStore.getState(); + const runtime = chatStore.getSessionRuntime(args.session_id); + if ( + args.startup_name && + ((chatStore.queuedMessageBySession[args.session_id]?.length ?? 0) > 0 || + session.messageCount > 0) + ) { + throw new CommandError( + "invalid_args", + "startup_name is only valid for the first send when workspace setup is available.", + ); + } + if (!isQueuedSessionReady(runtime)) { + switch (args.if_running) { + case "refuse": throw new CommandError( "target_session_running", - `Refusing to steer session "${args.session_id}" while cancellation is pending; use --if-running queue or wait for cancellation to finish.`, + runningTargetMessage(args.session_id), ); - } - await steerPromptInSession( - args.session_id, - args.prompt, - undefined, - berdctlCrossSessionSendOptions(), - { throwOnError: true }, - ); - return { session_id: session.id, send_status: "steered" }; - - case "queue": - chatStore.enqueueTransportReadyMessage( - args.session_id, - admitSystemInheritedQueuedMessage({ - text: args.prompt, - sendOptions: berdctlCrossSessionSendOptions(), - }), - ); - return { session_id: session.id, send_status: "queued" }; - default: - throw new Error( - `Unhandled if_running mode: ${String(args.if_running satisfies never)}`, - ); + case "steer": + if (runtime.isRunCancellationPending) { + throw new CommandError( + "target_session_running", + `Refusing to steer session "${args.session_id}" while cancellation is pending; use --if-running queue or wait for cancellation to finish.`, + ); + } + await steerPromptInSession( + args.session_id, + args.prompt, + undefined, + sendOptions, + { throwOnError: true }, + ); + return { session_id: session.id, send_status: "steered" }; + + case "queue": + chatStore.enqueueTransportReadyMessage( + args.session_id, + admitSystemInheritedQueuedMessage({ + text: args.prompt, + sendOptions, + }), + ); + return { session_id: session.id, send_status: "queued" }; + + default: + throw new Error( + `Unhandled if_running mode: ${String(args.if_running satisfies never)}`, + ); + } } - } - const project = session.projectId - ? await findProjectOrThrow(session.projectId) - : null; - const firstSend = acceptFirstSend( - args.session_id, - createDeferredQueuedMessagePayload({ - text: args.prompt, - persona: { kind: "inherit" }, - sendOptions: berdctlCrossSessionSendOptions(), - }), - { startupName: args.startup_name, project }, - ); - if (firstSend.needsName) { - throw new CommandError( - "workspace_name_required", - "This session needs a workspace startup name before its first send.", - ); - } - if (firstSend.accepted) { - return { session_id: session.id, send_status: "queued" }; - } - if ((chatStore.queuedMessageBySession[args.session_id]?.length ?? 0) > 0) { - chatStore.enqueueTransportReadyMessage( + const project = session.projectId + ? await findProjectOrThrow(session.projectId) + : null; + const firstSend = acceptFirstSend( args.session_id, - admitSystemInheritedQueuedMessage({ + createDeferredQueuedMessagePayload({ text: args.prompt, - sendOptions: berdctlCrossSessionSendOptions(), + persona: { kind: "inherit" }, + sendOptions, }), + { startupName: args.startup_name, project }, ); - return { session_id: session.id, send_status: "queued" }; - } - - try { - await sendPromptToExistingSessionInBackground( - args.session_id, - args.prompt, - () => { - const liveChatStore = useChatStore.getState(); - assertQueuedSessionReady( - liveChatStore.getSessionRuntime(args.session_id), - (liveChatStore.queuedMessageBySession[args.session_id]?.length ?? - 0) === 0, - ); - }, - { returnOnDispatch: true }, - ); - } catch (error) { - if (error instanceof SessionDispatchContentionError) { - if (args.if_running === "queue") { - useChatStore.getState().enqueueTransportReadyMessage( - args.session_id, - admitSystemInheritedQueuedMessage({ - text: args.prompt, - sendOptions: berdctlCrossSessionSendOptions(), - }), - ); - return { session_id: session.id, send_status: "queued" }; - } + if (firstSend.needsName) { throw new CommandError( - "target_session_running", - runningTargetMessage(args.session_id), + "workspace_name_required", + "This session needs a workspace startup name before its first send.", ); } - if (error instanceof SessionDispatchUnresolvedError) { - throw new CommandError("invalid_args", error.message); + if (firstSend.accepted) { + return { session_id: session.id, send_status: "queued" }; } - if (error instanceof PreCommitSendRejectedError) { - if (args.if_running === "queue") { - useChatStore.getState().enqueueTransportReadyMessage( - args.session_id, - admitSystemInheritedQueuedMessage({ - text: args.prompt, - sendOptions: berdctlCrossSessionSendOptions(), - }), - ); - return { session_id: session.id, send_status: "queued" }; - } - throw new CommandError( - "target_session_running", - runningTargetMessage(args.session_id), + if ( + (chatStore.queuedMessageBySession[args.session_id]?.length ?? 0) > 0 + ) { + chatStore.enqueueTransportReadyMessage( + args.session_id, + admitSystemInheritedQueuedMessage({ + text: args.prompt, + sendOptions, + }), ); + return { session_id: session.id, send_status: "queued" }; } - if (error instanceof CommandError) { - throw error; + + try { + await sendPromptToExistingSessionInBackground( + args.session_id, + args.prompt, + () => { + const liveChatStore = useChatStore.getState(); + assertQueuedSessionReady( + liveChatStore.getSessionRuntime(args.session_id), + (liveChatStore.queuedMessageBySession[args.session_id]?.length ?? + 0) === 0, + ); + }, + { + returnOnDispatch: true, + sendOptions, + validateHydratedTranscript: () => { + if ( + args.delivery_id && + hasAcceptedBerdctlDelivery(args.session_id, args.delivery_id) + ) { + throw new BerdctlDeliveryAlreadyAcceptedError(); + } + }, + }, + ); + } catch (error) { + if (error instanceof BerdctlDeliveryAlreadyAcceptedError) { + return { session_id: session.id, send_status: "deduplicated" }; + } + if (error instanceof SessionDispatchContentionError) { + if (args.if_running === "queue") { + useChatStore.getState().enqueueTransportReadyMessage( + args.session_id, + admitSystemInheritedQueuedMessage({ + text: args.prompt, + sendOptions, + }), + ); + return { session_id: session.id, send_status: "queued" }; + } + throw new CommandError( + "target_session_running", + runningTargetMessage(args.session_id), + ); + } + if (error instanceof SessionDispatchUnresolvedError) { + throw new CommandError("invalid_args", error.message); + } + if (error instanceof PreCommitSendRejectedError) { + if (args.if_running === "queue") { + useChatStore.getState().enqueueTransportReadyMessage( + args.session_id, + admitSystemInheritedQueuedMessage({ + text: args.prompt, + sendOptions, + }), + ); + return { session_id: session.id, send_status: "queued" }; + } + throw new CommandError( + "target_session_running", + runningTargetMessage(args.session_id), + ); + } + if (error instanceof CommandError) { + throw error; + } + throw new Error(formatAcpErrorMessage(error)); } - throw new Error(formatAcpErrorMessage(error)); + return { session_id: session.id, send_status: "dispatched" }; + } finally { + releaseDeliveryReservation?.(); } - return { session_id: session.id, send_status: "dispatched" }; }, }); diff --git a/src/features/berdctl/commands/runtime/sessionSend.ts b/src/features/berdctl/commands/runtime/sessionSend.ts index 2121c3a24..e7a34cfb5 100644 --- a/src/features/berdctl/commands/runtime/sessionSend.ts +++ b/src/features/berdctl/commands/runtime/sessionSend.ts @@ -8,6 +8,7 @@ import { SessionDispatchUnresolvedError, } from "@/features/chat/lib/queuedSessionSend"; import { useChatSessionStore } from "@/features/chat/stores/chatSessionStore"; +import { useChatStore } from "@/features/chat/stores/chatStore"; export { sendQueuedPromptToExistingSessionInBackground, SessionDispatchContentionError, @@ -23,22 +24,91 @@ export { isBerdctlCrossSessionQueuedMessage } from "@/features/chat/lib/queuedMe export const BERDCTL_CROSS_SESSION_ORIGIN = "berdctl_cross_session" satisfies NonNullable; -export function berdctlCrossSessionSendOptions(): ChatSendOptions { +const reservedDeliveryIds = new Set(); + +export class BerdctlDeliveryAlreadyAcceptedError extends Error { + constructor() { + super("The Berd delivery was already accepted."); + this.name = "BerdctlDeliveryAlreadyAcceptedError"; + } +} + +export function berdctlCrossSessionSendOptions( + options: { senderLabel?: string; deliveryId?: string } = {}, +): ChatSendOptions { + const senderMetadata = options.senderLabel + ? { berdSenderLabel: options.senderLabel } + : {}; + const deliveryMetadata = options.deliveryId + ? { berdDeliveryId: options.deliveryId } + : {}; return { userMessageMetadata: { origin: BERDCTL_CROSS_SESSION_ORIGIN, + ...senderMetadata, + ...deliveryMetadata, }, acpGooseMetadata: { origin: BERDCTL_CROSS_SESSION_ORIGIN, + ...senderMetadata, + ...deliveryMetadata, }, }; } +export function hasAcceptedBerdctlDelivery( + sessionId: string, + deliveryId: string, +): boolean { + const chatStore = useChatStore.getState(); + return ( + hasAcceptedBerdctlDeliveryInTranscript(sessionId, deliveryId) || + (chatStore.queuedMessageBySession[sessionId] ?? []).some( + (record) => + record.payload.sendOptions?.userMessageMetadata?.berdDeliveryId === + deliveryId, + ) + ); +} + +export function hasAcceptedBerdctlDeliveryInTranscript( + sessionId: string, + deliveryId: string, +): boolean { + return (useChatStore.getState().messagesBySession[sessionId] ?? []).some( + (message) => message.metadata?.berdDeliveryId === deliveryId, + ); +} + +export function reserveBerdctlDelivery( + sessionId: string, + deliveryId: string, +): (() => void) | null { + const key = JSON.stringify([sessionId, deliveryId]); + if ( + reservedDeliveryIds.has(key) || + hasAcceptedBerdctlDelivery(sessionId, deliveryId) + ) { + return null; + } + reservedDeliveryIds.add(key); + let released = false; + return () => { + if (released) return; + released = true; + reservedDeliveryIds.delete(key); + }; +} + export async function sendPromptToExistingSessionInBackground( sessionId: string, prompt: string, beforeUserMessageCommitted?: () => void, - options: { returnOnDispatch?: boolean } = {}, + options: { + returnOnDispatch?: boolean; + sendOptions?: ChatSendOptions; + validateHydratedTranscript?: () => void; + } = {}, ): Promise { const acquisition = await acquireExistingSessionForBackgroundSend(sessionId); if (acquisition.status === "contended") { @@ -71,13 +141,14 @@ export async function sendPromptToExistingSessionInBackground( dispatchToken: targetLease.token, }); const session = useChatSessionStore.getState().getSession(sessionId); + options.validateHydratedTranscript?.(); await sendPromptInBackground( sessionId, prompt, providerId, persona, { - ...berdctlCrossSessionSendOptions(), + ...(options.sendOptions ?? berdctlCrossSessionSendOptions()), systemPrompt: session ? formatIncludedWorkspacesPrompt(session) : undefined, diff --git a/src/features/chat/ui/ChatInput.tsx b/src/features/chat/ui/ChatInput.tsx index dba6c0dc6..b9489bf39 100644 --- a/src/features/chat/ui/ChatInput.tsx +++ b/src/features/chat/ui/ChatInput.tsx @@ -107,7 +107,12 @@ function stripCrossSessionOrigin>( return metadata; } - const { origin: _origin, ...rest } = metadata; + const { + origin: _origin, + berdSenderLabel: _berdSenderLabel, + berdDeliveryId: _berdDeliveryId, + ...rest + } = metadata; return Object.keys(rest).length > 0 ? (rest as T) : undefined; } diff --git a/src/features/chat/ui/MessageBubble.tsx b/src/features/chat/ui/MessageBubble.tsx index 3882ae4ec..788b341db 100644 --- a/src/features/chat/ui/MessageBubble.tsx +++ b/src/features/chat/ui/MessageBubble.tsx @@ -900,6 +900,9 @@ export const MessageBubble = memo(function MessageBubble({ const isSteeredMessage = isUser && message.metadata?.delivery === "steer"; const isBerdctlCrossSessionMessage = isUser && message.metadata?.origin === "berdctl_cross_session"; + const berdSenderLabel = isBerdctlCrossSessionMessage + ? message.metadata?.berdSenderLabel + : undefined; const timestamp = ( - {t("message.berdctlCrossSessionLabel")} + {berdSenderLabel + ? t("message.berdctlCrossSessionNamedLabel", { + sender: berdSenderLabel, + }) + : t("message.berdctlCrossSessionLabel")} ) : null} {isSteeredMessage ? ( diff --git a/src/features/chat/ui/__tests__/ChatInput.test.tsx b/src/features/chat/ui/__tests__/ChatInput.test.tsx index cfab01afd..ee54a378a 100644 --- a/src/features/chat/ui/__tests__/ChatInput.test.tsx +++ b/src/features/chat/ui/__tests__/ChatInput.test.tsx @@ -3455,7 +3455,7 @@ describe("ChatInput", () => { ); }); - it("strips cross-session origin metadata when resending an edited queued message", async () => { + it("strips cross-session delivery metadata when resending an edited queued message", async () => { const onSend = vi.fn(() => true); const user = userEvent.setup(); @@ -3468,10 +3468,14 @@ describe("ChatInput", () => { sendOptions: { acpGooseMetadata: { origin: "berdctl_cross_session", + berdSenderLabel: "berd-monitor", + berdDeliveryId: "event-1", threadId: "thread-1", }, userMessageMetadata: { origin: "berdctl_cross_session", + berdSenderLabel: "berd-monitor", + berdDeliveryId: "event-1", }, }, }); diff --git a/src/features/chat/ui/__tests__/MessageBubble.test.tsx b/src/features/chat/ui/__tests__/MessageBubble.test.tsx index 5e258e613..ab0e28dca 100644 --- a/src/features/chat/ui/__tests__/MessageBubble.test.tsx +++ b/src/features/chat/ui/__tests__/MessageBubble.test.tsx @@ -617,6 +617,21 @@ describe("MessageBubble", () => { ); }); + it("shows a sender descriptor without replacing trusted Berd provenance", () => { + const message = userMessage("[monitor: PR checks] complete"); + message.metadata = { + ...message.metadata, + origin: "berdctl_cross_session", + berdSenderLabel: "Morgan", + }; + + render(); + + expect( + screen.getByText("Sent by Berd from another session · source: Morgan"), + ).toBeInTheDocument(); + }); + it("renders provenance and steer labels together", () => { const message = userMessage("steered from another session"); message.metadata = { diff --git a/src/shared/api/__tests__/acpReplayMetadata.test.ts b/src/shared/api/__tests__/acpReplayMetadata.test.ts index b33da9e2c..d4e576468 100644 --- a/src/shared/api/__tests__/acpReplayMetadata.test.ts +++ b/src/shared/api/__tests__/acpReplayMetadata.test.ts @@ -137,6 +137,24 @@ describe("getReplayUserMetadata", () => { ).toEqual({ origin: "berdctl_cross_session" }); }); + it("restores sender attribution on cross-session messages", () => { + expect( + getReplayUserMetadata({ + _meta: { + goose: { + origin: "berdctl_cross_session", + berdSenderLabel: "berd-monitor", + berdDeliveryId: "monitor-event-1", + }, + }, + }), + ).toEqual({ + origin: "berdctl_cross_session", + berdSenderLabel: "berd-monitor", + berdDeliveryId: "monitor-event-1", + }); + }); + it("restores voice conversation origin metadata", () => { expect( getReplayUserMetadata({ diff --git a/src/shared/api/acpReplayMetadata.ts b/src/shared/api/acpReplayMetadata.ts index 6c26091b7..09eeb1e4d 100644 --- a/src/shared/api/acpReplayMetadata.ts +++ b/src/shared/api/acpReplayMetadata.ts @@ -13,6 +13,8 @@ export type ReplayUserMetadata = Pick< MessageMetadata, | "delivery" | "origin" + | "berdSenderLabel" + | "berdDeliveryId" | "voiceUtteranceId" | "voiceConversationLifecycleId" | "voiceConversationRevision" @@ -75,6 +77,14 @@ export function getReplayUserMetadata( : goose.origin === "voice_conversation" ? "voice_conversation" : undefined; + const berdSenderLabel = + origin === "berdctl_cross_session" + ? boundedSingleLineString(goose.berdSenderLabel, 120) + : undefined; + const berdDeliveryId = + origin === "berdctl_cross_session" + ? boundedSingleLineString(goose.berdDeliveryId, 200) + : undefined; const voiceUtteranceId = origin === "voice_conversation" ? nonEmptyString(goose.voiceUtteranceId) @@ -97,6 +107,8 @@ export function getReplayUserMetadata( return { ...(delivery ? { delivery } : {}), ...(origin ? { origin } : {}), + ...(berdSenderLabel ? { berdSenderLabel } : {}), + ...(berdDeliveryId ? { berdDeliveryId } : {}), ...(voiceUtteranceId ? { voiceUtteranceId } : {}), ...(voiceConversationLifecycleId ? { voiceConversationLifecycleId } : {}), ...(voiceConversationRevision !== undefined @@ -105,6 +117,19 @@ export function getReplayUserMetadata( }; } +function boundedSingleLineString( + value: unknown, + maxLength: number, +): string | undefined { + const normalized = nonEmptyString(value); + return normalized && + normalized.length <= maxLength && + !normalized.includes("\n") && + !normalized.includes("\r") + ? normalized + : undefined; +} + function getGooseReplayMeta( source: ReplayMetadataSource, ): Record | null { diff --git a/src/shared/i18n/locales/en/chat.json b/src/shared/i18n/locales/en/chat.json index aa5775f70..2111b9a26 100644 --- a/src/shared/i18n/locales/en/chat.json +++ b/src/shared/i18n/locales/en/chat.json @@ -406,6 +406,7 @@ "anthropicThinkingHistory": "This chat can't continue with a Claude model because its earlier reasoning history is no longer in a form Claude will accept. Start a new chat, or switch this chat to a non-Claude model to keep going." }, "berdctlCrossSessionLabel": "Sent by Berd from another session", + "berdctlCrossSessionNamedLabel": "Sent by Berd from another session · source: {{sender}}", "steerLabel": "Steered", "viewMore": "View more", "viewLess": "View less", diff --git a/src/shared/i18n/locales/es/chat.json b/src/shared/i18n/locales/es/chat.json index da05f2ef1..23669d44a 100644 --- a/src/shared/i18n/locales/es/chat.json +++ b/src/shared/i18n/locales/es/chat.json @@ -405,6 +405,7 @@ "anthropicThinkingHistory": "Este chat no puede continuar con un modelo Claude porque su historial de razonamiento previo ya no tiene una forma que Claude acepte. Inicia un chat nuevo o cambia este chat a un modelo que no sea Claude para continuar." }, "berdctlCrossSessionLabel": "Enviado por Berd desde otra sesión", + "berdctlCrossSessionNamedLabel": "Enviado por Berd desde otra sesión · fuente: {{sender}}", "steerLabel": "Guiado", "viewMore": "Ver más", "viewLess": "Ver menos", diff --git a/src/shared/types/messages.ts b/src/shared/types/messages.ts index b0aedb915..a8539defc 100644 --- a/src/shared/types/messages.ts +++ b/src/shared/types/messages.ts @@ -255,6 +255,8 @@ export interface MessageMetadata { delivery?: "steering" | "steer"; steeringRequestId?: string; origin?: "berdctl_cross_session" | "voice_conversation"; + berdSenderLabel?: string; + berdDeliveryId?: string; voiceUtteranceId?: string; voiceConversationLifecycleId?: string; voiceConversationRevision?: number;