superdoc_sdk/runtime/

process.rs

use super::{framing::read_frame, policy};
use crate::{Error, Result};
use serde_json::{json, Value};
use std::{
    collections::{BTreeMap, BTreeSet, HashMap},
    path::{Path, PathBuf},
    process::Stdio,
    sync::{
        atomic::{AtomicBool, AtomicU8, Ordering},
        Arc, Mutex, OnceLock,
    },
    time::Duration,
};
use tokio::{
    io::{AsyncReadExt, AsyncWriteExt, BufReader},
    process::{Child, ChildStdin, Command},
    sync::{broadcast, mpsc, oneshot, watch, Notify, OwnedSemaphorePermit, Semaphore},
    time::{timeout, Instant},
};

const MAX_REQUEST_TIMEOUT: Duration = Duration::from_millis(2_147_483_647);

#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum HostMode {
    #[default]
    Cli,
    Document,
}

/// An executable is always spawned directly. Prefix arguments support explicit
/// development wrappers without shell quoting or command interpolation.
#[derive(Clone)]
pub struct HostOptions {
    pub executable: PathBuf,
    pub prefix_args: Vec<String>,
    pub mode: HostMode,
    pub env: BTreeMap<String, String>,
    pub startup_timeout: Duration,
    pub request_timeout: Duration,
    pub shutdown_timeout: Duration,
    pub max_pending: usize,
    pub max_frame_bytes: usize,
    pub state_directory: Option<PathBuf>,
}

impl HostOptions {
    pub fn new(executable: impl Into<PathBuf>) -> Self {
        let default_ms = policy()["hostDefaultRequestTimeoutMs"]
            .as_u64()
            .unwrap_or(30_000);
        Self {
            executable: executable.into(),
            prefix_args: Vec::new(),
            mode: HostMode::Cli,
            env: BTreeMap::new(),
            startup_timeout: Duration::from_secs(10),
            request_timeout: Duration::from_millis(default_ms),
            shutdown_timeout: Duration::from_secs(5),
            max_pending: 64,
            max_frame_bytes: 64 * 1024 * 1024,
            state_directory: None,
        }
    }
}

impl std::fmt::Debug for HostOptions {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("HostOptions")
            .field("mode", &self.mode)
            .field("max_pending", &self.max_pending)
            .field("max_frame_bytes", &self.max_frame_bytes)
            .finish_non_exhaustive()
    }
}

#[derive(Default)]
pub(crate) struct CallState {
    // 0: queued, 1: committed to the writer, 2: cancelled before transmission.
    // The compare-exchange is the linearization point for cancellation.
    phase: AtomicU8,
    written: AtomicBool,
    session_revision_changed: AtomicU8,
    asset: Mutex<Option<tempfile::TempPath>>,
}
impl CallState {
    pub(crate) fn session_revision_changed(&self) -> Option<bool> {
        match self.session_revision_changed.load(Ordering::Acquire) {
            1 => Some(false),
            2 => Some(true),
            _ => None,
        }
    }
    pub(crate) fn transmission_started(&self) -> bool {
        self.phase.load(Ordering::Acquire) == 1
    }
    pub(crate) fn with_asset(asset: tempfile::TempPath) -> Arc<Self> {
        Arc::new(Self {
            asset: Mutex::new(Some(asset)),
            ..Self::default()
        })
    }
    pub(crate) fn cancel_before_send(&self) -> bool {
        self.phase
            .compare_exchange(0, 2, Ordering::AcqRel, Ordering::Acquire)
            != Err(1)
    }
}

struct CancelGuard {
    state: Arc<CallState>,
    retire: Option<Arc<Notify>>,
    disarmed: bool,
}
impl Drop for CancelGuard {
    fn drop(&mut self) {
        if !self.disarmed
            && self
                .state
                .phase
                .compare_exchange(0, 2, Ordering::AcqRel, Ordering::Acquire)
                == Err(1)
        {
            if let Some(stop) = &self.retire {
                stop.notify_one();
            }
        }
    }
}

type Sessions = Arc<Mutex<BTreeMap<String, Arc<AtomicU8>>>>;

struct Request {
    method: String,
    params: Value,
    request_timeout: Option<Duration>,
    watchdog: Duration,
    mutates: bool,
    abandon_open: bool,
    state: Arc<CallState>,
    reply: oneshot::Sender<Result<Value>>,
    permit: OwnedSemaphorePermit,
}

struct Pending {
    reply: oneshot::Sender<Result<Value>>,
    state: Arc<CallState>,
    deadline: Instant,
    mutates: bool,
    abandon_open: bool,
    _permit: OwnedSemaphorePermit,
}

struct Write {
    id: u64,
    frame: Vec<u8>,
    state: Arc<CallState>,
}

