agentmux_srv\backend\blockcontroller/

shell.rs

// Copyright 2025-2026, AgentMux Corp.
// SPDX-License-Identifier: Apache-2.0

//! ShellController: manages lifecycle of shell and command blocks.
//! Port of Go's pkg/blockcontroller/shellcontroller.go.
//!
//! State machine:
//!   INIT ─(start)─> RUNNING ─(exit/stop)─> DONE
//!   DONE ─(resync+force)─> RUNNING
//!
//! I/O model (3 async tasks when running):
//! 1. PTY read loop: process stdout → FileStore + WPS event
//! 2. Input loop: input channel → process stdin
//! 3. Wait loop: monitor process exit, update status


use std::io::Read as _;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::sync::Mutex;
use std::time::{Instant, SystemTime, UNIX_EPOCH};

#[cfg(unix)]
use libc;

use base64::Engine as _;
use portable_pty::{native_pty_system, CommandBuilder, PtySize};
use tokio::sync::mpsc;

use super::{
    BlockControllerRuntimeStatus, BlockInputUnion, Controller, META_KEY_CMD, META_KEY_CMD_ARGS,
    META_KEY_CMD_CLEAR_ON_START, META_KEY_CMD_CLOSE_ON_EXIT, META_KEY_CMD_CLOSE_ON_EXIT_DELAY,
    META_KEY_CMD_CLOSE_ON_EXIT_FORCE, META_KEY_CMD_ENV, META_KEY_CMD_RUN_ONCE,
    META_KEY_CMD_RUN_ON_START, META_KEY_CONNECTION, STATUS_DONE, STATUS_INIT, STATUS_RUNNING,
};
use crate::backend::eventbus::EventBus;
use crate::backend::shellexec::{ConnInterface, ShellProc};
use crate::backend::storage::filestore::FileStore;
use crate::backend::storage::store::Store;
use crate::backend::obj::{self, MetaMapType, RuntimeOpts};
use crate::backend::wps;

/// Cap on the out-of-order input reorder buffer (`input_seq_buf`).
///
/// The input channel itself is now **unbounded** (`unbounded_channel`): a
/// bounded `try_send` silently dropped input on burst (large pastes arriving
/// faster than the PTY write loop drains — the original truncation bug), and
/// the proper backpressure remedy (`send().await`) can't be applied here
/// because `send_input` is a synchronous trait method holding a `std::Mutex`.
/// An unbounded channel guarantees no input is ever dropped; terminal input is
/// human-paced (and the frontend now chunks large pastes at 4 KB / 5 ms), and
/// the PTY drain loop keeps up, so the queue does not grow in practice. This
/// constant only bounds the reorder buffer (pathological out-of-order seqs).
const SHELL_INPUT_CH_SIZE: usize = 256;

/// Detect the best available interactive shell on Windows.
///
/// Mirrors the original Go logic from pkg/util/shellutil/shellutil.go DetectLocalShellPath():
///   1. Try `pwsh`  (PowerShell 7 — cross-platform)
///   2. Try `powershell` (Windows PowerShell 5.x)
///   3. Fall back to `cmd.exe`
#[cfg(windows)]
fn detect_local_shell_path_windows() -> String {
    use std::os::windows::process::CommandExt;
    use std::process::Command;
    const CREATE_NO_WINDOW: u32 = 0x08000000;
    // Try pwsh (PowerShell 7)
    if Command::new("where")
        .arg("pwsh")
        .creation_flags(CREATE_NO_WINDOW)
        .output()
        .map(|o| o.status.success())
        .unwrap_or(false)
    {
        return "pwsh".to_string();
    }
    // Try powershell (Windows PowerShell 5.x)
    if Command::new("where")
        .arg("powershell")
        .creation_flags(CREATE_NO_WINDOW)
        .output()
        .map(|o| o.status.success())
        .unwrap_or(false)
    {
        return "powershell".to_string();
    }
    "cmd.exe".to_string()
}

/// Stub for non-Windows builds (never called due to cfg!(windows) guard).
#[cfg(not(windows))]
fn detect_local_shell_path_windows() -> String {
    "cmd.exe".to_string()
}

/// PTY read buffer size (matches Go's 4096).
const PTY_READ_BUF_SIZE: usize = 4096;

/// Inner state protected by mutex.
/// Grace period (seconds) between SIGTERM and SIGKILL during stop().
#[allow(dead_code)]
const KILL_GRACE_SECS: u64 = 5;

struct ShellControllerInner {
    /// Current process status.
    proc_status: String,
    /// Process exit code.
    proc_exit_code: i32,
    /// Status version counter (incremented on each change).
    status_version: i32,
    /// Connection name for the shell process.
    conn_name: String,
    /// Input channel sender (sends to the PTY input loop). Unbounded so input
    /// is never dropped on burst — see `SHELL_INPUT_CH_SIZE` doc.
    input_tx: Option<mpsc::UnboundedSender<BlockInputUnion>>,
    /// Input channel receiver (consumed by the PTY input loop).
    #[allow(dead_code)]
    input_rx: Option<mpsc::UnboundedReceiver<BlockInputUnion>>,
    /// OS PID of the running child process, kept for signal delivery in stop().
    child_pid: Option<u32>,
    /// Unix timestamp (ms) when the process was spawned; None until first spawn.
    spawn_ts_ms: Option<i64>,
    /// Monotonic instant of the most recent PTY read; None until first output.
    last_pty_output: Option<Instant>,
    /// True if this pane is running an agent CLI (e.g. claude).
    is_agent_pane: bool,
    /// Next expected input seq number (per-TermViewModel monotonic counter).
    input_seq_next: u64,
    /// Out-of-order input packets waiting for their seq slot (capped at SHELL_INPUT_CH_SIZE).
    input_seq_buf: std::collections::BTreeMap<u64, BlockInputUnion>,
}

/// Factory function type for creating ConnInterface instances.
/// This allows dependency injection for testing.
pub type ConnFactory =
    Box<dyn Fn(&str, &MetaMapType) -> Result<Box<dyn ConnInterface>, String> + Send + Sync>;

/// ShellController manages one shell or command block.
pub struct ShellController {
    /// Controller type: "shell" or "cmd".
    controller_type: String,
    tab_id: String,
    block_id: String,
    /// Prevents concurrent run() calls.
    run_lock: Arc<AtomicBool>,
    /// Protected inner state.
    inner: Arc<Mutex<ShellControllerInner>>,
    /// Optional factory for creating ConnInterface (for testing).
    conn_factory: Mutex<Option<ConnFactory>>,
    /// WPS broker for publishing events (blockfile, controllerstatus).
    broker: Option<Arc<wps::Broker>>,
    /// Event bus (unused for now, reserved for future event routing).
    #[allow(dead_code)]
    event_bus: Option<Arc<EventBus>>,
    /// Wave object store — used to seed cmd:cwd on shell spawn.
    wstore: Option<Arc<Store>>,
}

impl ShellController {
    /// Create a new ShellController.
    pub fn new(
        controller_type: String,
        tab_id: String,
        block_id: String,
        broker: Option<Arc<wps::Broker>>,
        event_bus: Option<Arc<EventBus>>,
        wstore: Option<Arc<Store>>,
    ) -> Self {
        Self {
            controller_type,
            tab_id,
            block_id,
            run_lock: Arc::new(AtomicBool::new(false)),
            inner: Arc::new(Mutex::new(ShellControllerInner {
                proc_status: STATUS_INIT.to_string(),
                proc_exit_code: 0,
                status_version: 0,
                conn_name: String::new(),
                input_tx: None,
                input_rx: None,
                child_pid: None,
                spawn_ts_ms: None,
                last_pty_output: None,
                is_agent_pane: false,
                input_seq_next: 0,
                input_seq_buf: std::collections::BTreeMap::new(),
            })),
            conn_factory: Mutex::new(None),
            broker,
            event_bus,
            wstore,
        }
    }

    /// Set a custom ConnInterface factory (for testing).
    #[allow(dead_code)]
    pub fn set_conn_factory(&self, factory: ConnFactory) {
        *self.conn_factory.lock().unwrap() = Some(factory);
    }

    /// Try to acquire the run lock. Returns false if already running.
    fn try_lock_run(&self) -> bool {
        self.run_lock
            .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
            .is_ok()
    }

    /// Release the run lock.
    fn unlock_run(&self) {
        self.run_lock.store(false, Ordering::SeqCst);
    }

    /// Update process status and increment version (must hold inner lock).
    fn set_status(inner: &mut ShellControllerInner, status: &str) {
        inner.proc_status = status.to_string();
        inner.status_version += 1;
    }

    /// Get the runtime status (snapshot).
    fn get_status_snapshot(&self) -> BlockControllerRuntimeStatus {
        let inner = self.inner.lock().unwrap();
        BlockControllerRuntimeStatus {
            blockid: self.block_id.clone(),
            version: inner.status_version,
            shellprocstatus: inner.proc_status.clone(),
            shellprocconnname: inner.conn_name.clone(),
            shellprocexitcode: inner.proc_exit_code,
            spawn_ts_ms: inner.spawn_ts_ms,
            is_agent_pane: inner.is_agent_pane,
        }
    }

    /// Seconds since last PTY output, or None if no output yet.
    pub fn last_output_secs_ago(&self) -> Option<u64> {
        self.inner.lock().unwrap().last_pty_output.map(|t| t.elapsed().as_secs())
    }

    /// True if this pane is running an agent CLI.
    #[allow(dead_code)]
    pub fn is_agent_pane(&self) -> bool {
        self.inner.lock().unwrap().is_agent_pane
    }

    /// Check block meta for whether to run on start.
    fn should_run_on_start(meta: &MetaMapType) -> bool {
        obj::meta_get_bool(meta, META_KEY_CMD_RUN_ON_START, true)
    }

    /// Check block meta for run-once mode (used in full lifecycle integration).
    #[allow(dead_code)]
    fn should_run_once(meta: &MetaMapType) -> bool {
        obj::meta_get_bool(meta, META_KEY_CMD_RUN_ONCE, false)
    }

    /// Check block meta for clear-on-start (used in full lifecycle integration).
    #[allow(dead_code)]
    fn should_clear_on_start(meta: &MetaMapType) -> bool {
        obj::meta_get_bool(meta, META_KEY_CMD_CLEAR_ON_START, false)
    }

    /// Check block meta for close-on-exit (used in full lifecycle integration).
    #[allow(dead_code)]
    fn should_close_on_exit(meta: &MetaMapType) -> bool {
        obj::meta_get_bool(meta, META_KEY_CMD_CLOSE_ON_EXIT, false)
    }

    /// Check block meta for force close-on-exit (used in full lifecycle integration).
    #[allow(dead_code)]
    fn should_close_on_exit_force(meta: &MetaMapType) -> bool {
        obj::meta_get_bool(meta, META_KEY_CMD_CLOSE_ON_EXIT_FORCE, false)
    }

    /// Get the close-on-exit delay in ms (defaults to 2000, used in full lifecycle integration).
    #[allow(dead_code)]
    fn close_on_exit_delay_ms(meta: &MetaMapType) -> u64 {
        match meta.get(META_KEY_CMD_CLOSE_ON_EXIT_DELAY) {
            Some(serde_json::Value::Number(n)) => n.as_u64().unwrap_or(2000),
            _ => 2000,
        }
    }

