car-ir 0.55.0

//! Streaming / long-running tool execution contract (EPIC C / C1).
//!
//! CAR's tool dispatch has always been one-shot: a tool is called and the
//! executor `await`s a single `Result<Value, String>`, blocking that DAG
//! slot until the call returns. That model can't express SSH tunnels,
//! long database queries, large uploads, or token-by-token streaming —
//! anything that produces output incrementally or runs longer than a
//! single request.
//!
//! This module defines the *contract* for such tools in the IR (the
//! executor wiring is C2). The shape is start → handle → (poll | stream
//! chunks) → cancel:
//!
//! - [`ToolInvocationMode`] is the capability marker a tool advertises
//!   (`one_shot` by default — fully back-compatible).
//! - Starting a streaming/long-running tool yields a [`ToolHandle`].
//! - The tool emits a sequence of [`ToolStreamChunk`]s (text deltas, data,
//!   progress, a terminal `done`/`error`).
//! - The caller drives it with a [`ToolControl`] (`poll` / `cancel`) and
//!   observes a [`ToolStatus`].
//! - [`ToolStreamEvent`] is the handle-tagged envelope the daemon streams
//!   to a client (the WS surface C2 adds).
//!
//! Every type is `serde(rename_all = "snake_case")` so it round-trips
//! identically across NAPI, PyO3, and the WebSocket protocol — the same
//! contract discipline the rest of the IR follows.

use serde::{Deserialize, Serialize};
use serde_json::Value;

/// How a tool is invoked — the capability marker a tool advertises so the
/// planner/executor negotiate instead of assuming one-shot.
///
/// Defaults to [`ToolInvocationMode::OneShot`], so a tool that doesn't set
/// it behaves exactly as tools always have.
#[derive(
    Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, schemars::JsonSchema,
)]
#[serde(rename_all = "snake_case")]
pub enum ToolInvocationMode {
    /// Classic blocking call returning a single result. The default.
    #[default]
    OneShot,
    /// Produces output incrementally as a sequence of [`ToolStreamChunk`]s.
    Streaming,
    /// Runs longer than a single request; started to a [`ToolHandle`] and
    /// driven via poll/cancel. May or may not also stream chunks.
    LongRunning,
}

impl ToolInvocationMode {
    /// Whether this mode runs detached from the initial call (streaming or
    /// long-running) rather than blocking for a single result.
    pub fn is_detached(self) -> bool {
        !matches!(self, ToolInvocationMode::OneShot)
    }
}

/// An opaque handle to a started streaming/long-running tool invocation.
/// The caller passes it back to poll, observe chunks, or cancel.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, schemars::JsonSchema)]
pub struct ToolHandle {
    /// Server-assigned id, unique within a runtime.
    pub id: String,
}

impl ToolHandle {
    pub fn new(id: impl Into<String>) -> Self {
        Self { id: id.into() }
    }
}

/// One unit of output from a streaming/long-running tool.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ToolStreamChunk {
    /// An incremental text delta (e.g. command output line, token).
    Text { text: String },
    /// A structured data fragment.
    Data { data: Value },
    /// Progress signal: `fraction` in `[0,1]` plus an optional message.
    Progress {
        fraction: f64,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        message: Option<String>,
    },
    /// Terminal success. Carries the final result, if any. After `done`
    /// no further chunks are emitted for the handle.
    Done {
        #[serde(default, skip_serializing_if = "Option::is_none")]
        result: Option<Value>,
    },
    /// Terminal failure with an error message. Also terminates the stream.
    Error { message: String },
}

impl ToolStreamChunk {
    /// Whether this chunk terminates the stream (`done` or `error`).
    pub fn is_terminal(&self) -> bool {
        matches!(
            self,
            ToolStreamChunk::Done { .. } | ToolStreamChunk::Error { .. }
        )
    }
}

/// A control operation a caller issues against a running tool handle.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum ToolControl {
    /// Ask for the current [`ToolStatus`] without blocking.
    Poll,
    /// Request cancellation. Cooperative — the tool transitions to
    /// `cancelled` once it observes the request.
    Cancel,
}

/// The lifecycle state of a streaming/long-running tool invocation.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum ToolStatus {
    /// Still executing; more chunks may arrive.
    Running,
    /// Finished successfully (a `done` chunk was emitted).
    Succeeded,
    /// Finished with an error (an `error` chunk was emitted).
    Failed,
    /// Cancellation was requested and honored.
    Cancelled,
}

impl ToolStatus {
    /// Whether the invocation has reached a terminal state.
    pub fn is_terminal(self) -> bool {
        !matches!(self, ToolStatus::Running)
    }
}

/// A handle-tagged stream chunk — the envelope the daemon streams to a
/// client over the WS surface (added in C2).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
pub struct ToolStreamEvent {
    pub handle: ToolHandle,
    pub chunk: ToolStreamChunk,
}

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

    #[test]
    fn mode_defaults_to_one_shot_and_roundtrips() {
        assert_eq!(ToolInvocationMode::default(), ToolInvocationMode::OneShot);
        assert!(!ToolInvocationMode::OneShot.is_detached());
        assert!(ToolInvocationMode::Streaming.is_detached());
        assert!(ToolInvocationMode::LongRunning.is_detached());
        // snake_case wire form.
        let j = serde_json::to_string(&ToolInvocationMode::LongRunning).unwrap();
        assert_eq!(j, "\"long_running\"");
        let back: ToolInvocationMode = serde_json::from_str(&j).unwrap();
        assert_eq!(back, ToolInvocationMode::LongRunning);
    }

    #[test]
    fn chunk_tagged_serialization() {
        let c = ToolStreamChunk::Text {
            text: "hello".into(),
        };
        let j = serde_json::to_value(&c).unwrap();
        assert_eq!(j["kind"], "text");
        assert_eq!(j["text"], "hello");
        // round-trip
        let back: ToolStreamChunk = serde_json::from_value(j).unwrap();
        assert_eq!(back, c);
    }

    #[test]
    fn terminal_chunks_and_status() {
        assert!(ToolStreamChunk::Done { result: None }.is_terminal());
        assert!(ToolStreamChunk::Error {
            message: "x".into()
        }
        .is_terminal());
        assert!(!ToolStreamChunk::Text { text: "x".into() }.is_terminal());
        assert!(ToolStatus::Succeeded.is_terminal());
        assert!(ToolStatus::Cancelled.is_terminal());
        assert!(!ToolStatus::Running.is_terminal());
    }

    #[test]
    fn stream_event_roundtrips() {
        let ev = ToolStreamEvent {
            handle: ToolHandle::new("h1"),
            chunk: ToolStreamChunk::Progress {
                fraction: 0.5,
                message: Some("halfway".into()),
            },
        };
        let j = serde_json::to_string(&ev).unwrap();
        let back: ToolStreamEvent = serde_json::from_str(&j).unwrap();
        assert_eq!(back, ev);
    }

    #[test]
    fn control_wire_forms() {
        assert_eq!(
            serde_json::to_string(&ToolControl::Cancel).unwrap(),
            "\"cancel\""
        );
        assert_eq!(
            serde_json::to_string(&ToolControl::Poll).unwrap(),
            "\"poll\""
        );
    }
}