enum Event {
    Frame(Value),
    Failed(Error),
    Skipped(u64),
}

enum Control {
    Request(Request),
    Shutdown,
}

struct Connection {
    tx: mpsc::Sender<Control>,
    permits: Arc<Semaphore>,
    closed: Arc<AtomicBool>,
    stop: Arc<Notify>,
    features: OnceLock<BTreeSet<String>>,
    notifications: Mutex<broadcast::Receiver<Value>>,
    state_directory: PathBuf,
    options: HostOptions,
    termination: watch::Receiver<Option<Result<()>>>,
    sessions: Sessions,
    shutdown_confirmation: Arc<Mutex<Option<Result<()>>>>,
    retain_directory: Arc<AtomicU8>,
}

impl Drop for Connection {
    fn drop(&mut self) {
        self.stop.notify_one();
    }
}

#[derive(Clone)]
pub(crate) struct Transport(Arc<Connection>);

fn blank_state_directory(path: &Path) -> bool {
    path.as_os_str()
        .to_string_lossy()
        .chars()
        .all(super::is_js_whitespace)
}

impl Transport {
    pub async fn connect(options: HostOptions) -> Result<Self> {
        if options.executable.as_os_str().is_empty()
            || options
                .state_directory
                .as_deref()
                .is_some_and(blank_state_directory)
            || options.max_pending == 0
            || options.max_frame_bytes == 0
            || options.startup_timeout.is_zero()
            || options.request_timeout.as_millis() == 0
            || options.request_timeout > MAX_REQUEST_TIMEOUT
            || options.shutdown_timeout.is_zero()
            || options.max_pending > 1024
            || [options.startup_timeout, options.shutdown_timeout]
                .iter()
                .any(|duration| duration.as_millis() == 0 || *duration > Duration::from_secs(86400))
        {
            return Err(Error::new(
                "INVALID_ARGUMENT",
                "host paths, limits and timeouts must be non-empty and positive",
            ));
        }
        let (directory, owned) = match &options.state_directory {
            Some(path) => {
                std::fs::create_dir_all(path).map_err(|_| {
                    Error::new(
                        "STATE_DIRECTORY_ERROR",
                        "could not prepare the configured state directory",
                    )
                })?;
                (path.clone(), false)
            }
            None => (
                tempfile::Builder::new()
                    .prefix("superdoc-rust-")
                    .tempdir()
                    .map_err(|_| {
                        Error::new(
                            "STATE_DIRECTORY_ERROR",
                            "could not create an SDK state directory",
                        )
                    })?
                    .keep(),
                true,
            ),
        };
        let mut command = Command::new(&options.executable);
        command
            .args(&options.prefix_args)
            .envs(&options.env)
            .env("SUPERDOC_CLI_STATE_DIR", &directory)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true);
        if options.mode == HostMode::Cli {
            // A process-wide explicit timeout disables native save/replace.
            // Negotiate per-request deadlines for eligible methods instead.
            command.args(["host", "--stdio"]);
        }
        let mut child = match command.spawn() {
            Ok(child) => child,
            Err(_) => {
                if owned {
                    let _ = std::fs::remove_dir_all(&directory);
                }
                return Err(Error::new(
                    "HOST_SPAWN_FAILED",
                    "could not start the configured SuperDoc host",
                ));
            }
        };
        let stdin = child.stdin.take().expect("piped stdin");
        let stdout = child.stdout.take().expect("piped stdout");
        let stderr = child.stderr.take().expect("piped stderr");
        let (tx, rx) = mpsc::channel(options.max_pending);
        let (write_tx, write_rx) = mpsc::channel(options.max_pending);
        let (event_tx, event_rx) = mpsc::channel(2);
        let (notifications, notification_receiver) = broadcast::channel(64);
        let (termination_tx, termination) = watch::channel(None);
        let stop = Arc::new(Notify::new());
        let closed = Arc::new(AtomicBool::new(false));
        let sessions = Arc::new(Mutex::new(BTreeMap::new()));
        let shutdown_confirmation = Arc::new(Mutex::new(None));
        let retain_directory = Arc::new(AtomicU8::new(0));
        let reader_events = event_tx.clone();
        let limit = options.max_frame_bytes;
        let reader = tokio::spawn(async move {
            let mut input = BufReader::new(stdout);
            loop {
                let event = match read_frame(&mut input, limit).await {
                    Ok(Some(frame)) => Event::Frame(frame),
                    Ok(None) => {
                        Event::Failed(Error::new("HOST_DISCONNECTED", "host stdout closed"))
                    }
                    Err(error) => Event::Failed(error),
                };
                let failed = matches!(event, Event::Failed(_));
                if reader_events.send(event).await.is_err() || failed {
                    break;
                }
            }
        });
        let writer = tokio::spawn(write_frames(stdin, write_rx, event_tx));
        let stderr_reader = tokio::spawn(async move {
            let mut stderr = stderr;
            let mut bytes = [0u8; 4096];
            // Drain without logging arbitrary document data or authentication.
            while let Ok(count) = stderr.read(&mut bytes).await {
                if count == 0 {
                    break;
                }
            }
        });
        let transport = Self(Arc::new(Connection {
            tx,
            permits: Arc::new(Semaphore::new(options.max_pending)),
            closed: closed.clone(),
            stop: stop.clone(),
            features: OnceLock::new(),
            notifications: Mutex::new(notification_receiver),
            state_directory: directory.clone(),
            options: options.clone(),
            termination,
            sessions: sessions.clone(),
            shutdown_confirmation: shutdown_confirmation.clone(),
            retain_directory: retain_directory.clone(),
        }));
        tokio::spawn(run_actor(
            child,
            rx,
            write_tx,
            event_rx,
            stop,
            closed,
            notifications,
            directory,
            owned,
            options.shutdown_timeout,
            reader,
            writer,
            stderr_reader,
            termination_tx,
            sessions,
            shutdown_confirmation,
            retain_directory,
        ));
        let capabilities = transport
            .request(
                "host.capabilities",
                json!({}),
                None,
                options.startup_timeout,
                false,
                false,
            )
            .await;
        let features = capabilities.and_then(|value| validate_capabilities(&value, options.mode));
        match features {
            Ok(features) => {
                let _ = transport.0.features.set(features);
                Ok(transport)
            }
            Err(error) => {
                let _ = transport.shutdown().await;
                Err(Error {
                    code: "HOST_HANDSHAKE_FAILED".to_owned(),
                    ..error
                })
            }
        }
    }

    pub fn features(&self) -> &BTreeSet<String> {
        self.0.features.get().expect("completed handshake")
    }
    pub fn options(&self) -> &HostOptions {
        &self.0.options
    }
    pub fn is_closed(&self) -> bool {
        self.0.closed.load(Ordering::Acquire)
    }
    pub fn state_directory(&self) -> &Path {
        &self.0.state_directory
    }
    pub fn subscribe(&self) -> broadcast::Receiver<Value> {
        self.0
            .notifications
            .lock()
            .expect("SDK notification lock")
            .resubscribe()
    }
    pub fn session_state(&self, id: &str) -> Arc<AtomicU8> {
        self.0
            .sessions
            .lock()
            .expect("SDK session lock")
            .entry(id.to_owned())
            .or_insert_with(|| Arc::new(AtomicU8::new(0)))
            .clone()
    }
    pub fn has_session(&self, id: &str) -> bool {
        self.0
            .sessions
            .lock()
            .expect("SDK session lock")
            .get(id)
            .is_some_and(|state| state.load(Ordering::Acquire) < 2)
    }
    pub fn remove_session(&self, id: &str, state: &Arc<AtomicU8>) {
        let mut sessions = self.0.sessions.lock().expect("SDK session lock");
        if sessions
            .get(id)
            .is_some_and(|current| Arc::ptr_eq(current, state))
        {
            sessions.remove(id);
        }
    }

    pub async fn request(
        &self,
        method: &str,
        params: Value,
        request_timeout: Option<Duration>,
        watchdog: Duration,
        mutates: bool,
        abandon_open: bool,
    ) -> Result<Value> {
        self.request_tracked(
            method,
            params,
            request_timeout,
            watchdog,
            mutates,
            abandon_open,
            Arc::new(CallState::default()),
        )
        .await
    }

    #[allow(clippy::too_many_arguments)]
    pub(crate) async fn request_tracked(
        &self,
        method: &str,
        params: Value,
        request_timeout: Option<Duration>,
        watchdog: Duration,
        mutates: bool,
        abandon_open: bool,
        state: Arc<CallState>,
    ) -> Result<Value> {
        if request_timeout
            .is_some_and(|duration| duration.as_millis() == 0 || duration > MAX_REQUEST_TIMEOUT)
        {
            return Err(Error::new(
                "INVALID_ARGUMENT",
                "request timeouts must be between one and 2147483647 milliseconds",
            ));
        }
        let headroom =
            Duration::from_millis(policy()["watchdogHeadroomMs"].as_u64().unwrap_or(5000));
        if watchdog.as_millis() == 0 || watchdog > MAX_REQUEST_TIMEOUT.saturating_add(headroom) {
            return Err(Error::new(
                "INVALID_ARGUMENT",
                "host response watchdog is outside the supported timeout range",
            ));
        }
        if self.is_closed() {
            return Err(Error::new("HOST_DISCONNECTED", "host connection is closed"));
        }
        let permit = self
            .0
            .permits
            .clone()
            .try_acquire_owned()
            .map_err(|_| Error::new("HOST_QUEUE_FULL", "host request capacity is exhausted"))?;
        let (reply, result) = oneshot::channel();
        let mut guard = CancelGuard {
            state: state.clone(),
            retire: abandon_open.then(|| self.0.stop.clone()),
            disarmed: false,
        };
        self.0
            .tx
            .send(Control::Request(Request {
                method: method.to_owned(),
                params,
                request_timeout,
                watchdog,
                mutates,
                abandon_open,
                state,
                reply,
                permit,
            }))
            .await
            .map_err(|_| Error::new("HOST_DISCONNECTED", "host connection is closed"))?;
        let result = result.await.unwrap_or_else(|_| {
            Err(Error::new(
                "HOST_DISCONNECTED",
                "host connection stopped before replying",
            ))
        });
        guard.disarmed = true;
        result
    }

    pub async fn shutdown(&self) -> Result<()> {
        if !self.is_closed() {
            let _ = self.0.tx.send(Control::Shutdown).await;
        }
        self.wait_termination().await
    }

    async fn wait_termination(&self) -> Result<()> {
        let mut termination = self.0.termination.clone();
        loop {
            if let Some(result) = termination.borrow().clone() {
                return result;
            }
            termination.changed().await.map_err(|_| {
                Error::new(
                    "HOST_SHUTDOWN_FAILED",
                    "host cleanup task stopped before confirming process exit",
                )
            })?;
        }
    }

    pub(crate) async fn drain(&self) -> Result<()> {
        let permits = self
            .0
            .permits
            .clone()
            .acquire_many_owned(self.0.options.max_pending as u32)
            .await
            .map_err(|_| Error::new("HOST_DISCONNECTED", "host request admission closed"))?;
        drop(permits);
        Ok(())
    }

    pub(crate) fn retain_state(&self) {
        self.0.retain_directory.fetch_max(1, Ordering::AcqRel);
    }

    pub(crate) async fn retire_preserving_state(&self) -> Result<()> {
        self.0.retain_directory.store(2, Ordering::Release);
        self.0.stop.notify_one();
        self.wait_termination().await
    }

    pub(crate) async fn shutdown_checked(&self) -> Result<()> {
        self.retain_state();
        self.shutdown().await?;
        self.0
            .shutdown_confirmation
            .lock()
            .expect("SDK shutdown confirmation lock")
            .clone()
            .unwrap_or_else(|| {
                Err(Error::new(
                    "HOST_SHUTDOWN_FAILED",
                    "host shutdown was not confirmed",
                ))
            })
    }
}