    /// Get the connection name from block meta.
    fn get_conn_name(meta: &MetaMapType) -> String {
        obj::meta_get_string(meta, META_KEY_CONNECTION, "local")
    }

    /// Get the command string from block meta.
    fn get_cmd_str(meta: &MetaMapType) -> String {
        obj::meta_get_string(meta, META_KEY_CMD, "")
    }

    /// Get cmd:args array from block meta.
    fn get_cmd_args(meta: &MetaMapType) -> Vec<String> {
        match meta.get(META_KEY_CMD_ARGS) {
            Some(serde_json::Value::Array(arr)) => arr
                .iter()
                .filter_map(|v| v.as_str().map(|s| s.to_string()))
                .collect(),
            _ => vec![],
        }
    }

    /// Check if cmd:interactive is set in block meta.
    fn is_interactive(meta: &MetaMapType) -> bool {
        obj::meta_get_bool(meta, "cmd:interactive", false)
    }

    /// Publish current controller status via the WPS broker.
    fn publish_status(&self) {
        if let Some(ref broker) = self.broker {
            let status = self.get_status_snapshot();
            super::publish_controller_status(broker, &status);
        }
    }

    /// Resolve the initial PTY geometry from the resync `rt_opts` payload.
    ///
    /// The agent pane is a custom UI (not xterm.js), so the PTY never receives
    /// a `fitAddon` resize and must be born at the right width — otherwise the
    /// first batch of agent/tool output wraps at the fallback width until a
    /// post-spawn resize RPC lands, and that RPC races controller startup (it
    /// can fail outright). The frontend computes cols from the pane and passes
    /// them as `rtopts.termsize` on the `controllerresync` command (see
    /// `usePtyWidth.ts` / `launch-flow.ts`).
    ///
    /// Falls back to the historical 25x200 default when `rt_opts` is absent,
    /// unparseable, or carries the serde-default termsize (`rows==0 && cols==0`,
    /// per `obj::is_default_term_size`). Per-field guards let a cols-only
    /// payload keep the default row count. Each axis is clamped to `[1, 1000]`
    /// so the `i64 → u16` cast is lossless and a bogus value cannot open a
    /// zero-size or wrapped-size PTY.
    /// See docs/analysis/AGENT_PANE_PTY_RESIZE_RACE_2026_06_16.md.
    fn pty_size_from_rt_opts(rt_opts: &Option<serde_json::Value>) -> PtySize {
        // Historical fallback geometry. Cols 200 keeps the agent-pane live-log
        // from hard-wrapping at ~80 before the dynamic resize lands.
        const DEFAULT_PTY_ROWS: u16 = 25;
        const DEFAULT_PTY_COLS: u16 = 200;
        let (mut rows, mut cols) = (DEFAULT_PTY_ROWS, DEFAULT_PTY_COLS);
        if let Some(v) = rt_opts {
            if let Ok(rt) = serde_json::from_value::<RuntimeOpts>(v.clone()) {
                let ts = &rt.termsize;
                // rows==0 && cols==0 is the serde default → treat as absent.
                if !(ts.rows == 0 && ts.cols == 0) {
                    if ts.cols > 0 {
                        cols = ts.cols.clamp(1, 1000) as u16;
                    }
                    if ts.rows > 0 {
                        rows = ts.rows.clamp(1, 1000) as u16;
                    }
                }
            }
        }
        PtySize {
            rows,
            cols,
            pixel_width: 0,
            pixel_height: 0,
        }
    }
}

