superdoc-sdk 2.26.0-next.16

use crate::{
    generated::{
        operations::{DocApi, DocApiPreset},
        types::{DocCloseBoundParams, DocCloseResult, DocOpenParams, DocOpenResult},
    },
    invoke::Invoker,
    runtime::{
        contract, policy,
        process::{CallState, HostOptions, Transport},
        routing, trim_js_whitespace,
    },
    Error, InvokeOptions, Result,
};
use base64::{engine::general_purpose::STANDARD as BASE64, Engine};
use serde_json::{json, Value};
use std::{
    collections::BTreeSet,
    path::Path,
    sync::{
        atomic::{AtomicU64, AtomicU8, Ordering},
        Arc, Mutex,
    },
    time::{Duration, SystemTime, UNIX_EPOCH},
};
use tokio::sync::{broadcast, watch, Notify};

#[derive(Clone)]
pub struct CollaborationAuth {
    token: String,
}
impl CollaborationAuth {
    pub fn token(token: impl Into<String>) -> Result<Self> {
        let token = token.into();
        if token.is_empty() {
            return Err(Error::new(
                "INVALID_ARGUMENT",
                "collaboration token must not be empty",
            ));
        }
        Ok(Self { token })
    }
    fn json(&self) -> Value {
        json!({"type":"token","token":self.token})
    }
}
impl std::fmt::Debug for CollaborationAuth {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("CollaborationAuth([redacted])")
    }
}

#[derive(Clone, Debug, Default)]
pub struct OpenOptions {
    pub invoke: InvokeOptions,
    pub collaboration_auth: Option<CollaborationAuth>,
}

#[derive(Clone, Default)]
pub struct ClientOptions {
    pub default_change_mode: Option<String>,
    pub user_name: Option<String>,
    pub user_email: Option<String>,
}

struct ClientInner {
    transport: Transport,
    options: ClientOptions,
    opening: Mutex<BTreeSet<String>>,
    lifecycle: Mutex<ClientLifecycle>,
    idle: Notify,
}

#[derive(Default)]
struct ClientLifecycle {
    active: usize,
    stopping: bool,
    opened_documents: bool,
    shutdown: Option<watch::Receiver<Option<Result<()>>>>,
}

struct ActiveCall(Arc<ClientInner>);

impl Drop for ActiveCall {
    fn drop(&mut self) {
        let mut lifecycle = self.0.lifecycle.lock().expect("SDK lifecycle lock");
        lifecycle.active -= 1;
        if lifecycle.active == 0 {
            self.0.idle.notify_one();
        }
    }
}

#[derive(Clone)]
pub struct Client(Arc<ClientInner>);

struct OpenGuard {
    client: Client,
    id: String,
}
impl Drop for OpenGuard {
    fn drop(&mut self) {
        self.client
            .0
            .opening
            .lock()
            .expect("SDK opening lock")
            .remove(&self.id);
    }
}

impl Client {
    pub async fn connect(host: HostOptions) -> Result<Self> {
        Self::connect_with_options(host, ClientOptions::default()).await
    }

    pub async fn connect_with_options(host: HostOptions, options: ClientOptions) -> Result<Self> {
        if options
            .default_change_mode
            .as_deref()
            .is_some_and(|mode| !["direct", "tracked"].contains(&mode))
        {
            return Err(Error::new(
                "INVALID_ARGUMENT",
                "default change mode must be direct or tracked",
            ));
        }
        if options
            .user_name
            .as_deref()
            .is_some_and(|name| trim_js_whitespace(name).is_empty())
            || (options.user_email.is_some() && options.user_name.is_none())
        {
            return Err(Error::new(
                "INVALID_ARGUMENT",
                "user identity requires a non-empty name",
            ));
        }
        Ok(Self(Arc::new(ClientInner {
            transport: Transport::connect(host).await?,
            options,
            opening: Mutex::new(BTreeSet::new()),
            lifecycle: Mutex::new(ClientLifecycle::default()),
            idle: Notify::new(),
        })))
    }

    /// Local contract introspection requires no running host.
    pub fn describe() -> Value {
        contract()["introspection"]["overview"].clone()
    }