fn validate_capabilities(value: &Value, mode: HostMode) -> Result<BTreeSet<String>> {
    if value["protocolVersion"] != policy()["protocolVersion"] {
        let mut error = Error::new(
            "HOST_HANDSHAKE_FAILED",
            "host protocol version is incompatible",
        );
        error.details = Some(Box::new(json!({
            "expected": policy()["protocolVersion"],
            "actual": value["protocolVersion"]
        })));
        return Err(error);
    }
    let values = value["features"]
        .as_array()
        .ok_or_else(|| Error::new("HOST_HANDSHAKE_FAILED", "host capabilities were malformed"))?;
    let features: Option<BTreeSet<_>> = values
        .iter()
        .map(|value| value.as_str().map(str::to_owned))
        .collect();
    let features = features
        .ok_or_else(|| Error::new("HOST_HANDSHAKE_FAILED", "host features must be strings"))?;
    let required = if mode == HostMode::Cli {
        "cliRequiredFeatures"
    } else {
        "documentRequiredFeatures"
    };
    let expected = policy()[required]
        .as_array()
        .ok_or_else(|| Error::new("HOST_PROTOCOL_ERROR", "generated host policy is missing"))?;
    if expected
        .iter()
        .any(|feature| !features.contains(feature.as_str().unwrap_or("")))
    {
        return Err(Error::new(
            "HOST_HANDSHAKE_FAILED",
            "host lacks a required SDK capability",
        ));
    }
    Ok(features)
}