impl Controller for ShellController {
    fn start(
        &self,
        block_meta: MetaMapType,
        rt_opts: Option<serde_json::Value>,
        force: bool,
    ) -> Result<(), String> {
        let cmd_str_preview = Self::get_cmd_str(&block_meta);
        let interactive_preview = Self::is_interactive(&block_meta);
        tracing::info!(
            block_id = %self.block_id,
            controller = %self.controller_type,
            cmd = %cmd_str_preview,
            interactive = interactive_preview,
            force = force,
            "block start requested"
        );

        // Check if we should run
        if !force && !Self::should_run_on_start(&block_meta) {
            tracing::info!(block_id = %self.block_id, "skipping start: run_on_start is false");
            return Ok(());
        }

        // Try to acquire run lock
        if !self.try_lock_run() {
            return Err("controller is already running".to_string());
        }

        // Get connection info
        let conn_name = Self::get_conn_name(&block_meta);

        // Update status to running.
        {
            let mut inner = self.inner.lock().unwrap();
            Self::set_status(&mut inner, STATUS_RUNNING);
            inner.conn_name = conn_name.clone();
        }

        // Create input channel. Unbounded: input must never be silently
        // dropped on burst (the paste-truncation bug). See SHELL_INPUT_CH_SIZE.
        let (input_tx, input_rx) = mpsc::unbounded_channel();
        {
            let mut inner = self.inner.lock().unwrap();
            inner.input_tx = Some(input_tx);
            inner.input_seq_next = 0;
            inner.input_seq_buf.clear();
        }

        // Publish "running" only AFTER `input_tx` exists, so the event is a
        // truthful readiness signal: a frontend that resizes the PTY the instant
        // it sees "running" will not hit `send_input`'s "controller is not
        // running" guard (which fires while `input_tx` is None). The channel is
        // unbounded, so a resize enqueued before the input task spawns is
        // buffered, not lost.
        // See docs/analysis/AGENT_PANE_PTY_RESIZE_RACE_2026_06_16.md.
        self.publish_status();

        // Check if we have a conn_factory (test/mock path)
        let has_factory = self.conn_factory.lock().unwrap().is_some();

        if has_factory {
            // Mock path: use ConnInterface factory (synchronous, for tests)
            let conn_result = {
                let factory = self.conn_factory.lock().unwrap();
                factory.as_ref().unwrap()(&conn_name, &block_meta)
            };

            let mut conn = match conn_result {
                Ok(c) => c,
                Err(e) => {
                    let mut inner = self.inner.lock().unwrap();
                    Self::set_status(&mut inner, STATUS_DONE);
                    inner.proc_exit_code = -1;
                    inner.input_tx = None;
                    self.unlock_run();
                    return Err(format!("failed to create connection: {e}"));
                }
            };

            if let Err(e) = conn.start() {
                let mut inner = self.inner.lock().unwrap();
                Self::set_status(&mut inner, STATUS_DONE);
                inner.proc_exit_code = -1;
                inner.input_tx = None;
                self.unlock_run();
                return Err(format!("failed to start process: {e}"));
            }

            let mut shell_proc = ShellProc::new(conn_name, conn);
            let _done_rx = shell_proc.take_done_rx();
            let exit_code = shell_proc.wait_and_signal();

            {
                let mut inner = self.inner.lock().unwrap();
                inner.proc_exit_code = exit_code;
                Self::set_status(&mut inner, STATUS_DONE);
                inner.input_tx = None;
            }
            self.publish_status();
            self.unlock_run();
            return Ok(());
        }

        // Real PTY path.
        //
        // Open the PTY at the size the frontend computed from the pane (passed
        // as `rtopts.termsize` on the resync), so the agent CLI and its tools
        // wrap correctly from the very first byte. The earlier code always
        // opened at a fixed 200 cols and relied on a post-spawn resize RPC to
        // correct it — but that RPC races controller startup and could fail
        // outright, leaving output wrapped at 200 all session. Seeding the size
        // here removes that race; `pty_size_from_rt_opts` falls back to the
        // historical 25x200 when no size was supplied (e.g. programmatic
        // spawns). See docs/analysis/AGENT_PANE_PTY_RESIZE_RACE_2026_06_16.md.
        let pty_system = native_pty_system();
        let pty_size = Self::pty_size_from_rt_opts(&rt_opts);

        let pair = pty_system.openpty(pty_size).map_err(|e| {
            tracing::error!(block_id = %self.block_id, error = %e, "failed to open PTY");
            let mut inner = self.inner.lock().unwrap();
            Self::set_status(&mut inner, STATUS_DONE);
            inner.proc_exit_code = -1;
            inner.input_tx = None;
            self.unlock_run();
            format!("failed to open PTY: {e}")
        })?;
        tracing::info!(block_id = %self.block_id, rows = pty_size.rows, cols = pty_size.cols, "PTY opened");

        // Determine shell command
        let cmd_str = Self::get_cmd_str(&block_meta);
        let cmd_args = Self::get_cmd_args(&block_meta);
        let interactive = Self::is_interactive(&block_meta);

        // Resolve effective AGENTMUX_AGENT_ID for jekt auto-registration.
        // Priority: block metadata > global settings > WAVEMUX_AGENT_ID env compat.
        let agent_id_for_jekt: Option<String> = block_meta
            .get(META_KEY_CMD_ENV)
            .and_then(|m| m.as_object())
            .and_then(|obj| obj.get("AGENTMUX_AGENT_ID"))
            .and_then(|v| v.as_str())
            .map(|s| s.to_string())
            .or_else(|| {
                let cfg = crate::backend::wconfig::ConfigWatcher::with_config(
                    crate::backend::wconfig::build_default_config(),
                );
                cfg.get_settings().cmd_env.get("AGENTMUX_AGENT_ID").cloned()
            })
            .or_else(|| std::env::var("WAVEMUX_AGENT_ID").ok());

        let mut cmd = if !cmd_str.is_empty() && (!cmd_args.is_empty() || interactive) {
            // Direct spawn: cmd:args provided or cmd:interactive set.
            // Spawn the CLI directly (no sh -c wrapper) so args are passed correctly.
            tracing::info!(block_id = %self.block_id, cmd = %cmd_str, args = ?cmd_args, "direct spawn path");
            let mut c = CommandBuilder::new(&cmd_str);
            if !cmd_args.is_empty() {
                let arg_refs: Vec<&str> = cmd_args.iter().map(|s| s.as_str()).collect();
                c.args(arg_refs);
            }
            c
        } else if !cmd_str.is_empty() {
            // "cmd" controller: run a specific command string via shell wrapper
            tracing::info!(block_id = %self.block_id, cmd = %cmd_str, "shell-wrapped spawn path");
            if cfg!(windows) {
                let mut c = CommandBuilder::new("cmd.exe");
                c.args(["/C", &cmd_str]);
                c
            } else {
                let mut c = CommandBuilder::new("/bin/sh");
                c.args(["-c", &cmd_str]);
                c
            }
        } else {
            // "shell" controller: interactive shell with AgentMux integration
            // On Windows: prefer pwsh (PowerShell 7), fall back to powershell.exe (5.x), then cmd.exe
            let shell_path = if cfg!(windows) {
                detect_local_shell_path_windows()
            } else {
                std::env::var("SHELL").unwrap_or_else(|_| "/bin/bash".to_string())
            };

            let shell_type = crate::backend::shellintegration::detect_shell_type(&shell_path);

            // Deploy shell integration scripts to ~/.agentmux/ (the user's home-based
            // data dir) instead of AGENTMUX_DATA_HOME.  MSIX packages virtualise writes
            // to %LocalAppData%, so files written by the packaged backend aren't visible
            // to child processes (pwsh, bash, etc.) spawned via ConPTY.  The home dir is
            // never virtualised, so the scripts are always reachable at their literal path.
            let shell_home = crate::backend::base::get_home_dir().join(".agentmux");
            crate::backend::shellintegration::deploy_scripts(&shell_home);

            tracing::info!(block_id = %self.block_id, shell = %shell_path, shell_type = ?shell_type, "interactive shell path");

            let mut c = CommandBuilder::new(&shell_path);

            // Apply shell-specific startup args (--rcfile, -File, etc.)
            if let Some(startup) = crate::backend::shellintegration::get_shell_startup(shell_type, &shell_home) {
                for arg in &startup.extra_args {
                    c.arg(arg);
                }
                for (k, v) in &startup.env_vars {
                    c.env(k, v);
                }
            }

            // Inject terminal capability env vars into the PTY environment.
            // ConPTY on Windows fully supports VT/ANSI sequences, so set TERM
            // on all platforms. Without this, CLI tools (e.g. Claude Code) use
            // different Unicode width tables, causing ANSI color offset on Windows.
            c.env("TERM", "xterm-256color");
            c.env("COLORTERM", "truecolor");
            c.env("TERM_PROGRAM", "agentmux");
            c.env("AGENTMUX_BLOCKID", &self.block_id);
            c.env("AGENTMUX_TABID", &self.tab_id);
            c.env("AGENTMUX_VERSION", env!("CARGO_PKG_VERSION"));

            // Inject log directory so agents can find logs without guessing.
            // Always ~/.agentmux/logs/ — matches both host and sidecar.
            let log_dir = dirs::home_dir()
                .unwrap_or_default()
                .join(".agentmux")
                .join("logs");
            c.env("AGENTMUX_LOG_DIR", log_dir.to_string_lossy().as_ref());

            // Propagate local backend URL so agentbus-client prefers local PTY delivery.
            // Set by main.rs after binding; absent in test/mock contexts (graceful no-op).
            if let Ok(local_url) = std::env::var("AGENTMUX_LOCAL_URL") {
                c.env("AGENTMUX_LOCAL_URL", &local_url);
            }

            // AGENTMUX is a plain "1" sentinel — wsh has been retired.
            // Shell integrations check for the presence of AGENTMUX but no
            // longer prepend a path to $PATH based on its value.
            // See specs/SPEC_RETIRE_WSH_2026_04_12.md.
            c.env("AGENTMUX", "1");

            // Wire AgentMux-managed tool dirs into the agent's PATH.
            //
            // Two stores, two precedence rules:
            //
            //   • Bundled store (`<exe_dir>/tools/bin`) ships the app's OWN
            //     version-locked binaries — notably `agentmux-bashwrap`, the
            //     streaming hook that MUST match the running build. It is
            //     PREPENDED so it wins over any stale copy elsewhere on the
            //     system PATH. A stale system-PATH `agentmux-bashwrap` (from a
            //     leftover portable) silently shadowed the fixed one and
            //     reintroduced the exit-130 bug — see
            //     docs/retro/RETRO_BASHWRAP_STALE_BUNDLE_2026_06_13.md. The
            //     bundled jq/rg winning too is intended: agents get the app's
            //     curated, deterministic tool versions regardless of host.
            //
            //   • User-managed store (`~/.agentmux/tools/bin`) holds tools the
            //     user installed via /tools. APPENDED so the user's own system
            //     PATH still wins for those.
            {
                let sep = if cfg!(windows) { ";" } else { ":" };
                let current_path = std::env::var("PATH").unwrap_or_default();
                let mut prepend: Vec<String> = Vec::new();
                let mut append: Vec<String> = Vec::new();

                // Bundled store — prepended (app-owned, version-locked).
                if let Some(bundled_bin) = crate::backend::tool_store::bundled_tools_dir() {
                    if bundled_bin.exists() {
                        // Guardrail: log which agentmux-bashwrap the agent will
                        // actually run. A stale system-PATH copy silently
                        // shadowing the bundled one is exactly the exit-130
                        // trap (RETRO_BASHWRAP_STALE_BUNDLE_2026_06_13.md); this
                        // one line makes "which binary?" answerable at a glance
                        // (cross-check the version with `agentmux-bashwrap
                        // --version`).
                        let bashwrap_exe = if cfg!(windows) {
                            "agentmux-bashwrap.exe"
                        } else {
                            "agentmux-bashwrap"
                        };
                        let bw = bundled_bin.join(bashwrap_exe);
                        if bw.exists() {
                            tracing::info!(
                                target: "agent-tools",
                                path = %bw.display(),
                                "agent bashwrap: bundled (version-locked, prepended to PATH)"
                            );
                        } else {
                            tracing::warn!(
                                target: "agent-tools",
                                dir = %bundled_bin.display(),
                                "agent bashwrap: bundled store present but agentmux-bashwrap MISSING — agent will resolve via system PATH (risk of a stale copy; see RETRO_BASHWRAP_STALE_BUNDLE_2026_06_13.md)"
                            );
                        }
                        prepend.push(bundled_bin.to_string_lossy().into_owned());
                    } else {
                        tracing::warn!(
                            target: "agent-tools",
                            "agent bashwrap: no bundled tools dir — agent will resolve agentmux-bashwrap via system PATH (risk of a stale copy; see RETRO_BASHWRAP_STALE_BUNDLE_2026_06_13.md)"
                        );
                    }
                }

                // User-managed store — appended (system PATH still wins).
                if let Some(user_bin) = crate::backend::tool_store::user_tools_dir() {
                    if user_bin.exists() {
                        append.push(user_bin.to_string_lossy().into_owned());
                    }
                }

                if !prepend.is_empty() || !append.is_empty() {
                    let mut parts = prepend;
                    if !current_path.is_empty() {
                        parts.push(current_path);
                    }
                    parts.extend(append);
                    c.env("PATH", parts.join(sep));
                }
            }

            // Inject cmd:env from wconfig settings and block metadata.
            // Track whether AGENTMUX_AGENT_ID is explicitly set so we know
            // whether to apply the backward-compat WAVEMUX bridge.
            let mut has_agent_id = false;

            // Settings (global defaults, lowest priority)
            let config = crate::backend::wconfig::ConfigWatcher::with_config(
                crate::backend::wconfig::build_default_config(),
            );
            let settings = config.get_settings();
            for (k, v) in &settings.cmd_env {
                if k == "AGENTMUX_AGENT_ID" {
                    has_agent_id = true;
                }
                let expanded = crate::backend::base::expand_home_dir_safe(v);
                c.env(k, expanded.to_string_lossy().as_ref());
            }

            // Block metadata (per-block overrides, highest priority)
            if let Some(env_map) = block_meta.get(META_KEY_CMD_ENV) {
                if let Some(obj) = env_map.as_object() {
                    for (k, v) in obj {
                        if let Some(val) = v.as_str() {
                            if k == "AGENTMUX_AGENT_ID" {
                                has_agent_id = true;
                            }
                            let expanded = crate::backend::base::expand_home_dir_safe(val);
                            c.env(k, expanded.to_string_lossy().as_ref());
                        }
                    }
                }
            }

            // Strip host-inherited agent identity unless explicitly configured
            // in settings.cmd_env or block cmd:env metadata.
            // This also supersedes the old WAVEMUX backward-compat bridge —
            // both AGENTMUX_* and WAVEMUX_* vars are removed so new panes
            // start as plain "Terminal".
            if !has_agent_id {
                c.env_remove("AGENTMUX_AGENT_ID");
                c.env_remove("AGENTMUX_AGENT_COLOR");
                c.env_remove("AGENTMUX_AGENT_TEXT_COLOR");
                c.env_remove("WAVEMUX_AGENT_ID");
                c.env_remove("WAVEMUX_AGENT_COLOR");
            }

            c
        };

        // Set working directory if specified
        let cwd = obj::meta_get_string(&block_meta, super::META_KEY_CMD_CWD, "");
        if !cwd.is_empty() {
            cmd.cwd(&cwd);
        }

        let mut child = pair.slave.spawn_command(cmd).map_err(|e| {
            tracing::error!(block_id = %self.block_id, error = %e, cmd = %cmd_str, "spawn failed");
            let mut inner = self.inner.lock().unwrap();
            Self::set_status(&mut inner, STATUS_DONE);
            inner.proc_exit_code = -1;
            inner.input_tx = None;
            self.unlock_run();
            format!("failed to spawn command: {e}")
        })?;
        tracing::info!(block_id = %self.block_id, "process spawned successfully");

        // Detect agent pane: cmd contains a known agent CLI or has AGENTMUX_AGENT_ID set.
        let is_agent = agent_id_for_jekt.is_some()
            || cmd_str.to_lowercase().contains("claude")
            || cmd_str.to_lowercase().contains("codex")
            || cmd_str.to_lowercase().contains("gemini")
            || cmd_str.to_lowercase().contains("qwen");

        // Register PID and record spawn metadata.
        let spawn_ts_ms = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_millis() as i64)
            .unwrap_or(0);
        {
            let mut inner = self.inner.lock().unwrap();
            if let Some(pid) = child.process_id() {
                super::pidregistry::register(&self.block_id, pid);
                inner.child_pid = Some(pid);
            }
            inner.spawn_ts_ms = Some(spawn_ts_ms);
            inner.is_agent_pane = is_agent;
        }

        // Auto-register with jekt if AGENTMUX_AGENT_ID was set in the block env.
        // This maps agent_id → block_id in the ReactiveHandler so jekt can deliver
        // messages directly to this PTY without a separate /agentmux/reactive/register call.
        if let Some(ref agent_id) = agent_id_for_jekt {
            match crate::backend::reactive::get_global_handler()
                .register_agent(agent_id, &self.block_id, Some(&self.tab_id))
            {
                Ok(()) => {
                    tracing::info!(
                        block_id = %self.block_id,
                        agent_id = %agent_id,
                        "jekt: auto-registered"
                    );
                    // Also write to cross-instance file registry.
                    if let Ok(local_url) = std::env::var("AGENTMUX_LOCAL_URL") {
                        let data_dir = crate::backend::base::get_wave_data_dir();
                        crate::backend::reactive::registry::write(
                            &data_dir,
                            agent_id,
                            &local_url,
                            &self.block_id,
                        );
                    }
                }
                Err(e) => tracing::warn!(
                    block_id = %self.block_id,
                    agent_id = %agent_id,
                    error = %e,
                    "jekt: auto-register failed"
                ),
            }
        }
        tracing::info!(
            block_id = %self.block_id,
            wstore_present = self.wstore.is_some(),
            event_bus_present = self.event_bus.is_some(),
            "[dnd-debug] pre-seed state after spawn"
        );