    pub fn describe_command(lookup: &str) -> Result<Value> {
        let index = contract()["introspection"]["lookup"]
            [trim_js_whitespace(lookup).to_lowercase()]
        .as_u64()
        .ok_or_else(|| {
            Error::new(
                "OPERATION_NOT_FOUND",
                "no contract operation matches the requested lookup",
            )
        })?;
        Ok(contract()["introspection"]["overview"]["operations"][index as usize].clone())
    }

    pub fn subscribe(&self) -> Notifications {
        Notifications(self.0.transport.subscribe())
    }
    pub fn state_directory(&self) -> &Path {
        self.0.transport.state_directory()
    }

    pub async fn open(&self, params: &DocOpenParams) -> Result<Document> {
        self.open_with_options(params, OpenOptions::default()).await
    }

    pub async fn open_with_options(
        &self,
        params: &DocOpenParams,
        options: OpenOptions,
    ) -> Result<Document> {
        let _active = self.begin_call()?;
        let mut params = serde_json::to_value(params)
            .map_err(|_| Error::new("INVALID_ARGUMENT", "open parameters could not be encoded"))?;
        let session_id = match params.get("sessionId") {
            Some(value) => value
                .as_str()
                .filter(|id| !id.is_empty())
                .ok_or_else(|| Error::new("INVALID_ARGUMENT", "session ID must be non-empty"))?
                .to_owned(),
            None => fresh_session_id(),
        };
        {
            let mut opening = self.0.opening.lock().expect("SDK opening lock");
            if self.0.transport.has_session(&session_id) || !opening.insert(session_id.clone()) {
                let mut error = Error::new(
                    "SESSION_ALREADY_OPEN",
                    "session ID is already open or opening",
                );
                error.details = Some(Box::new(json!({"sessionId":session_id})));
                return Err(error);
            }
        }
        let _guard = OpenGuard {
            client: self.clone(),
            id: session_id.clone(),
        };
        params["sessionId"] = json!(session_id);

        let result = {
            if let Some(name) = &self.0.options.user_name {
                if params.get("userName").is_none() {
                    params["userName"] = json!(name);
                }
            }
            if let Some(email) = &self.0.options.user_email {
                if params.get("userEmail").is_none() {
                    params["userEmail"] = json!(email);
                }
            }
            self.invoke_cli(
                "doc.open",
                params,
                &options.invoke,
                options.collaboration_auth.as_ref(),
            )
            .await
        };
        let result = match result {
            Ok(value) => serde_json::from_value::<DocOpenResult>(value).map_err(Error::decode),
            Err(mut error) => {
                error.operation_id = Some("doc.open".into());
                if ["TIMEOUT", "HOST_DISCONNECTED", "HOST_PROTOCOL_ERROR"]
                    .contains(&error.code.as_str())
                {
                    let _ = self.0.transport.shutdown().await;
                    error.outcome_unknown = true;
                }
                return Err(error);
            }
        };
        let result = match result {
            Ok(result) if result.context_id == session_id && result.active => result,
            invalid => {
                // An invalid open result cannot leave an unreachable native
                // session alive. Retiring also invalidates any sibling handles.
                let _ = self.0.transport.shutdown().await;
                let mut error = match invalid {
                    Err(error) => error,
                    Ok(_) => Error::new(
                        "HOST_PROTOCOL_ERROR",
                        "host returned an invalid open result",
                    ),
                };
                error.operation_id = Some("doc.open".into());
                error.outcome_unknown = true;
                return Err(error);
            }
        };
        let state = self.0.transport.session_state(&session_id);
        if state.load(Ordering::Acquire) != 0 {
            return Err(Error::new(
                "DOCUMENT_CLOSED",
                "session closed while it was opening",
            ));
        }
        self.0
            .lifecycle
            .lock()
            .expect("SDK lifecycle lock")
            .opened_documents = true;

        Ok(Document(Arc::new(DocumentInner {
            client: self.clone(),
            session_id,
            state,
            open_result: result,
        })))
    }