async fn write_frames(
    mut stdin: ChildStdin,
    mut commands: mpsc::Receiver<Write>,
    events: mpsc::Sender<Event>,
) {
    while let Some(command) = commands.recv().await {
        if command
            .state
            .phase
            .compare_exchange(0, 1, Ordering::AcqRel, Ordering::Acquire)
            .is_err()
        {
            if events.send(Event::Skipped(command.id)).await.is_err() {
                break;
            }
            continue;
        }
        if stdin.write_all(&command.frame).await.is_err() {
            let _ = events
                .send(Event::Failed(Error::new(
                    "HOST_DISCONNECTED",
                    "could not write a complete host request",
                )))
                .await;
            break;
        }
        command.state.written.store(true, Ordering::Release);
    }
}

#[allow(clippy::too_many_arguments)]
async fn run_actor(
    mut child: Child,
    mut commands: mpsc::Receiver<Control>,
    writes: mpsc::Sender<Write>,
    mut events: mpsc::Receiver<Event>,
    stop: Arc<Notify>,
    closed: Arc<AtomicBool>,
    notifications: broadcast::Sender<Value>,
    directory: PathBuf,
    owned: bool,
    shutdown_timeout: Duration,
    reader: tokio::task::JoinHandle<()>,
    writer: tokio::task::JoinHandle<()>,
    stderr: tokio::task::JoinHandle<()>,
    termination: watch::Sender<Option<Result<()>>>,
    sessions: Sessions,
    shutdown_confirmation: Arc<Mutex<Option<Result<()>>>>,
    retain_directory: Arc<AtomicU8>,
) {
    let mut pending: HashMap<u64, Pending> = HashMap::new();
    let mut next_id = 1u64;
    let mut shutdown_id = None;
    let mut shutdown_deadline = None;
    let mut shutdown_response = None;
    let mut child_exited = false;
    let mut failure = Error::new("HOST_DISCONNECTED", "host connection stopped");
    loop {
        let next_deadline = pending
            .values()
            .map(|request| request.deadline)
            .chain(shutdown_deadline)
            .min()
            .unwrap_or_else(|| Instant::now() + Duration::from_secs(86400));
        tokio::select! {
            _=stop.notified()=>break,
            _=child.wait(),if !child_exited=>{child_exited=true;if shutdown_id.is_none(){break;}},
            _=tokio::time::sleep_until(next_deadline)=>{
                if shutdown_deadline.is_some_and(|deadline|deadline<=Instant::now()){failure=Error::new("TIMEOUT","host shutdown deadline expired");break;}
                let expired:Vec<_>=pending.iter().filter_map(|(id,request)|(request.deadline<=Instant::now()).then_some(*id)).collect();
                let mut sent_timeout=false;
                for id in expired {
                    let request=pending.get(&id).expect("pending deadline");
                    if request.state.phase.compare_exchange(0,2,Ordering::AcqRel,Ordering::Acquire).is_err()
                        && request.state.phase.load(Ordering::Acquire)==1 {sent_timeout=true;break;}
                    if let Some(request)=pending.remove(&id) {let _=request.reply.send(Err(Error::new("TIMEOUT","request expired before transmission")));}
                }
                if sent_timeout {failure=Error::new("TIMEOUT","host response watchdog expired");break;}
            }
            control=commands.recv(),if shutdown_id.is_none()=>{
                match control {
                    None=>break,
                    Some(Control::Request(request))=>{
                        if request.state.phase.load(Ordering::Acquire)==2 {continue;}
                        let id=next_id;next_id+=1;
                        let mut frame=json!({"jsonrpc":"2.0","id":id,"method":request.method,"params":request.params});
                        if let Some(value)=request.request_timeout {frame["requestTimeoutMs"]=json!(value.as_millis() as u64);}
                        let mut bytes=serde_json::to_vec(&frame).expect("serializable JSON value");bytes.push(b'\n');
                        if bytes.len()>64*1024*1024 {let _=request.reply.send(Err(Error::new("PAYLOAD_TOO_LARGE","host request exceeded the byte limit")));continue;}
                        let state=request.state.clone();
                        pending.insert(id,Pending{reply:request.reply,state:request.state,deadline:Instant::now()+request.watchdog,mutates:request.mutates,abandon_open:request.abandon_open,_permit:request.permit});
                        match writes.try_send(Write{id,frame:bytes,state}) {
                            Ok(())=>{},
                            Err(mpsc::error::TrySendError::Full(_))=>{
                                if let Some(request)=pending.remove(&id) {let _=request.state.phase.compare_exchange(0,2,Ordering::AcqRel,Ordering::Acquire);let _=request.reply.send(Err(Error::new("HOST_QUEUE_FULL","host writer capacity is exhausted")));}
                            }
                            Err(mpsc::error::TrySendError::Closed(_))=>{failure=Error::new("HOST_DISCONNECTED","host writer stopped");break;}
                        }
                    }
                    Some(Control::Shutdown)=>{
                        closed.store(true,Ordering::Release);
                        let id=next_id;next_id+=1;shutdown_id=Some(id);shutdown_deadline=Some(Instant::now()+shutdown_timeout);
                        let mut frame=serde_json::to_vec(&json!({"jsonrpc":"2.0","id":id,"method":"host.shutdown","params":{}})).expect("shutdown frame");frame.push(b'\n');
                        if writes.try_send(Write{id,frame,state:Arc::new(CallState::default())}).is_err(){break;}
                    }
                }
            }
            event=events.recv()=>{
                match event {
                    None=>break,
                    Some(Event::Failed(error))=>{failure=error;break;},
                    Some(Event::Skipped(id))=>{pending.remove(&id);},
                    Some(Event::Frame(frame))=>{
                        if frame["jsonrpc"]!="2.0" {failure=Error::new("HOST_PROTOCOL_ERROR","host sent an invalid JSON-RPC version");break;}
                        if frame.get("id").is_none() && frame["method"].is_string(){
                            if frame["method"]=="event.sessionClosed" {
                                if let Some(id)=frame["params"]["sessionId"].as_str(){if let Some(state)=sessions.lock().expect("SDK session lock").get(id){state.store(2,Ordering::Release);}}
                            }
                            let _=notifications.send(frame);continue;
                        }
                        let Some(id)=frame["id"].as_u64() else {failure=Error::new("HOST_PROTOCOL_ERROR","host response ID was invalid");break;};
                        if Some(id)==shutdown_id {shutdown_response=Some(decode_response(frame).map(|_|()));break;}
                        let Some(request)=pending.remove(&id) else {failure=Error::new("HOST_PROTOCOL_ERROR","host returned an unknown or duplicate response ID");break;};
                        if let Some(changed) = frame.get("result").and_then(|result| result.get("sessionRevisionChanged"))
                            .or_else(|| frame.get("error").and_then(|error| error.get("data")).and_then(|data| data.get("sessionRevisionChanged")))
                            .and_then(Value::as_bool) {
                            request.state.session_revision_changed.store(if changed { 2 } else { 1 }, Ordering::Release);
                        }
                        let result=decode_response(frame);
                        if request.abandon_open {
                            if let Ok(value)=&result {
                                if let Some(id)=value["data"]["contextId"].as_str().or_else(||value["sessionId"].as_str()) {
                                    let mut sessions=sessions.lock().expect("SDK session lock");
                                    if sessions.get(id).is_none_or(|state|state.load(Ordering::Acquire)>=2){sessions.insert(id.to_owned(),Arc::new(AtomicU8::new(0)));}
                                }
                            }
                        }
                        if request.abandon_open && request.reply.is_closed() && request.state.phase.load(Ordering::Acquire)==1 {
                            failure=Error::new("HOST_DISCONNECTED","a cancelled open retired the host to avoid leaking a session");break;
                        }
                        let _=request.reply.send(result.map_err(|mut error|{error.outcome_unknown=request.mutates && ["TIMEOUT","HOST_PROTOCOL_ERROR"].contains(&error.code.as_str());error}));
                    }
                }
            }
        }
    }
    closed.store(true, Ordering::Release);
    for state in sessions.lock().expect("SDK session lock").values() {
        if state.load(Ordering::Acquire) < 2 {
            state.store(3, Ordering::Release);
        }
    }
    drop(writes);
    if shutdown_id.is_none() || failure.code == "TIMEOUT" {
        let _ = child.start_kill();
    }
    let (reaped, clean_exit, exit_code) = match timeout(shutdown_timeout, child.wait()).await {
        Ok(Ok(status)) => (true, status.success(), status.code()),
        _ => {
            let _ = child.start_kill();
            match timeout(shutdown_timeout, child.wait()).await {
                Ok(Ok(status)) => (true, status.success(), status.code()),
                _ => (false, false, None),
            }
        }
    };
    reader.abort();
    writer.abort();
    stderr.abort();
    let _ = reader.await;
    let _ = writer.await;
    let _ = stderr.await;
    for (_, request) in pending {
        if !reaped {
            if let Some(asset) = request.state.asset.lock().expect("SDK asset lock").take() {
                let _ = asset.keep();
            }
        }
        let mut error = failure.clone();
        error.outcome_unknown = request.mutates && request.state.phase.load(Ordering::Acquire) == 1;
        let _ = request.reply.send(Err(error));
    }
    // Requests accepted into the bounded channel but not committed to the
    // writer have a known unexecuted outcome.
    while let Ok(control) = commands.try_recv() {
        match control {
            Control::Request(request) => {
                let _ = request.reply.send(Err(failure.clone()));
            }
            Control::Shutdown => {}
        }
    }
    let mut confirmation = match shutdown_response {
        Some(Err(error)) => Err(error),
        Some(Ok(())) if reaped && clean_exit => Ok(()),
        _ => Err(Error::new(
            "HOST_SHUTDOWN_FAILED",
            "host shutdown response and clean exit were not confirmed",
        )),
    };
    if let Err(error) = &mut confirmation {
        error.exit_code = exit_code.map(|value| Box::new(i64::from(value)));
    }
    if confirmation.is_ok() {
        let _ = retain_directory.compare_exchange(1, 0, Ordering::AcqRel, Ordering::Acquire);
    }
    *shutdown_confirmation
        .lock()
        .expect("SDK shutdown confirmation lock") = Some(confirmation);
    if owned && reaped && retain_directory.load(Ordering::Acquire) == 0 {
        let _ = std::fs::remove_dir_all(&directory);
    }
    termination.send_replace(Some(if reaped {
        Ok(())
    } else {
        Err(Error::new(
            "HOST_SHUTDOWN_FAILED",
            "host process exit could not be confirmed; state was retained",
        ))
    }));
}