        // Seed cmd:cwd in block meta immediately after spawn so drag-and-drop works
        // before the shell emits its first OSC 7 (or for shells without integration).
        if let Some(ref store) = self.wstore {
            let effective_cwd = if !cwd.is_empty() {
                cwd.clone()
            } else {
                std::env::current_dir()
                    .map(|p| p.to_string_lossy().to_string())
                    .unwrap_or_default()
            };
            tracing::debug!(block_id = %self.block_id, cwd = %effective_cwd, "seeding cmd:cwd");
            if !effective_cwd.is_empty() {
                let oref_str = format!("block:{}", self.block_id);
                let mut meta_update = MetaMapType::new();
                meta_update.insert(
                    super::META_KEY_CMD_CWD.to_string(),
                    serde_json::Value::String(effective_cwd),
                );
                // Only set if not already populated — don't clobber a restored session CWD
                match store.must_get::<crate::backend::obj::Block>(&self.block_id) {
                    Ok(block) if obj::meta_get_string(&block.meta, super::META_KEY_CMD_CWD, "").is_empty() => {
                        match crate::server::service::update_object_meta(store, &oref_str, &meta_update) {
                            Ok(()) => {
                                // Re-read updated block and broadcast obj:update so the
                                // frontend Jotai atom refreshes (update_object_meta only writes
                                // to SQLite — it does NOT send a WebSocket event on its own).
                                if let Ok(updated_block) = store.must_get::<crate::backend::obj::Block>(&self.block_id) {
                                    if let Some(ref event_bus) = self.event_bus {
                                        let update_data = serde_json::to_value(&obj::WaveObjUpdate {
                                            updatetype: "update".into(),
                                            otype: "block".into(),
                                            oid: self.block_id.clone(),
                                            obj: Some(obj::wave_obj_to_value(&updated_block)),
                                        }).ok();
                                        event_bus.broadcast_event(&crate::backend::eventbus::WSEventType {
                                            eventtype: "waveobj:update".to_string(),
                                            oref: oref_str.clone(),
                                            data: update_data,
                                        });
                                        tracing::info!(block_id = %self.block_id, "cmd:cwd seeded and broadcast to frontend");
                                    } else {
                                        tracing::warn!(block_id = %self.block_id, "cmd:cwd written to store but no event_bus to broadcast — frontend won't update");
                                    }
                                }
                            }
                            Err(e) => {
                                tracing::warn!(block_id = %self.block_id, error = %e, "failed to seed cmd:cwd in store");
                            }
                        }
                    }
                    Ok(_) => {
                        tracing::debug!(block_id = %self.block_id, "cmd:cwd already set, skipping seed");
                    }
                    Err(e) => {
                        tracing::warn!(block_id = %self.block_id, error = %e, "failed to read block for cmd:cwd seed");
                    }
                }
            }
        }

        // Get reader/writer from master
        let reader = pair.master.try_clone_reader().map_err(|e| {
            let _ = child.kill();
            let mut inner = self.inner.lock().unwrap();
            Self::set_status(&mut inner, STATUS_DONE);
            inner.proc_exit_code = -1;
            inner.input_tx = None;
            self.unlock_run();
            format!("failed to clone PTY reader: {e}")
        })?;

        let writer = pair.master.take_writer().map_err(|e| {
            let _ = child.kill();
            let mut inner = self.inner.lock().unwrap();
            Self::set_status(&mut inner, STATUS_DONE);
            inner.proc_exit_code = -1;
            inner.input_tx = None;
            self.unlock_run();
            format!("failed to take PTY writer: {e}")
        })?;

        // Spawn PTY read task (blocking I/O → spawn_blocking)
        let block_id_read = self.block_id.clone();
        let broker_read = self.broker.clone();
        let inner_read = self.inner.clone();
        let is_agent_read = is_agent;
        tokio::task::spawn_blocking(move || {
            let mut reader = reader;
            let mut buf = [0u8; PTY_READ_BUF_SIZE];

            // Phase 1.5 PR 1 (additive): if this is an agent pane,
            // also try to interpret stdout as Claude Code stream-json
            // line-by-line, feeding successful parses through
            // ClaudeTranslator and emitting AgentEvents on a new WPS
            // scope `agent_event:<block_id>`. The existing raw-chunk
            // path stays byte-equal — interactive panes (which don't
            // emit JSON) see no behavior change because every line
            // fails the JSON parse and is silently dropped. The
            // future stream-json-mode pane and the drone inspector
            // (issue #830 / Phase 1.5 PR 3) will be the first real
            // consumers.
            let mut translator: Option<crate::agents::translator::claude::ClaudeTranslator> =
                if is_agent_read {
                    Some(crate::agents::translator::claude::ClaudeTranslator::new())
                } else {
                    None
                };
            // Per-block line-buffer (raw bytes — see
            // `extract_agent_events`). Capped to AGENT_LINE_BUFFER_CAP
            // so a producer that never emits a newline can't grow
            // the buffer unboundedly.
            let mut line_buf: Vec<u8> = Vec::new();

            // OSC extractor: only for agent panes. Terminal panes forward
            // raw bytes to xterm.js which handles OSC natively — stripping
            // there would suppress the native window-title update. Agent
            // panes don't use xterm.js; OSC bytes in FileStore corrupt the
            // document renderer, so we extract and strip them here.
            let mut osc_extractor: Option<crate::backend::osc_extractor::OscExtractor> =
                if is_agent_read {
                    Some(crate::backend::osc_extractor::OscExtractor::new())
                } else {
                    None
                };

            loop {
                match reader.read(&mut buf) {
                    Ok(0) => break, // EOF
                    Ok(n) => {
                        inner_read.lock().unwrap().last_pty_output = Some(Instant::now());
                        if let Some(ref broker) = broker_read {
                            let raw = &buf[..n];

                            // Extract OSC sequences from agent-pane output.
                            // `cleaned` has OSC bytes removed; `osc_events` carries
                            // any normalised Claude Code title strings found in this chunk.
                            let mut cleaned_storage: Vec<u8> = Vec::new();
                            let mut osc_events: Vec<crate::backend::osc_extractor::OscEvent> = Vec::new();
                            if let Some(ref mut ext) = osc_extractor {
                                let (cleaned, evs) = ext.feed(raw);
                                cleaned_storage = cleaned;
                                osc_events = evs;
                            }
                            let chunk: &[u8] = if osc_extractor.is_some() {
                                &cleaned_storage
                            } else {
                                raw
                            };

                            handle_append_block_file(
                                broker,
                                &block_id_read,
                                "term",
                                chunk,
                                None, // PTY output is raw terminal data; no FileStore write-through
                                None, // not an agent output stream; no global mirror
                            );

                            for ev in &osc_events {
                                wps::publish_block_activity(broker, &block_id_read, &ev.payload);
                            }

                            if let Some(ref mut t) = translator {
                                accumulate_and_translate(
                                    broker,
                                    &block_id_read,
                                    &mut line_buf,
                                    chunk,
                                    t,
                                );
                            }
                        }
                    }
                    Err(e) => {
                        tracing::debug!("PTY read error for {}: {}", block_id_read, e);
                        break;
                    }
                }
            }
        });

        // Spawn input task (routes input channel → PTY writer + resize + signals)
        // Owns writer and master — dropping them closes the PTY, causing child to exit.
        let master = pair.master;
        tokio::spawn(async move {
            let mut writer = writer;
            let mut input_rx = input_rx;
            while let Some(input) = input_rx.recv().await {
                if let Some(data) = input.input_data {
                    use std::io::Write;
                    if let Err(e) = writer.write_all(&data) {
                        tracing::debug!("PTY write error: {}", e);
                        break;
                    }
                }
                if let Some(ref size) = input.term_size {
                    let pty_size = PtySize {
                        rows: size.rows as u16,
                        cols: size.cols as u16,
                        pixel_width: 0,
                        pixel_height: 0,
                    };
                    if let Err(e) = master.resize(pty_size) {
                        tracing::debug!("PTY resize error: {}", e);
                    }
                }
                if input.sig_name.is_some() {
                    // Drop writer + master to close PTY, which terminates the child
                    break;
                }
            }
            // writer and master drop here → PTY closes → child gets EOF/terminates
        });

        // Spawn wait task (monitors process exit)
        let inner_wait = Arc::clone(&self.inner);
        let block_id_wait = self.block_id.clone();
        let agent_id_wait = agent_id_for_jekt.clone();
        let broker_wait = self.broker.clone();
        let run_lock = Arc::clone(&self.run_lock);
        tokio::task::spawn_blocking(move || {
            let mut child = child;

            // Wait for child to exit (blocking)
            let exit_status = child.wait();
            let exit_code = match exit_status {
                Ok(status) => {
                    if status.success() {
                        0
                    } else {
                        // portable-pty ExitStatus doesn't expose raw code on all platforms
                        1
                    }
                }
                Err(e) => {
                    tracing::warn!("wait error for block {}: {}", block_id_wait, e);
                    -1
                }
            };

            tracing::info!(block_id = %block_id_wait, exit_code = exit_code, "process exited");

            // Unregister PID from per-pane metrics
            super::pidregistry::unregister(&block_id_wait);

            // Deregister from jekt — removes the agent_id → block_id mapping so
            // subsequent jekt attempts fall back to MessageBus rather than a dead PTY.
            crate::backend::reactive::get_global_handler().unregister_block(&block_id_wait);

            // Also remove from cross-instance file registry and cloud subscriber.
            if let Some(ref agent_id) = agent_id_wait {
                let data_dir = crate::backend::base::get_wave_data_dir();
                crate::backend::reactive::registry::remove(&data_dir, agent_id);
                if let Some(sub) = crate::muxbus::cloud_subscriber::get_global_subscriber() {
                    sub.remove_agent(agent_id);
                }
            }

            // Update inner state
            {
                let mut inner = inner_wait.lock().unwrap();
                inner.proc_exit_code = exit_code;
                ShellController::set_status(&mut inner, STATUS_DONE);
                inner.input_tx = None;
            }

            // Publish done status
            if let Some(ref broker) = broker_wait {
                let status = {
                    let inner = inner_wait.lock().unwrap();
                    BlockControllerRuntimeStatus {
                        blockid: block_id_wait.clone(),
                        version: inner.status_version,
                        shellprocstatus: inner.proc_status.clone(),
                        shellprocconnname: inner.conn_name.clone(),
                        shellprocexitcode: inner.proc_exit_code,
                        spawn_ts_ms: inner.spawn_ts_ms,
                        is_agent_pane: inner.is_agent_pane,
                    }
                };
                super::publish_controller_status(broker, &status);
            }

            // Release run lock
            run_lock.store(false, Ordering::SeqCst);
        });