    pub async fn shutdown(&self) -> Result<()> {
        let mut completion = {
            let mut lifecycle = self.0.lifecycle.lock().expect("SDK lifecycle lock");
            if let Some(completion) = &lifecycle.shutdown {
                completion.clone()
            } else {
                let (result, completion) = watch::channel(None);
                lifecycle.stopping = true;
                lifecycle.shutdown = Some(completion.clone());
                let client = self.clone();
                tokio::spawn(async move {
                    let outcome = client.shutdown_owned().await;
                    if outcome.as_ref().is_err_and(|error| !error.outcome_unknown)
                        && !client.0.transport.is_closed()
                    {
                        let mut lifecycle = client.0.lifecycle.lock().expect("SDK lifecycle lock");
                        lifecycle.stopping = false;
                        lifecycle.shutdown = None;
                    }
                    result.send_replace(Some(outcome));
                });
                completion
            }
        };
        loop {
            if let Some(result) = completion.borrow().clone() {
                return result;
            }
            completion.changed().await.map_err(|_| {
                Error::new(
                    "HOST_SHUTDOWN_FAILED",
                    "managed shutdown owner stopped before confirming completion",
                )
            })?;
        }
    }

    fn begin_call(&self) -> Result<ActiveCall> {
        let mut lifecycle = self.0.lifecycle.lock().expect("SDK lifecycle lock");
        if lifecycle.stopping {
            return Err(Error::new("HOST_DISCONNECTED", "host is disposing"));
        }
        lifecycle.active += 1;
        Ok(ActiveCall(self.0.clone()))
    }

    async fn shutdown_owned(&self) -> Result<()> {
        loop {
            let idle = self.0.idle.notified();
            if self.0.lifecycle.lock().expect("SDK lifecycle lock").active == 0 {
                break;
            }
            idle.await;
        }
        self.0.transport.drain().await?;
        if self.0.transport.is_closed() {
            return self.0.transport.shutdown().await;
        }
        if !self
            .0
            .lifecycle
            .lock()
            .expect("SDK lifecycle lock")
            .opened_documents
        {
            return self.0.transport.shutdown().await;
        }
        self.0.transport.shutdown_checked().await.map_err(|error| {
            let mut failure = Error::new(
                "HOST_SHUTDOWN_FAILED",
                "host shutdown failed; its state directory has been retained",
            );
            failure.outcome_unknown = error.outcome_unknown;
            failure.exit_code = error.exit_code;
            failure.details = Some(Box::new(json!({
                "path": self.state_directory(),
                "exitCode": failure.exit_code.as_deref(),
                "cause": {"code": error.code, "details": error.details}
            })));
            failure
        })
    }