fn decode_response(frame: Value) -> Result<Value> {
    if frame.get("result").is_some() == frame.get("error").is_some() {
        return Err(Error::new(
            "HOST_PROTOCOL_ERROR",
            "host response must contain exactly one result or error",
        ));
    }
    if let Some(error) = frame.get("error") {
        let code = error["code"]
            .as_i64()
            .ok_or_else(|| Error::new("HOST_PROTOCOL_ERROR", "host error code was invalid"))?;
        if !error["message"].is_string() {
            return Err(Error::new(
                "HOST_PROTOCOL_ERROR",
                "host error message was invalid",
            ));
        }
        let data = &error["data"];
        let mut result = Error::new("COMMAND_FAILED", "SuperDoc host rejected the operation");
        result.rpc_code = Some(code);
        result.host_message = error["message"].as_str().map(Into::into);
        if let Some(cli_code) = data["cliCode"].as_str() {
            result.code = cli_code.to_owned();
            result.details = data.get("details").cloned().map(Box::new);
            result.exit_code = data["exitCode"].as_i64().map(Box::new);
            result.host_message = data["message"]
                .as_str()
                .or_else(|| error["message"].as_str())
                .map(Into::into);
        } else if let Some(domain_code) = data["domainCode"].as_str() {
            result.code = domain_code.to_owned();
            let mut details = serde_json::Map::new();
            for (source, target) in [
                ("details", "domainDetails"),
                ("stage", "stage"),
                ("byteLength", "byteLength"),
            ] {
                if let Some(value) = data.get(source) {
                    details.insert(target.to_owned(), value.clone());
                }
            }
            result.details = Some(Box::new(Value::Object(details)));
        } else {
            if code == -32011 {
                result.code = "TIMEOUT".to_owned();
            }
            result.details = error.get("data").cloned().map(Box::new);
        }
        Err(result)
    } else {
        Ok(frame["result"].clone())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn blank_state_directories_match_node_path_validation() {
        for path in ["", " ", "\t\n", "\u{00a0}\u{2003}", "\u{feff}"] {
            assert!(blank_state_directory(Path::new(path)));
        }
        for path in ["\u{0085}", " state ", "a\u{feff}", "."] {
            assert!(!blank_state_directory(Path::new(path)));
        }
    }

    fn host(mode: &str, log: &Path) -> HostOptions {
        let mut options = HostOptions::new(
            std::env::var("SUPERDOC_SDK_TEST_NODE").unwrap_or_else(|_| "node".to_owned()),
        );
        let file = std::env::var_os("SUPERDOC_SDK_TEST_HOST")
            .map(PathBuf::from)
            .unwrap_or_else(|| {
                PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/host.mjs")
            });
        options.prefix_args.push(file.to_str().unwrap().to_owned());
        options
            .env
            .insert("SUPERDOC_RUST_TEST_MODE".to_owned(), mode.to_owned());
        options.env.insert(
            "SUPERDOC_RUST_TEST_LOG".to_owned(),
            log.to_str().unwrap().to_owned(),
        );
        options.shutdown_timeout = Duration::from_millis(250);
        options
    }

    async fn wait_for(mut condition: impl FnMut() -> bool) {
        timeout(Duration::from_secs(5), async {
            while !condition() {
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
        })
        .await
        .expect("controlled host barrier was not reached");
    }

    #[tokio::test]
    async fn correlates_out_of_order_replies() {
        let directory = tempfile::tempdir().unwrap();
        let transport = Transport::connect(host("reorder", &directory.path().join("requests")))
            .await
            .unwrap();
        let a = transport.request(
            "cli.invoke",
            json!({"argv":["first"]}),
            None,
            Duration::from_secs(5),
            false,
            false,
        );
        let b = transport.request(
            "cli.invoke",
            json!({"argv":["second"]}),
            None,
            Duration::from_secs(5),
            false,
            false,
        );
        let (a, b) = tokio::join!(a, b);
        assert_eq!(a.unwrap()["data"]["marker"], "first");
        assert_eq!(b.unwrap()["data"]["marker"], "second");
        transport.shutdown().await.unwrap();
    }

    #[tokio::test]
    async fn watchdog_and_disconnection_preserve_uncertain_sent_mutation() {
        for mode in ["hang", "exit", "bad-frame", "unknown-id"] {
            let directory = tempfile::tempdir().unwrap();
            let log = directory.path().join("requests");
            let transport = Transport::connect(host(mode, &log)).await.unwrap();
            let error = transport
                .request(
                    "cli.invoke",
                    json!({"argv":["mutation"]}),
                    None,
                    Duration::from_secs(1),
                    true,
                    false,
                )
                .await
                .unwrap_err();
            assert!(
                error.outcome_unknown,
                "{mode} must not claim a sent mutation was undone"
            );
            assert!(matches!(
                error.code.as_str(),
                "TIMEOUT" | "HOST_DISCONNECTED" | "HOST_PROTOCOL_ERROR"
            ));
            transport.shutdown().await.unwrap();
            assert_eq!(
                std::fs::read_to_string(log)
                    .unwrap()
                    .matches("mutation")
                    .count(),
                1,
                "a sent mutation must never be replayed"
            );
        }
    }

    #[tokio::test]
    async fn cancelled_sent_open_retires_host() {
        let directory = tempfile::tempdir().unwrap();
        let log = directory.path().join("requests");
        let transport = Transport::connect(host("hang", &log)).await.unwrap();
        let state_directory = transport.state_directory().to_path_buf();
        let copy = transport.clone();
        let task = tokio::spawn(async move {
            copy.request(
                "cli.invoke",
                json!({"argv":["open"]}),
                None,
                Duration::from_secs(30),
                true,
                true,
            )
            .await
        });
        wait_for(|| std::fs::read_to_string(&log).is_ok_and(|log| log.contains("cli.invoke")))
            .await;
        task.abort();
        let _ = task.await;
        transport.shutdown().await.unwrap();
        assert!(!state_directory.exists());
    }

    #[tokio::test]
    async fn queued_cancellation_sends_no_frame_and_releases_capacity() {
        let directory = tempfile::tempdir().unwrap();
        let log = directory.path().join("requests");
        let paused = directory.path().join("paused");
        let resume = directory.path().join("resume");
        let mut options = host("backpressure", &log);
        options.max_pending = 2;
        options.env.insert(
            "SUPERDOC_RUST_TEST_PAUSED".to_owned(),
            paused.to_str().unwrap().to_owned(),
        );
        options.env.insert(
            "SUPERDOC_RUST_TEST_RESUME".to_owned(),
            resume.to_str().unwrap().to_owned(),
        );
        let transport = Transport::connect(options).await.unwrap();
        wait_for(|| paused.exists()).await;
        let first_state = Arc::new(CallState::default());
        let (reply, result) = oneshot::channel();
        let permit = transport.0.permits.clone().try_acquire_owned().unwrap();
        transport
            .0
            .tx
            .send(Control::Request(Request {
                method: "cli.invoke".to_owned(),
                params: json!({"argv":["large"],"padding":"x".repeat(16*1024*1024)}),
                request_timeout: None,
                watchdog: Duration::from_secs(20),
                mutates: false,
                abandon_open: false,
                state: first_state.clone(),
                reply,
                permit,
            }))
            .await
            .unwrap_or_else(|_| panic!("host command channel closed"));
        wait_for(|| first_state.phase.load(Ordering::Acquire) == 1).await;
        assert!(
            !first_state.written.load(Ordering::Acquire),
            "fixture must hold the writer inside the large frame"
        );
        let copy = transport.clone();
        let cancelled = tokio::spawn(async move {
            copy.request(
                "cli.invoke",
                json!({"argv":["must-not-send"]}),
                None,
                Duration::from_secs(10),
                false,
                false,
            )
            .await
        });
        wait_for(|| transport.0.permits.available_permits() == 0).await;
        cancelled.abort();
        let _ = cancelled.await;
        std::fs::write(resume, b"resume").unwrap();
        result.await.unwrap().unwrap();
        wait_for(|| transport.0.permits.available_permits() == 2).await;
        transport.shutdown().await.unwrap();
        assert!(!std::fs::read_to_string(log)
            .unwrap()
            .contains("must-not-send"));
    }

    #[test]
    fn malformed_rpc_envelopes_do_not_become_success() {
        for value in [
            json!({"result":{},"error":{}}),
            json!({}),
            json!({"error":{"code":"wrong","message":"bad"}}),
        ] {
            assert_eq!(
                decode_response(value).unwrap_err().code,
                "HOST_PROTOCOL_ERROR"
            );
        }
    }
}