        // Return immediately — PTY tasks run in background
        Ok(())
    }

    fn stop(&self, _graceful: bool, new_status: &str) -> Result<(), String> {
        // Extract what we need from the lock, release it before any async work.
        #[allow(unused_variables)] // used under #[cfg(unix)] only
        let pid_to_kill = {
            let mut inner = self.inner.lock().unwrap();
            if inner.proc_status == new_status {
                return Ok(());
            }
            let pid = inner.child_pid;
            // Drop the input channel — closes PTY writer → delivers EOF/SIGHUP as
            // belt-and-suspenders in case signal delivery fails on the platform.
            inner.input_tx = None;
            Self::set_status(&mut inner, new_status);
            pid
        };

        // Send SIGTERM to the process group so that child processes spawned by
        // the shell (e.g. `claude --dangerously-skip-permissions` and its subtree)
        // are also signalled. Negative pid targets the whole process group.
        // Schedule SIGKILL after KILL_GRACE_SECS as a backstop for processes
        // that ignore or delay on SIGTERM.
        #[cfg(unix)]
        if let Some(pid) = pid_to_kill {
            // SAFETY: kill() is a well-defined POSIX syscall.
            unsafe { libc::kill(-(pid as libc::pid_t), libc::SIGTERM) };
            tokio::spawn(async move {
                tokio::time::sleep(tokio::time::Duration::from_secs(KILL_GRACE_SECS)).await;
                unsafe { libc::kill(-(pid as libc::pid_t), libc::SIGKILL) };
            });
        }

        Ok(())
    }

    fn get_runtime_status(&self) -> BlockControllerRuntimeStatus {
        self.get_status_snapshot()
    }

    fn send_input(&self, input: BlockInputUnion, seq: Option<u64>) -> Result<(), String> {
        let mut inner = self.inner.lock().unwrap();
        let tx = match &inner.input_tx {
            Some(tx) => tx.clone(),
            None => return Err("controller is not running".to_string()),
        };
        match seq {
            None => tx.send(input).map_err(|e| format!("send_input: {e}")),
            Some(s) => {
                // Detect session reset: seq==0 means the TermViewModel
                // restarted and its per-block counter is back at zero.
                // (A gap-threshold heuristic was tried and removed — a
                // stale/duplicate packet far behind could falsely trigger
                // a reset and replay old input.)
                if s == 0 && inner.input_seq_next > 0 {
                    tracing::info!(
                        block_id = %self.block_id,
                        prev_next = inner.input_seq_next,
                        new_seq = s,
                        "input seq reset (session reset detected)"
                    );
                    inner.input_seq_next = s;
                    inner.input_seq_buf.clear();
                }

                let next = inner.input_seq_next;
                if s == next {
                    // Advance before sending — a send failure must not leave the
                    // backend stuck waiting for a seq the frontend will never resend.
                    inner.input_seq_next += 1;
                    if let Err(e) = tx.send(input) {
                        // Unbounded send only fails if the receiver is gone
                        // (controller stopped) — nothing to do but drop quietly.
                        tracing::warn!(
                            block_id = %self.block_id,
                            seq = s,
                            "send_input: input channel closed, discarding packet: {e}"
                        );
                        return Ok(());
                    }
                    // Drain any buffered out-of-order packets now in order.
                    loop {
                        let expected = inner.input_seq_next;
                        match inner.input_seq_buf.remove(&expected) {
                            Some(buffered) => {
                                inner.input_seq_next += 1;
                                if let Err(e) = tx.send(buffered) {
                                    tracing::warn!(
                                        block_id = %self.block_id,
                                        seq = expected,
                                        "send_input drain: input channel closed, discarding buffered packet: {e}"
                                    );
                                }
                            }
                            None => break,
                        }
                    }
                    Ok(())
                } else if s > next {
                    if inner.input_seq_buf.len() < SHELL_INPUT_CH_SIZE {
                        inner.input_seq_buf.insert(s, input);
                    } else {
                        tracing::warn!(block_id = %self.block_id, seq = s, "input reorder buffer full, dropping");
                    }
                    Ok(())
                } else {
                    tracing::warn!(block_id = %self.block_id, seq = s, next, "duplicate input seq, discarding");
                    Ok(())
                }
            }
        }
    }

    fn controller_type(&self) -> &str {
        &self.controller_type
    }

    fn block_id(&self) -> &str {
        &self.block_id
    }

    fn as_any(&self) -> &dyn std::any::Any {
        self
    }
}

// ---- File operation helpers ----

/// Append data to a block's terminal output file, publish a WPS event,
/// Maximum size of the per-block line buffer used by the agent-event
/// translation path. Past this, the buffer is reset (a producer that
/// never emits a newline can't grow it unboundedly). 1 MiB is far
/// beyond any plausible stream-json frame size.
const AGENT_LINE_BUFFER_CAP: usize = 1024 * 1024;

/// Append `chunk` to `line_buf`, drain complete lines, JSON-parse
/// each, and run successful parses through `translator`. Returns the
/// events the translator emitted. Caller is responsible for
/// publishing them.
///
/// Buffers RAW BYTES (not lossy-decoded strings) so a multi-byte
/// UTF-8 character split across two PTY reads decodes cleanly when
/// the complete line arrives. Decoding each chunk lossily would
/// insert U+FFFD into the middle of words for non-ASCII content
/// (CJK, emoji, accented chars), silently corrupting
/// `AssistantText`/`Done.response` while the parallel raw-byte WPS
/// path stays correct — drone consumers would have no way to
/// recover. Reagent P1 + codex P2 on PR #833.
///
/// Pure function — split out from `accumulate_and_translate` so the
/// line-buffering + JSON-fast-reject + translator-call logic is
/// unit-testable without spinning up a broker.
fn extract_agent_events(
    line_buf: &mut Vec<u8>,
    chunk: &[u8],
    translator: &mut crate::agents::translator::claude::ClaudeTranslator,
) -> Vec<crate::agents::types::AgentEvent> {
    use crate::agents::translator::Translator as _;
    let mut out: Vec<crate::agents::types::AgentEvent> = Vec::new();
    line_buf.extend_from_slice(chunk);
    if line_buf.len() > AGENT_LINE_BUFFER_CAP {
        // No newline in a megabyte — definitely not stream-json.
        // Reset to keep memory bounded.
        line_buf.clear();
        return out;
    }
    while let Some(nl) = line_buf.iter().position(|&b| b == b'\n') {
        let line_bytes: Vec<u8> = line_buf.drain(..=nl).collect();
        // Decode the COMPLETE line as lossy UTF-8 — split codepoints
        // are now fully present, so lossy-vs-strict only matters for
        // genuinely malformed bytes which would also fail strict.
        let line = String::from_utf8_lossy(&line_bytes);
        let trimmed = line.trim_end_matches(['\n', '\r']);
        if !trimmed.starts_with('{') {
            // Fast-reject: stream-json frames are JSON objects.
            // Skips ANSI escapes, prompts, blank lines, etc.
            continue;
        }
        let Ok(frame) = serde_json::from_str::<serde_json::Value>(trimmed) else {
            continue;
        };
        out.extend(translator.translate(frame));
    }
    out
}

/// Publish the events `extract_agent_events` produced on the WPS
/// scope `agent_event:<block_id>`. Phase 1.5 PR 1 hook for agent
/// panes. Called only when `is_agent` is true at spawn time (see
/// read-task closure in `start()`).
fn accumulate_and_translate(
    broker: &wps::Broker,
    block_id: &str,
    line_buf: &mut Vec<u8>,
    chunk: &[u8],
    translator: &mut crate::agents::translator::claude::ClaudeTranslator,
) {
    for event in extract_agent_events(line_buf, chunk, translator) {
        broker.publish(wps::WaveEvent {
            event: format!("agent_event:{}", block_id),
            scopes: vec![],
            sender: String::new(),
            persist: 0,
            data: Some(serde_json::to_value(&event).unwrap_or_default()),
        });
    }
}

/// and write-through to FileStore (if provided).
/// Persist `data` to the block's output file and the global transcript zone
/// **without** publishing a WPS event. Used by the persistent controller to
/// record user-message lines for future history loads (so that
/// `parseHistoryLines` can reconstruct `user_message` nodes on reopen) without
/// triggering a live-stream append that would produce a duplicate node alongside
/// the `agent-message-accepted` UUID node. Same lazy-create semantics as
/// `handle_append_block_file`.
pub fn persist_to_blockfile_silent(
    block_id: &str,
    filename: &str,
    data: &[u8],
    filestore: Option<&Arc<FileStore>>,
    global_output_zone: Option<&str>,
) {
    if let Some(fs) = filestore {
        let needs_create = match fs.stat(block_id, filename) {
            Ok(None) => true,
            Ok(Some(_)) => false,
            Err(e) => {
                tracing::warn!(
                    block_id = %block_id, filename = %filename, error = %e,
                    "persist_silent: stat failed; skipping"
                );
                return;
            }
        };
        if needs_create {
            if let Err(e) = fs.make_file(
                block_id,
                filename,
                std::collections::HashMap::new(),
                crate::backend::storage::filestore::FileOpts::default(),
            ) {
                use crate::backend::storage::error::StoreError;
                if !matches!(e, StoreError::AlreadyExists) {
                    tracing::warn!(
                        block_id = %block_id, filename = %filename, error = %e,
                        "persist_silent: make_file failed; skipping"
                    );
                    return;
                }
            }
        }
        if let Err(e) = fs.append_data(block_id, filename, data) {
            tracing::warn!(
                block_id = %block_id, filename = %filename, error = %e,
                "persist_silent: append_data failed"
            );
        }
    }
    if let Some(zone) = global_output_zone {
        if let Some(gfs) = crate::backend::agent_session::global_transcript_store() {
            mirror_append_to_global(gfs, zone, data);
        }
    }
}

///
/// Port of Go's `HandleAppendBlockFile`.
///
/// The FileStore write is fire-and-forget: if it fails we emit a warning but
/// never propagate the error back to the hot stdout-reader path.
pub fn handle_append_block_file(
    broker: &wps::Broker,
    block_id: &str,
    filename: &str,
    data: &[u8],
    filestore: Option<&Arc<FileStore>>,
    global_output_zone: Option<&str>,
) {
    let data64 = base64::engine::general_purpose::STANDARD.encode(data);

    let event_data = wps::WSFileEventData {
        zoneid: block_id.to_string(),
        filename: filename.to_string(),
        fileop: wps::FILE_OP_APPEND.to_string(),
        data64,
    };

    let event = wps::WaveEvent {
        event: wps::EVENT_BLOCK_FILE.to_string(),
        scopes: vec![format!("block:{block_id}")],
        sender: String::new(),
        persist: 0,
        data: serde_json::to_value(&event_data).ok(),
    };

    broker.publish(event);

    // Write-through to FileStore for persistent history (Phase 1.3).
    // Create the file lazily on first append; if the file already exists
    // we skip make_file and go straight to append_data.
    if let Some(fs) = filestore {
        let needs_create = match fs.stat(block_id, filename) {
            Ok(None) => true,
            Ok(Some(_)) => false,
            Err(e) => {
                tracing::warn!(
                    block_id = %block_id,
                    filename = %filename,
                    error = %e,
                    "filestore stat failed; skipping write-through"
                );
                return;
            }
        };

        if needs_create {
            if let Err(e) = fs.make_file(
                block_id,
                filename,
                std::collections::HashMap::new(),
                crate::backend::storage::filestore::FileOpts::default(),
            ) {
                // AlreadyExists is benign (race between two appends); anything
                // else is worth warning about.
                use crate::backend::storage::error::StoreError;
                if !matches!(e, StoreError::AlreadyExists) {
                    tracing::warn!(
                        block_id = %block_id,
                        filename = %filename,
                        error = %e,
                        "filestore make_file failed; skipping write-through"
                    );
                    return;
                }
            }
        }

        if let Err(e) = fs.append_data(block_id, filename, data) {
            tracing::warn!(
                block_id = %block_id,
                filename = %filename,
                error = %e,
                "filestore append_data failed"
            );
        }
        // Note: output.idx is NOT updated incrementally here. It is a lazily-built,
        // self-validating cache rebuilt by the read path whenever output grows (see
        // `rebuild_output_idx` below, invoked from the blockfile:read_range handler
        // in app_api.rs). This avoids every incremental-index failure mode (desync on
        // write failure, chunk-split lines, blank-line miscounting) at the cost of one
        // rescan per output-size change.
    }

    // Mirror the agent's `output` stream into the GLOBAL transcript zone
    // (`agent:<defId>:current`) so the conversation loads when this agent is
    // opened from another build/channel. Fire-and-forget, exactly like the
    // per-channel write above: a global-store hiccup must never stall the live
    // pane. Callers pass `Some(zone)` only for the agent output stream, so we
    // never mirror PTY `term` data or non-agent blocks.
    if let Some(zone) = global_output_zone {
        if let Some(gfs) = crate::backend::agent_session::global_transcript_store() {
            mirror_append_to_global(gfs, zone, data);
        }
    }
}