    pub fn presets(&self) -> DocApiPreset<'_, Self> {
        DocApiPreset::new(self)
    }

    fn require_feature(&self, key: &str) -> Result<()> {
        if !self
            .0
            .transport
            .features()
            .contains(policy()["features"][key].as_str().unwrap_or(""))
        {
            return Err(Error::new(
                "CAPABILITY_UNSUPPORTED",
                "connected host lacks the requested optional capability",
            ));
        }
        Ok(())
    }

    fn unsupported_collaboration_auth() -> Error {
        let mut error = Error::new("CAPABILITY_UNSUPPORTED", {
            "The connected CLI host does not support per-open collaboration authentication."
        });
        error.details = Some(Box::new(json!({
            "feature": policy()["features"]["collaborationAuthPerOpen"],
        })));
        error.operation_id = Some("doc.open".into());
        error
    }

    #[allow(clippy::too_many_arguments)]
    async fn rpc_tracked(
        &self,
        method: &str,
        params: Value,
        options: &InvokeOptions,
        mutates: bool,
        open: bool,
        supports_deadline: bool,
        state: Option<Arc<CallState>>,
    ) -> Result<Value> {
        let effective = options
            .timeout
            .unwrap_or(self.0.transport.options().request_timeout);
        let headroom =
            Duration::from_millis(policy()["watchdogHeadroomMs"].as_u64().unwrap_or(5000));
        if options.timeout.is_some() {
            self.require_feature("requestTimeout")?;
            if !supports_deadline {
                return Err(Error::new(
                    "DOCUMENT_RPC_INPUT_UNSUPPORTED",
                    "this native operation does not support an explicit request deadline",
                ));
            }
        }
        let deadline = (supports_deadline
            && self.0.transport.features().contains(
                policy()["features"]["requestTimeout"]
                    .as_str()
                    .unwrap_or(""),
            ))
        .then_some(effective);
        self.0
            .transport
            .request_tracked(
                method,
                params,
                deadline,
                effective.saturating_add(headroom),
                mutates,
                open,
                state.unwrap_or_default(),
            )
            .await
    }

    fn reject_unsupported_dispatch_preview(
        &self,
        id: &str,
        options: &InvokeOptions,
        full: bool,
    ) -> Result<()> {
        if id == "doc.preset.dispatch" && options.dry_run == Some(true) && !full {
            let mut error = Error::new(
                "CAPABILITY_UNSUPPORTED",
                "The core preset dispatcher does not support dryRun previews.",
            );
            error.operation_id = Some(id.into());
            return Err(error);
        }
        Ok(())
    }

    async fn invoke_cli(
        &self,
        id: &str,
        params: Value,
        options: &InvokeOptions,
        auth: Option<&CollaborationAuth>,
    ) -> Result<Value> {
        self.invoke_cli_tracked(id, params, options, auth, None)
            .await
    }

    async fn invoke_cli_tracked(
        &self,
        id: &str,
        params: Value,
        options: &InvokeOptions,
        auth: Option<&CollaborationAuth>,
        state: Option<Arc<CallState>>,
    ) -> Result<Value> {
        let operation = operation(id)?;
        if auth.is_some() {
            if id != "doc.open" {
                return Err(Error::new(
                    "INVALID_ARGUMENT",
                    "authentication is scoped to document open",
                ));
            }
            self.require_feature("collaborationAuthPerOpen")
                .map_err(|_| Self::unsupported_collaboration_auth())?;
        }
        if options
            .stdin_bytes
            .as_ref()
            .is_some_and(|bytes| bytes.len() > 32 * 1024 * 1024)
        {
            return Err(Error::new(
                "PAYLOAD_TOO_LARGE",
                "stdin exceeds the host's 32 MiB limit",
            ));
        }
        validate_json(&params)?;
        let mut request = json!({"argv":routing::argv(operation,&params,options,self.0.options.default_change_mode.as_deref())?,"stdinBase64":options.stdin_bytes.as_ref().map(|bytes|BASE64.encode(bytes)).unwrap_or_default()});
        if let Some(auth) = auth {
            request["collaborationAuth"] = auth.json();
        }
        let response = self
            .rpc_tracked(
                "cli.invoke",
                request,
                options,
                operation["mutates"] == true,
                id == "doc.open",
                true,
                state,
            )
            .await?;
        routing::map_cli(operation, response).map_err(|mut error| {
            error.outcome_unknown = operation["mutates"] == true;
            error
        })
    }

    pub async fn get_tools(&self, provider: &str) -> Result<Value> {
        let _active = self.begin_call()?;
        self.invoke_cli(
            "doc.preset.getTools",
            json!({"provider":provider,"preset":"core"}),
            &InvokeOptions::default(),
            None,
        )
        .await
    }

    pub async fn get_system_prompt(&self) -> Result<Value> {
        let _active = self.begin_call()?;
        self.invoke_cli(
            "doc.preset.getSystemPrompt",
            json!({"preset":"core"}),
            &InvokeOptions::default(),
            None,
        )
        .await
    }
}

impl Invoker for Client {
    async fn invoke(&self, id: &str, params: Value, options: InvokeOptions) -> Result<Value> {
        let _active = self.begin_call()?;
        if ![
            "doc.preset.list",
            "doc.preset.getCatalog",
            "doc.preset.getTools",
            "doc.preset.getSystemPrompt",
            "doc.preset.getMcpPrompt",
        ]
        .contains(&id)
            || operation(id)?["requiresDocumentContext"] != false
            || operation(id)?["mutates"] != false
        {
            return Err(Error::new(
                "INVALID_ARGUMENT",
                "client only invokes sessionless preset getters",
            ));
        }
        self.invoke_cli(id, params, &options, None).await
    }
}

fn operation(id: &str) -> Result<&'static Value> {
    contract()["operations"].get(id).ok_or_else(|| {
        Error::new(
            "OPERATION_NOT_FOUND",
            "operation is not in the generated contract",
        )
    })
}

fn fresh_session_id() -> String {
    static NEXT: AtomicU64 = AtomicU64::new(1);
    format!(
        "rust-{}-{}-{}",
        std::process::id(),
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos(),
        NEXT.fetch_add(1, Ordering::Relaxed)
    )
}

fn validate_json(value: &Value) -> Result<()> {
    let mut stack = vec![(value, 0)];
    while let Some((value, depth)) = stack.pop() {
        if depth > 128 {
            return Err(Error::new(
                "INVALID_ARGUMENT",
                "request JSON exceeds the supported nesting depth",
            ));
        }
        match value {
            Value::Number(number)
                if number.is_i64()
                    && number.as_i64().is_some_and(|n| {
                        !(-crate::wire::Integer::MAX..=crate::wire::Integer::MAX).contains(&n)
                    }) =>
            {
                return Err(Error::new(
                    "INVALID_ARGUMENT",
                    "JSON integer cannot be represented exactly by the host",
                ))
            }
            Value::Number(number)
                if number.is_u64()
                    && number
                        .as_u64()
                        .is_some_and(|n| n > crate::wire::Integer::MAX as u64) =>
            {
                return Err(Error::new(
                    "INVALID_ARGUMENT",
                    "JSON integer cannot be represented exactly by the host",
                ))
            }
            Value::Array(values) => stack.extend(values.iter().map(|value| (value, depth + 1))),
            Value::Object(values) => stack.extend(values.values().map(|value| (value, depth + 1))),
            _ => {}
        }
    }
    Ok(())
}

struct DocumentInner {
    client: Client,
    session_id: String,
    // Active, closing, closed, or outcome unknown. All clones share this state.
    state: Arc<AtomicU8>,
    open_result: DocOpenResult,
}

impl Drop for DocumentInner {
    fn drop(&mut self) {
        if self.state.load(Ordering::Acquire) >= 2 {
            self.client
                .0
                .transport
                .remove_session(&self.session_id, &self.state);
        }
    }
}

#[derive(Clone)]
pub struct Document(Arc<DocumentInner>);

struct CloseGuard {
    state: Arc<AtomicU8>,
    transmission: Arc<CallState>,
}
impl Drop for CloseGuard {
    fn drop(&mut self) {
        let _ = self.state.compare_exchange(
            1,
            if self.transmission.cancel_before_send() {
                0
            } else {
                3
            },
            Ordering::AcqRel,
            Ordering::Acquire,
        );
    }
}