/// Append `data` to the global transcript zone's `output` file, creating it
/// lazily on first write. Mirrors the per-channel write-through in
/// [`handle_append_block_file`]; all errors are logged and swallowed so the
/// hot stdout path is never blocked by the global store.
fn mirror_append_to_global(gfs: &Arc<FileStore>, zone: &str, data: &[u8]) {
    use crate::backend::agent_session::OUTPUT_FILE;
    use crate::backend::storage::error::StoreError;

    match gfs.stat(zone, OUTPUT_FILE) {
        Ok(None) => {
            if let Err(e) = gfs.make_file(
                zone,
                OUTPUT_FILE,
                std::collections::HashMap::new(),
                crate::backend::storage::filestore::FileOpts::default(),
            ) {
                if !matches!(e, StoreError::AlreadyExists) {
                    tracing::warn!(zone = %zone, error = %e, "global transcripts: make_file failed; skipping mirror");
                    return;
                }
            }
        }
        Ok(Some(_)) => {}
        Err(e) => {
            tracing::warn!(zone = %zone, error = %e, "global transcripts: stat failed; skipping mirror");
            return;
        }
    }
    if let Err(e) = gfs.append_data(zone, OUTPUT_FILE, data) {
        tracing::warn!(zone = %zone, error = %e, "global transcripts: append_data failed");
    }
}

/// Resolve a block's GLOBAL transcript zone (`agent:<defId>:current`) from its
/// `agentId` meta, looking the block up in `wstore`. Returns `None` for
/// non-agent blocks, when there's no store, or when the block can't be loaded —
/// the caller then passes `None` and no global mirror happens. Shared by the
/// subprocess / persistent / acp agent controllers.
pub(crate) fn resolve_global_output_zone(
    wstore: &Option<Arc<crate::backend::storage::store::Store>>,
    block_id: &str,
) -> Option<String> {
    let store = wstore.as_ref()?;
    let block = store
        .must_get::<crate::backend::obj::Block>(block_id)
        .ok()?;
    crate::backend::agent_session::agent_zone_for_block_meta(&block.meta)
}

/// Magic header size for `output.idx`: the first 8 bytes are the `output` byte-size
/// the index was built for. The index is valid iff this equals `output`'s current
/// size; otherwise it is stale and must be rebuilt.
pub(crate) const OUTPUT_IDX_HEADER_LEN: i64 = 8;

/// Rebuild `output.idx` from `output` in a single streaming scan and atomically
/// replace it. The index is the byte offset of every **non-blank** line, matching
/// the reader's line addressing (`String::lines().filter(!trim().is_empty())`).
///
/// Layout: `[covered_size: u64-LE][offset_0: u64-LE][offset_1]...`. `covered_size`
/// records the `output` size this index reflects so the read path can detect
/// staleness in O(1) and rebuild only when `output` actually grew.
///
/// The scan streams `output` in 1 MiB windows so memory stays O(one line + offsets)
/// rather than loading the whole (potentially multi-GB) file. Line splitting is
/// done on raw bytes; since UTF-8 continuation bytes never collide with `\n`, lines
/// are never split mid-codepoint, so per-line `from_utf8_lossy` matches the reader.
///
/// Returns the number of indexed (non-blank) lines on success, or `None` if
/// `output` is unreadable or the index write fails (caller falls back to slow path).
pub(crate) fn rebuild_output_idx(
    fs: &FileStore,
    block_id: &str,
    output_size: u64,
) -> Option<u64> {
    const IDX: &str = "output.idx";
    const WIN: i64 = 1 << 20; // 1 MiB read window

    // Offsets buffer starts with the covered-size header.
    let mut buf: Vec<u8> = Vec::new();
    buf.extend_from_slice(&output_size.to_le_bytes());

    let mut line_count: u64 = 0;
    let mut cursor: u64 = 0; // byte offset where the current line begins
    let mut line_buf: Vec<u8> = Vec::new(); // bytes of the current line, excluding '\n'
    let mut read_pos: i64 = 0;

    let flush_line = |line_buf: &mut Vec<u8>,
                      cursor: &mut u64,
                      buf: &mut Vec<u8>,
                      line_count: &mut u64,
                      had_newline: bool| {
        // The reader strips a trailing '\r' (CRLF) and treats trim-empty as blank.
        let is_blank = String::from_utf8_lossy(line_buf).trim().is_empty();
        if !is_blank {
            buf.extend_from_slice(&cursor.to_le_bytes());
            *line_count += 1;
        }
        // Advance cursor past this line's bytes (+1 for the consumed '\n').
        *cursor += line_buf.len() as u64 + if had_newline { 1 } else { 0 };
        line_buf.clear();
    };

    while read_pos < output_size as i64 {
        let (_, chunk) = fs.read_at(block_id, "output", read_pos, WIN).ok()?;
        if chunk.is_empty() {
            break;
        }
        for &b in &chunk {
            if b == b'\n' {
                flush_line(&mut line_buf, &mut cursor, &mut buf, &mut line_count, true);
            } else {
                line_buf.push(b);
            }
        }
        read_pos += chunk.len() as i64;
    }
    // Trailing line with no final '\n'.
    if !line_buf.is_empty() {
        flush_line(&mut line_buf, &mut cursor, &mut buf, &mut line_count, false);
    }

    if let Ok(None) = fs.stat(block_id, IDX) {
        let _ = fs.make_file(
            block_id,
            IDX,
            std::collections::HashMap::new(),
            crate::backend::storage::filestore::FileOpts::default(),
        );
    }
    match fs.write_file(block_id, IDX, &buf) {
        Ok(()) => {
            tracing::info!(block_id = %block_id, lines = line_count, covered = output_size, "output.idx rebuilt");
            Some(line_count)
        }
        Err(e) => {
            tracing::warn!(block_id = %block_id, error = %e, "output.idx rebuild write failed");
            None
        }
    }
}