impl Document {
    pub fn session_id(&self) -> &str {
        &self.0.session_id
    }
    pub fn open_result(&self) -> &DocOpenResult {
        &self.0.open_result
    }
    pub fn api(&self) -> DocApi<'_, Self> {
        DocApi::new(self)
    }

    pub async fn close(&self, params: &DocCloseBoundParams) -> Result<DocCloseResult> {
        self.api().close(params).await
    }

    pub async fn replace_file(
        &self,
        path: impl AsRef<Path>,
        options: InvokeOptions,
    ) -> Result<Value> {
        self.active()?;
        let _active = self.0.client.begin_call()?;
        self.replace_path(path.as_ref(), options, None).await
    }

    /// The transport owns staged bytes until the native reply or confirmed exit,
    /// including when the caller drops this future after transmission.
    pub async fn replace_bytes(&self, bytes: &[u8], options: InvokeOptions) -> Result<Value> {
        use std::io::Write;
        if bytes.len() > 32 * 1024 * 1024 {
            return Err(Error::new(
                "PAYLOAD_TOO_LARGE",
                "replacement exceeds 32 MiB",
            ));
        }
        self.active()?;
        let _active = self.0.client.begin_call()?;
        let mut file = tempfile::NamedTempFile::new_in(self.0.client.state_directory())
            .map_err(|_| Error::new("IO_ERROR", "could not stage replacement bytes"))?;
        file.write_all(bytes)
            .map_err(|_| Error::new("IO_ERROR", "could not stage replacement bytes"))?;
        let asset = file.into_temp_path();
        let path = asset.to_path_buf();
        self.replace_path(&path, options, Some(CallState::with_asset(asset)))
            .await
    }

    async fn replace_path(
        &self,
        path: &Path,
        options: InvokeOptions,
        state: Option<Arc<CallState>>,
    ) -> Result<Value> {
        self.active()?;
        self.0.client.require_feature("replaceFile")?;
        let path = path
            .to_str()
            .ok_or_else(|| Error::new("INVALID_ARGUMENT", "replacement path must be UTF-8"))?;
        let transmission = state.unwrap_or_default();

        let response = self
            .0
            .client
            .rpc_tracked(
                "document.replaceFile",
                json!({"sessionId":self.0.session_id,"path":path}),
                &options,
                true,
                false,
                false,
                Some(transmission),
            )
            .await;

        let value = response?;
        if value["sessionId"] != self.0.session_id
            || value["replaced"] != true
            || !value["byteLength"].is_number()
            || !value["revision"].is_string()
        {
            let mut error = Error::new(
                "HOST_PROTOCOL_ERROR",
                "host returned an invalid replacement result",
            );
            error.outcome_unknown = true;
            return Err(error);
        }
        Ok(
            json!({"contextId":self.0.session_id,"runtime":"v2","replaced":true,"document":{"byteLength":value["byteLength"],"revision":value["revision"]}}),
        )
    }

    pub(crate) async fn check_core_revision(
        &self,
        tool_name: &str,
        args: &Value,
        options: &InvokeOptions,
        configured: Option<&crate::generated::types::DocPresetGetToolsParams>,
    ) -> Result<Option<Value>> {
        let Some(expected) = options.expected_revision.as_deref() else {
            return Ok(None);
        };
        let params = serde_json::from_value(
            serde_json::json!({"offset":crate::wire::Integer::MAX,"limit":1}),
        )
        .map_err(Error::decode)?;
        let page = self.api().blocks().list(&params).await?;
        if page.revision.is_empty() || page.revision == "unknown" {
            let cli_reader = self.0.open_result.session_type == "local"
                && self
                    .0
                    .client
                    .0
                    .transport
                    .features()
                    .contains("cli.invoke.localExecutionFacts");
            return Err(if cli_reader {
                Error::new(
                    "HOST_PROTOCOL_ERROR",
                    "CLI returned an invalid execution revision.",
                )
            } else {
                Error::new(
                    "EVIDENCE_UNAVAILABLE",
                    "Required document revision is unavailable.",
                )
            });
        }
        if expected == page.revision {
            return Ok(None);
        }
        if tool_name == "superdoc_perform_action" {
            return crate::agent::stale_action_revision_receipt(args, &page.revision, configured)
                .map(Some);
        }
        Err(Error::new(
            "REVISION_CONFLICT",
            crate::agent::stale_revision_message(tool_name),
        ))
    }

    pub async fn dispatch_tool(
        &self,
        name: &str,
        mut args: Value,
        mut options: InvokeOptions,
    ) -> Result<Value> {
        self.active()?;
        crate::agent::prepare_core_dispatch(name, &mut args, &mut options, None)?;
        self.0.client.reject_unsupported_dispatch_preview(
            "doc.preset.dispatch",
            &options,
            false,
        )?;
        if let Some(receipt) = self
            .check_core_revision(name, &args, &options, None)
            .await?
        {
            return Ok(json!({ "result": receipt }));
        }
        let _active = self.0.client.begin_call()?;
        let transmission = Arc::new(CallState::default());

        let result = self
            .0
            .client
            .invoke_cli_tracked(
                "doc.preset.dispatch",
                json!({"sessionId":self.0.session_id,"toolName":name,"args":args,"preset":"core"}),
                &options,
                None,
                Some(transmission),
            )
            .await;

        result
    }

    fn active(&self) -> Result<()> {
        match self.0.state.load(Ordering::Acquire) {
            0 if !self.0.client.0.transport.is_closed() => Ok(()),
            3 => Err(Error::new(
                "DOCUMENT_STATE_UNKNOWN",
                "document state is uncertain; use a new client to reopen verified saved content",
            )),
            _ => Err(Error::new(
                "DOCUMENT_CLOSED",
                "document handle is closed or closing",
            )),
        }
    }
}