/// Truncate a block's terminal output file and publish a WPS event.
/// Port of Go's `HandleTruncateBlockFile`.
#[allow(dead_code)]
pub fn handle_truncate_block_file(broker: &wps::Broker, block_id: &str, filename: &str) {
    let event_data = wps::WSFileEventData {
        zoneid: block_id.to_string(),
        filename: filename.to_string(),
        fileop: wps::FILE_OP_TRUNCATE.to_string(),
        data64: String::new(),
    };

    let event = wps::WaveEvent {
        event: wps::EVENT_BLOCK_FILE.to_string(),
        scopes: vec![format!("block:{block_id}")],
        sender: String::new(),
        persist: 0,
        data: serde_json::to_value(&event_data).ok(),
    };

    broker.publish(event);
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::backend::shellexec::MockConn;
    use std::sync::Arc;

    fn make_shell_meta() -> MetaMapType {
        let mut meta = MetaMapType::new();
        meta.insert(
            "controller".to_string(),
            serde_json::Value::String("shell".to_string()),
        );
        meta
    }

    fn make_cmd_meta(cmd: &str) -> MetaMapType {
        let mut meta = MetaMapType::new();
        meta.insert(
            "controller".to_string(),
            serde_json::Value::String("cmd".to_string()),
        );
        meta.insert(
            "cmd".to_string(),
            serde_json::Value::String(cmd.to_string()),
        );
        meta
    }

    // ── pty_size_from_rt_opts ────────────────────────────────────────────
    // Seeds the initial PTY geometry from the resync `rtopts` payload so the
    // agent pane is born at the right width (no post-spawn resize race).
    // See docs/analysis/AGENT_PANE_PTY_RESIZE_RACE_2026_06_16.md.

    #[test]
    fn pty_size_defaults_when_rt_opts_absent() {
        let sz = ShellController::pty_size_from_rt_opts(&None);
        assert_eq!((sz.rows, sz.cols), (25, 200));
    }

    #[test]
    fn pty_size_defaults_when_termsize_is_serde_default() {
        // rows==0 && cols==0 is the serde default → treat as absent.
        let v = serde_json::json!({ "termsize": { "rows": 0, "cols": 0 } });
        let sz = ShellController::pty_size_from_rt_opts(&Some(v));
        assert_eq!((sz.rows, sz.cols), (25, 200));
    }

    #[test]
    fn pty_size_honors_supplied_termsize() {
        let v = serde_json::json!({ "termsize": { "rows": 50, "cols": 130 } });
        let sz = ShellController::pty_size_from_rt_opts(&Some(v));
        assert_eq!((sz.rows, sz.cols), (50, 130));
    }

    #[test]
    fn pty_size_keeps_default_rows_for_cols_only_payload() {
        let v = serde_json::json!({ "termsize": { "rows": 0, "cols": 130 } });
        let sz = ShellController::pty_size_from_rt_opts(&Some(v));
        assert_eq!((sz.rows, sz.cols), (25, 130));
    }

    #[test]
    fn pty_size_clamps_oversized_values() {
        let v = serde_json::json!({ "termsize": { "rows": 99999, "cols": 99999 } });
        let sz = ShellController::pty_size_from_rt_opts(&Some(v));
        assert_eq!((sz.rows, sz.cols), (1000, 1000));
    }

    #[test]
    fn pty_size_defaults_on_unparseable_rt_opts() {
        // Unknown keys deserialize to RuntimeOpts default (termsize 0/0) → fallback.
        let v = serde_json::json!({ "totally": "unrelated" });
        let sz = ShellController::pty_size_from_rt_opts(&Some(v));
        assert_eq!((sz.rows, sz.cols), (25, 200));
    }

    #[test]
    fn pty_size_ignores_non_positive_axes() {
        // Negative axes fail the `> 0` guard and keep their default.
        let v = serde_json::json!({ "termsize": { "rows": -5, "cols": -1 } });
        let sz = ShellController::pty_size_from_rt_opts(&Some(v));
        assert_eq!((sz.rows, sz.cols), (25, 200));
    }

    #[test]
    fn test_shell_controller_new() {
        let ctrl = ShellController::new(
            "shell".to_string(),
            "tab-1".to_string(),
            "block-1".to_string(),
            None,
            None,
            None,
        );
        assert_eq!(ctrl.controller_type(), "shell");
        assert_eq!(ctrl.block_id(), "block-1");

        let status = ctrl.get_runtime_status();
        assert_eq!(status.shellprocstatus, STATUS_INIT);
        assert_eq!(status.blockid, "block-1");
        assert_eq!(status.version, 0);
    }

    #[test]
    fn test_shell_controller_start_stop() {
        let ctrl = ShellController::new(
            "shell".to_string(),
            "tab-1".to_string(),
            "block-1".to_string(),
            None,
            None,
            None,
        );

        // Use mock factory so we don't open a real PTY in tests
        ctrl.set_conn_factory(Box::new(|_conn_name, _meta| {
            Ok(Box::new(MockConn::new(0)) as Box<dyn ConnInterface>)
        }));

        let meta = make_shell_meta();
        let result = ctrl.start(meta, None, false);
        assert!(result.is_ok());

        // After start with mock, process immediately exits → status is done
        let status = ctrl.get_runtime_status();
        assert_eq!(status.shellprocstatus, STATUS_DONE);

        // Stop should work
        let result = ctrl.stop(true, STATUS_DONE);
        assert!(result.is_ok());
    }

    #[test]
    fn test_shell_controller_run_on_start_false() {
        let ctrl = ShellController::new(
            "shell".to_string(),
            "tab-1".to_string(),
            "block-1".to_string(),
            None,
            None,
            None,
        );

        let mut meta = make_shell_meta();
        meta.insert(
            META_KEY_CMD_RUN_ON_START.to_string(),
            serde_json::Value::Bool(false),
        );

        let result = ctrl.start(meta, None, false);
        assert!(result.is_ok());

        // Should still be in init state (didn't start)
        let status = ctrl.get_runtime_status();
        assert_eq!(status.shellprocstatus, STATUS_INIT);
    }

    #[test]
    fn test_shell_controller_force_start() {
        let ctrl = ShellController::new(
            "shell".to_string(),
            "tab-1".to_string(),
            "block-1".to_string(),
            None,
            None,
            None,
        );

        ctrl.set_conn_factory(Box::new(|_conn_name, _meta| {
            Ok(Box::new(MockConn::new(0)) as Box<dyn ConnInterface>)
        }));

        let mut meta = make_shell_meta();
        meta.insert(
            META_KEY_CMD_RUN_ON_START.to_string(),
            serde_json::Value::Bool(false),
        );

        // Force should override run_on_start=false
        let result = ctrl.start(meta, None, true);
        assert!(result.is_ok());

        let status = ctrl.get_runtime_status();
        // With mock, immediately exits to done
        assert_eq!(status.shellprocstatus, STATUS_DONE);
    }

    #[test]
    fn test_shell_controller_with_conn_factory() {
        let ctrl = ShellController::new(
            "cmd".to_string(),
            "tab-1".to_string(),
            "block-1".to_string(),
            None,
            None,
            None,
        );

        // Set a custom factory that returns a mock with exit code 42
        ctrl.set_conn_factory(Box::new(|_conn_name, _meta| {
            Ok(Box::new(MockConn::new(42)) as Box<dyn ConnInterface>)
        }));

        let meta = make_cmd_meta("echo hello");
        let result = ctrl.start(meta, None, true);
        assert!(result.is_ok());

        let status = ctrl.get_runtime_status();
        assert_eq!(status.shellprocstatus, STATUS_DONE);
        assert_eq!(status.shellprocexitcode, 42);
    }

    #[test]
    fn test_shell_controller_conn_factory_error() {
        let ctrl = ShellController::new(
            "shell".to_string(),
            "tab-1".to_string(),
            "block-1".to_string(),
            None,
            None,
            None,
        );

        ctrl.set_conn_factory(Box::new(|_conn_name, _meta| {
            Err("connection refused".to_string())
        }));

        let meta = make_shell_meta();
        let result = ctrl.start(meta, None, true);
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("connection refused"));

        let status = ctrl.get_runtime_status();
        assert_eq!(status.shellprocstatus, STATUS_DONE);
        assert_eq!(status.shellprocexitcode, -1);
    }

    #[test]
    fn test_shell_controller_send_input_not_running() {
        let ctrl = ShellController::new(
            "shell".to_string(),
            "tab-1".to_string(),
            "block-1".to_string(),
            None,
            None,
            None,
        );

        let result = ctrl.send_input(BlockInputUnion::data(b"hello".to_vec()), None);
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("not running"));
    }

    #[test]
    fn test_shell_controller_status_version_increments() {
        let ctrl = ShellController::new(
            "shell".to_string(),
            "tab-1".to_string(),
            "block-1".to_string(),
            None,
            None,
            None,
        );

        ctrl.set_conn_factory(Box::new(|_conn_name, _meta| {
            Ok(Box::new(MockConn::new(0)) as Box<dyn ConnInterface>)
        }));

        let v0 = ctrl.get_runtime_status().version;

        let meta = make_shell_meta();
        ctrl.start(meta, None, true).unwrap();

        let v_after = ctrl.get_runtime_status().version;
        // Status changed from init → running → done = at least 2 increments
        assert!(v_after > v0);
    }

    #[test]
    fn test_controller_trait_as_arc() {
        let ctrl: Arc<dyn Controller> = Arc::new(ShellController::new(
            "shell".to_string(),
            "tab-1".to_string(),
            "block-1".to_string(),
            None,
            None,
            None,
        ));

        assert_eq!(ctrl.controller_type(), "shell");
        assert_eq!(ctrl.block_id(), "block-1");
        let status = ctrl.get_runtime_status();
        assert_eq!(status.shellprocstatus, STATUS_INIT);
    }

    #[test]
    fn test_meta_helpers() {
        let mut meta = MetaMapType::new();
        assert!(ShellController::should_run_on_start(&meta)); // default true
        assert!(!ShellController::should_run_once(&meta)); // default false
        assert!(!ShellController::should_clear_on_start(&meta)); // default false
        assert!(!ShellController::should_close_on_exit(&meta)); // default false

        meta.insert(
            META_KEY_CMD_RUN_ON_START.to_string(),
            serde_json::Value::Bool(false),
        );
        assert!(!ShellController::should_run_on_start(&meta));

        meta.insert(
            META_KEY_CMD_RUN_ONCE.to_string(),
            serde_json::Value::Bool(true),
        );
        assert!(ShellController::should_run_once(&meta));

        meta.insert(
            META_KEY_CMD_CLEAR_ON_START.to_string(),
            serde_json::Value::Bool(true),
        );
        assert!(ShellController::should_clear_on_start(&meta));
    }

    #[test]
    fn test_close_on_exit_delay() {
        let mut meta = MetaMapType::new();
        assert_eq!(ShellController::close_on_exit_delay_ms(&meta), 2000); // default

        meta.insert(
            META_KEY_CMD_CLOSE_ON_EXIT_DELAY.to_string(),
            serde_json::json!(5000),
        );
        assert_eq!(ShellController::close_on_exit_delay_ms(&meta), 5000);
    }

    #[test]
    fn test_conn_name_from_meta() {
        let mut meta = MetaMapType::new();
        assert_eq!(ShellController::get_conn_name(&meta), "local"); // default

        meta.insert(
            META_KEY_CONNECTION.to_string(),
            serde_json::Value::String("user@host".to_string()),
        );
        assert_eq!(ShellController::get_conn_name(&meta), "user@host");
    }

    #[test]
    fn test_handle_append_block_file() {
        let broker = wps::Broker::new();

        // Subscribe to block file events
        broker.subscribe(
            "test-route",
            wps::SubscriptionRequest {
                event: wps::EVENT_BLOCK_FILE.to_string(),
                scopes: vec!["block:block-1".to_string()],
                allscopes: false,
            },
        );

        handle_append_block_file(&broker, "block-1", "term", b"hello world", None, None);

        // Check event was published
        let _history = broker.read_event_history(wps::EVENT_BLOCK_FILE, "block:block-1", 10);
        // Note: events are only persisted if persist > 0, so we verify via the publish mechanism
        // The broker successfully processed without panic, which verifies correctness
    }

    /// Helper: read all non-blank line offsets back out of a rebuilt output.idx,
    /// returning (covered_size, offsets).
    #[cfg(test)]
    fn read_idx(fs: &FileStore, block_id: &str) -> (u64, Vec<u64>) {
        let raw = fs.read_file(block_id, "output.idx").unwrap().unwrap();
        let covered = u64::from_le_bytes(raw[0..8].try_into().unwrap());
        let offsets = raw[8..]
            .chunks_exact(8)
            .map(|c| u64::from_le_bytes(c.try_into().unwrap()))
            .collect();
        (covered, offsets)
    }

    #[test]
    fn test_rebuild_output_idx_basic() {
        use crate::backend::storage::filestore::FileStore;
        let fs = FileStore::open_in_memory().expect("filestore");
        let bid = "idx-block";
        let data = b"line0\nline1\nline2\n";
        fs.make_file(bid, "output", Default::default(), Default::default()).unwrap();
        fs.append_data(bid, "output", data).unwrap();

        let n = rebuild_output_idx(&fs, bid, data.len() as u64).unwrap();
        assert_eq!(n, 3);
        let (covered, offsets) = read_idx(&fs, bid);
        assert_eq!(covered, data.len() as u64);
        // "line0\n"=0, "line1\n"=6, "line2\n"=12
        assert_eq!(offsets, vec![0, 6, 12]);
    }

    #[test]
    fn test_rebuild_output_idx_blank_and_crlf_and_no_trailing_nl() {
        use crate::backend::storage::filestore::FileStore;
        let fs = FileStore::open_in_memory().expect("filestore");
        let bid = "idx-block2";
        // Blank line (just spaces), a CRLF line, a blank line, and a final line
        // with no trailing newline. Non-blank lines start at: 0 ("a\n"),
        // 8 ("b\r\n" after "a\n   \n"=6 ... let's compute precisely below).
        // bytes: "a\n"   (0..2)
        //        "   \n" (2..6)   blank
        //        "b\r\n" (6..9)   non-blank -> offset 6
        //        "\n"    (9..10)  blank
        //        "tail"  (10..14) non-blank, no trailing nl -> offset 10
        let data = b"a\n   \nb\r\n\ntail";
        fs.make_file(bid, "output", Default::default(), Default::default()).unwrap();
        fs.append_data(bid, "output", data).unwrap();

        let n = rebuild_output_idx(&fs, bid, data.len() as u64).unwrap();
        assert_eq!(n, 3, "a, b(crlf), tail are the 3 non-blank lines");
        let (_covered, offsets) = read_idx(&fs, bid);
        assert_eq!(offsets, vec![0, 6, 10]);

        // Sanity: the recorded offsets really do start the expected non-blank lines.
        let full = fs.read_file(bid, "output").unwrap().unwrap();
        assert_eq!(&full[0..1], b"a");
        assert_eq!(&full[6..7], b"b");
        assert_eq!(&full[10..14], b"tail");
    }

    #[test]
    fn test_rebuild_output_idx_empty() {
        use crate::backend::storage::filestore::FileStore;
        let fs = FileStore::open_in_memory().expect("filestore");
        let bid = "idx-empty";
        fs.make_file(bid, "output", Default::default(), Default::default()).unwrap();
        let n = rebuild_output_idx(&fs, bid, 0).unwrap();
        assert_eq!(n, 0);
        let (covered, offsets) = read_idx(&fs, bid);
        assert_eq!(covered, 0);
        assert!(offsets.is_empty());
    }

    #[test]
    fn test_handle_truncate_block_file() {
        let broker = wps::Broker::new();
        // Should not panic
        handle_truncate_block_file(&broker, "block-1", "term");
    }

    #[test]
    fn test_register_and_get_controller() {
        let ctrl: Arc<dyn Controller> = Arc::new(ShellController::new(
            "shell".to_string(),
            "tab-1".to_string(),
            "test-register-block".to_string(),
            None,
            None,
            None,
        ));

        super::super::register_controller("test-register-block", ctrl.clone());

        let retrieved = super::super::get_controller("test-register-block");
        assert!(retrieved.is_some());
        assert_eq!(retrieved.unwrap().block_id(), "test-register-block");

        // Cleanup
        super::super::delete_controller("test-register-block");
        assert!(super::super::get_controller("test-register-block").is_none());
    }

    #[test]
    fn test_resync_creates_shell_controller() {
        use crate::backend::obj::Block;

        let mut meta = MetaMapType::new();
        meta.insert(
            "controller".to_string(),
            serde_json::Value::String("shell".to_string()),
        );
        // Disable auto-start so we don't open a real PTY in tests
        meta.insert(
            META_KEY_CMD_RUN_ON_START.to_string(),
            serde_json::Value::Bool(false),
        );

        let block = Block {
            oid: "resync-test-block".to_string(),
            version: 1,
            meta,
            ..Default::default()
        };

        let result = super::super::resync_controller(&block, "tab-1", None, false, None, None, None, None);
        assert!(result.is_ok());

        let ctrl = super::super::get_controller("resync-test-block");
        assert!(ctrl.is_some());
        assert_eq!(ctrl.unwrap().controller_type(), "shell");

        // Cleanup
        super::super::delete_controller("resync-test-block");
    }

    /// Phase 1.3 integration test: write output via handle_append_block_file with a
    /// FileStore, then verify the data is readable back from the store.
    #[test]
    fn test_handle_append_block_file_writes_to_filestore() {
        use crate::backend::storage::filestore::FileStore;
        use std::sync::Arc;

        let broker = wps::Broker::new();
        let fs = Arc::new(FileStore::open_in_memory().expect("open in-memory filestore"));

        let block_id = "test-block-fs";
        let filename = "output";

        // First append — file does not exist yet; handle_append_block_file must create it lazily.
        let line1 = b"line one\n";
        handle_append_block_file(&broker, block_id, filename, line1, Some(&fs), None);

        // Second append
        let line2 = b"line two\n";
        handle_append_block_file(&broker, block_id, filename, line2, Some(&fs), None);

        // Read back from FileStore
        let data = fs.read_file(block_id, filename)
            .expect("read_file ok")
            .expect("data present");

        let text = String::from_utf8(data).expect("valid utf8");
        assert!(text.contains("line one"), "expected 'line one' in {:?}", text);
        assert!(text.contains("line two"), "expected 'line two' in {:?}", text);

        // Also verify total size matches
        let stat = fs.stat(block_id, filename).unwrap().unwrap();
        assert_eq!(stat.size, (line1.len() + line2.len()) as i64);

        // Verify WPS events were also published (broker path still works)
        broker.subscribe(
            "test-route-fs",
            wps::SubscriptionRequest {
                event: wps::EVENT_BLOCK_FILE.to_string(),
                scopes: vec![format!("block:{}", block_id)],
                allscopes: false,
            },
        );
        // Re-publish one more line to confirm broker still receives events alongside filestore
        handle_append_block_file(&broker, block_id, filename, b"line three\n", Some(&fs), None);
        let stat_after = fs.stat(block_id, filename).unwrap().unwrap();
        assert_eq!(stat_after.size, (line1.len() + line2.len() + b"line three\n".len()) as i64);
    }

    // ────────────────────────────────────────────────────────────────
    // Cross-channel global transcript mirror
    // ────────────────────────────────────────────────────────────────

    #[test]
    fn mirror_append_to_global_creates_and_appends() {
        use crate::backend::agent_session::OUTPUT_FILE;
        let gfs = Arc::new(FileStore::open_in_memory().expect("global filestore"));
        let zone = "agent:def-mirror-1:current";

        // First append creates the file lazily.
        mirror_append_to_global(&gfs, zone, b"{\"type\":\"user\"}\n");
        // Second append extends it.
        mirror_append_to_global(&gfs, zone, b"{\"type\":\"assistant\"}\n");

        let data = gfs
            .read_file(zone, OUTPUT_FILE)
            .expect("read ok")
            .expect("present");
        let text = String::from_utf8(data).unwrap();
        assert!(text.contains("\"user\""), "got {text:?}");
        assert!(text.contains("\"assistant\""), "got {text:?}");
        // Two NDJSON lines.
        assert_eq!(text.lines().filter(|l| !l.trim().is_empty()).count(), 2);
    }

    #[test]
    fn resolve_global_output_zone_maps_agent_block() {
        let wstore = Arc::new(Store::open_in_memory().expect("wstore"));

        // Agent-anchored block → zone resolved from agentId meta.
        let oid = uuid::Uuid::new_v4().to_string();
        let mut meta = MetaMapType::new();
        meta.insert("view".to_string(), serde_json::json!("agent"));
        meta.insert("agentId".to_string(), serde_json::json!("def-zone-1"));
        let mut block = obj::Block {
            oid: oid.clone(),
            parentoref: String::new(),
            version: 1,
            runtimeopts: None,
            stickers: None,
            meta,
            subblockids: None,
        };
        wstore.insert(&mut block).expect("insert block");

        let some = Some(wstore.clone());
        assert_eq!(
            resolve_global_output_zone(&some, &oid).as_deref(),
            Some("agent:def-zone-1:current"),
        );

        // Unknown block id → None (no crash).
        assert_eq!(resolve_global_output_zone(&some, "no-such-block"), None);
        // No store → None.
        assert_eq!(resolve_global_output_zone(&None, &oid), None);
    }

    // ────────────────────────────────────────────────────────────────
    // extract_agent_events — Phase 1.5 PR 1
    // ────────────────────────────────────────────────────────────────

    use crate::agents::translator::claude::ClaudeTranslator;
    use crate::agents::types::AgentEvent;

    #[test]
    fn extract_agent_events_full_line_translates() {
        let mut t = ClaudeTranslator::new();
        let mut buf: Vec<u8> = Vec::new();
        let line =
            br#"{"type":"stream_event","event":{"type":"content_block_delta","delta":{"type":"text_delta","text":"hello"}}}
"#;
        let events = extract_agent_events(&mut buf, line, &mut t);
        assert_eq!(events.len(), 1);
        match &events[0] {
            AgentEvent::AssistantText { delta } => assert_eq!(delta, "hello"),
            other => panic!("expected AssistantText, got {other:?}"),
        }
        // Buffer drained.
        assert!(buf.is_empty());
    }

    #[test]
    fn extract_agent_events_chunked_line_accumulates() {
        // PTY can deliver a single logical line across multiple read()
        // calls. The buffer must accumulate across calls and only emit
        // once the newline arrives.
        let mut t = ClaudeTranslator::new();
        let mut buf: Vec<u8> = Vec::new();
        // First chunk: prefix of the JSON, no newline.
        let events = extract_agent_events(
            &mut buf,
            br#"{"type":"stream_event","event":{"type":"content_"#,
            &mut t,
        );
        assert!(events.is_empty());
        // Second chunk: rest of the JSON + newline.
        let events = extract_agent_events(
            &mut buf,
            br#"block_delta","delta":{"type":"text_delta","text":"hi"}}}
"#,
            &mut t,
        );
        assert_eq!(events.len(), 1);
        match &events[0] {
            AgentEvent::AssistantText { delta } => assert_eq!(delta, "hi"),
            other => panic!("expected AssistantText, got {other:?}"),
        }
    }

    #[test]
    fn extract_agent_events_drops_non_json_lines() {
        // Interactive pane output (ANSI escapes, prompts, blank lines)
        // must not produce events.
        let mut t = ClaudeTranslator::new();
        let mut buf: Vec<u8> = Vec::new();
        let pty_text = b"\x1b[2K\x1b[0;0H> some prompt\n[m\nplain text\n";
        let events = extract_agent_events(&mut buf, pty_text, &mut t);
        assert!(events.is_empty(), "got unexpected events: {events:?}");
        // Buffer drained — only the trailing line (if any) is retained.
        // Here every chunk had a newline at the end, so buf is empty.
        assert!(buf.is_empty());
    }

    #[test]
    fn extract_agent_events_drops_carriage_returns() {
        // PTYs in cooked mode often emit \r\n. Trimming should accept
        // both.
        let mut t = ClaudeTranslator::new();
        let mut buf: Vec<u8> = Vec::new();
        let line = br#"{"type":"stream_event","event":{"type":"content_block_delta","delta":{"type":"text_delta","text":"crlf"}}}
"#;
        // Replace the \n with \r\n.
        let mut bytes: Vec<u8> = line.to_vec();
        let last = bytes.len() - 1;
        bytes.insert(last, b'\r');
        let events = extract_agent_events(&mut buf, &bytes, &mut t);
        assert_eq!(events.len(), 1);
    }

    #[test]
    fn extract_agent_events_drops_malformed_json() {
        // A line that starts with `{` but isn't valid JSON should be
        // silently dropped.
        let mut t = ClaudeTranslator::new();
        let mut buf: Vec<u8> = Vec::new();
        let events = extract_agent_events(&mut buf, b"{not_valid_json\n", &mut t);
        assert!(events.is_empty());
    }

    #[test]
    fn extract_agent_events_resets_oversized_buffer() {
        // A producer that never emits newlines for >1 MiB triggers
        // the buffer reset so memory stays bounded.
        let mut t = ClaudeTranslator::new();
        let mut buf: Vec<u8> = Vec::new();
        let chunk = vec![b'x'; AGENT_LINE_BUFFER_CAP + 1];
        let events = extract_agent_events(&mut buf, &chunk, &mut t);
        assert!(events.is_empty());
        assert!(
            buf.is_empty(),
            "buffer should reset past cap, was {} bytes",
            buf.len()
        );
    }

    #[test]
    fn extract_agent_events_preserves_utf8_across_read_boundary() {
        // Reagent P1 / Codex P2 on PR #833: a multi-byte UTF-8
        // character split across two PTY reads must decode cleanly
        // once the complete line arrives, not lossy-decode each
        // half into U+FFFD.
        //
        // 'こんにちは' = e3 81 93 / e3 82 93 / e3 81 ab / e3 81 a1 / e3 81 af
        // Each char is 3 bytes. Split a line mid-character to verify
        // the buffered-bytes path preserves the codepoint.
        let mut t = ClaudeTranslator::new();
        let mut buf: Vec<u8> = Vec::new();
        let frame = r#"{"type":"stream_event","event":{"type":"content_block_delta","delta":{"type":"text_delta","text":"こんにちは"}}}
"#;
        let frame_bytes = frame.as_bytes();
        // Split at byte 60 — somewhere inside one of the multi-byte
        // sequences of こんにちは (which starts around byte 75 in the
        // JSON). Use a position that's clearly inside the codepoint
        // run.
        let split = frame_bytes
            .iter()
            .position(|&b| b == 0xe3)
            .expect("expected to find a multi-byte codepoint")
            + 1; // split right after the leading byte (mid-codepoint)
        let (a, b) = frame_bytes.split_at(split);
        let events = extract_agent_events(&mut buf, a, &mut t);
        // First chunk had no newline.
        assert!(events.is_empty());
        let events = extract_agent_events(&mut buf, b, &mut t);
        assert_eq!(events.len(), 1);
        match &events[0] {
            AgentEvent::AssistantText { delta } => {
                assert_eq!(
                    delta, "こんにちは",
                    "UTF-8 must round-trip cleanly across read boundary; got {delta:?}"
                );
                assert!(!delta.contains('\u{FFFD}'), "no replacement chars");
            }
            other => panic!("expected AssistantText, got {other:?}"),
        }
    }

    #[test]
    fn extract_agent_events_two_lines_one_chunk() {
        // PTY can also deliver multiple complete lines in a single
        // read(). Verify they're all processed.
        let mut t = ClaudeTranslator::new();
        let mut buf: Vec<u8> = Vec::new();
        let two_lines = br#"{"type":"stream_event","event":{"type":"content_block_delta","delta":{"type":"text_delta","text":"a"}}}
{"type":"stream_event","event":{"type":"content_block_delta","delta":{"type":"text_delta","text":"b"}}}
"#;
        let events = extract_agent_events(&mut buf, two_lines, &mut t);
        assert_eq!(events.len(), 2);
    }
}