impl Invoker for Document {
    fn handle_decode_error(&self, error: Error) -> Error {
        if error.outcome_unknown {
            let _ = self
                .0
                .state
                .fetch_update(Ordering::AcqRel, Ordering::Acquire, |state| {
                    (state < 2).then_some(3)
                });
        }
        error
    }

    async fn invoke(
        &self,
        id: &str,
        mut params: Value,
        mut options: InvokeOptions,
    ) -> Result<Value> {
        self.active().map_err(|mut error| {
            if error.code == "DOCUMENT_CLOSED" && operation(id).is_ok() {
                error.message =
                    format!("Document handle is closed or closing; cannot invoke {id}.");
                error.operation_id = Some(id.into());
                error.details = Some(Box::new(json!({
                    "sessionId": self.0.session_id,
                    "operationId": id
                })));
            }
            error
        })?;
        let _active = self.0.client.begin_call()?;
        let operation = operation(id)?;
        if operation["sdkSurface"] != "document" {
            return Err(Error::new(
                "INVALID_ARGUMENT",
                "operation is not on the public document surface",
            ));
        }
        self.0.client.reject_unsupported_dispatch_preview(
            id,
            &options,
            params.get("preset") == Some(&json!("full")),
        )?;
        let binds_session = operation["params"]
            .as_array()
            .is_some_and(|params| params.iter().any(|param| param["name"] == "sessionId"));
        if binds_session {
            params["sessionId"] = json!(self.0.session_id);
        }
        let closing = id == "doc.close";
        let transmission = closing.then(|| Arc::new(CallState::default()));
        let _close = if closing {
            self.0
                .state
                .compare_exchange(0, 1, Ordering::AcqRel, Ordering::Acquire)
                .map_err(|_| Error::new("DOCUMENT_CLOSED", "document is already closing"))?;
            Some(CloseGuard {
                state: self.0.state.clone(),
                transmission: transmission.clone().expect("close transmission state"),
            })
        } else {
            None
        };
        if options.change_mode.is_none()
            && operation["params"]
                .as_array()
                .is_some_and(|params| params.iter().any(|param| param["name"] == "changeMode"))
        {
            options.change_mode = self.0.client.0.options.default_change_mode.clone();
        }
        let result = self
            .0
            .client
            .invoke_cli_tracked(id, params, &options, None, transmission)
            .await
            .and_then(|value| routing::map_domain_result(operation, value));
        match result {
            Ok(value) => {
                if closing {
                    if value["closed"] != true || value["contextId"] != self.0.session_id {
                        let mut error = Error::new(
                            "HOST_PROTOCOL_ERROR",
                            "close was not confirmed by the host",
                        );
                        error.outcome_unknown = true;
                        return Err(error);
                    }
                    self.0.state.store(2, Ordering::Release);
                    self.0
                        .client
                        .0
                        .transport
                        .remove_session(&self.0.session_id, &self.0.state);
                }
                Ok(value)
            }
            Err(mut error) => {
                error.operation_id = Some(id.into());
                if closing && !error.outcome_unknown && !self.0.client.0.transport.is_closed() {
                    let _ =
                        self.0
                            .state
                            .compare_exchange(1, 0, Ordering::AcqRel, Ordering::Acquire);
                }
                Err(error)
            }
        }
    }
}

/// Lag is explicit; callers must resynchronize rather than assume every remote
/// change notification was observed. Session invalidation runs in the reader's
/// owner independently of this public notification stream.
pub struct Notifications(broadcast::Receiver<Value>);
impl Notifications {
    pub async fn recv(&mut self) -> Result<Value> {
        self.0.recv().await.map_err(|error| match error {
            broadcast::error::RecvError::Lagged(_) => Error::new(
                "NOTIFICATIONS_LAGGED",
                "resynchronize document state before using notification-derived caches",
            ),
            broadcast::error::RecvError::Closed => {
                Error::new("HOST_DISCONNECTED", "notification stream closed")
            }
        })
    }
}