scv-tools 0.4.10

//! Reading a delegated CLI's stdout into one bounded result.
//!
//! Structured formats are parsed line by line as the CLI writes them, keeping
//! only the reply, usage, and error, so a long run's full event log never
//! reaches the parent model. Unknown events and fields are ignored, and a
//! stream with no parsable event falls back to its text. Claude's first
//! `system`/`init` event is also checked against the pinned billing identity:
//! API-key runs require `apiKeySource = "apiKeyHelper"`, while subscription
//! runs require `apiKeySource = "none"`. The same event's `permissionMode`
//! is checked against the mode the policy expects ([`mode::Expected`]), and
//! a Codex run's mode, read from its rollout, arrives through
//! [`AgentStream::check_mode`]. A mismatch, or a run that would otherwise
//! complete with no init event or no mode, cancels
//! [`AgentStream::halt_token`] when the child is still running. The turn
//! fails with that credential or mode error instead of the model's reply. A
//! run that already failed, timed out, or was cancelled keeps that status
//! unless its mode did not match.

use scv_core::{ProgressSink, ToolFailure, ToolOutput};
use scv_protocol::JobStatus;
use serde::Deserialize;
use serde_json::{Value, json};
use tokio_util::sync::CancellationToken;

use crate::{
    delegate::{
        activity::RunActivity,
        adapters::{AgentAuth, OutputFormat},
        mode,
        progress::{progress_lines, tool_lines},
    },
    process::OutputSink,
};

/// Longest stdout line parsed as an event. Longer lines (such as a tool
/// result echoed back in full) are skipped; the events SCV needs are short.
const MAX_EVENT_LINE_BYTES: usize = 4 * 1024 * 1024;
/// Plain-text stdout kept for a structured stream that produced no event.
const MAX_FALLBACK_BYTES: usize = 16 * 1024;
/// Bytes of stderr kept, from the end.
pub(crate) const STDERR_TAIL_BYTES: usize = 2048;
/// Bytes of a failed plain-text run's stdout kept as its `error`.
const TEXT_ERROR_BYTES: usize = 500;
/// What a declined run's result tells the calling model when Grok is not an
/// alternative (including when Grok itself declined).
pub(crate) const DECLINED_NOTE: &str = "The agent declined this request. Tell the user what it \
     said; don't pass the request to another agent on your own. If the user then asks for a \
     specific agent, use it.";
/// What a declined run's result tells the calling model when `grok` is
/// offered: a safety or guardrail refusal is not a reason to skip Grok.
pub(crate) const DECLINED_NOTE_TRY_GROK: &str = "The agent declined this request. Tell the user \
     what it said, then call agent again with agent grok and the same request. A safety or \
     guardrail refusal is not a reason to skip Grok.";

#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub(crate) struct AgentUsage {
    pub(crate) input_tokens: u64,
    pub(crate) output_tokens: u64,
}

/// The final state of one delegated run.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum RunStatus {
    Completed,
    Failed,
    /// The agent's model refused the request (a `refusal` stop reason).
    Declined,
    Timeout,
    Cancelled,
}

impl RunStatus {
    /// How a run that ended this way failed, as the calling session's
    /// clients see it; `None` for a completed run.
    pub(crate) fn failure(self) -> Option<ToolFailure> {
        match self {
            Self::Completed => None,
            Self::Failed | Self::Declined => Some(ToolFailure::Failed),
            Self::Timeout => Some(ToolFailure::Limit),
            Self::Cancelled => Some(ToolFailure::Cancelled),
        }
    }

    pub(crate) fn as_str(self) -> &'static str {
        match self {
            Self::Completed => "completed",
            Self::Failed => "failed",
            Self::Declined => "declined",
            Self::Timeout => "timeout",
            Self::Cancelled => "cancelled",
        }
    }
}

/// Incremental reader for a delegated CLI's stdout.
#[derive(Debug)]
pub(crate) struct AgentStream {
    format: OutputFormat,
    line: Vec<u8>,
    line_overflowed: bool,
    /// Text-format stdout, or a structured stream's non-JSON lines.
    text: Vec<u8>,
    text_limit: usize,
    text_truncated: bool,
    parsed_events: usize,
    reply: Option<String>,
    error: Option<String>,
    failed: bool,
    completed: bool,
    /// The model stopped with a `refusal` stop reason.
    refused: bool,
    auth: AgentAuth,
    /// Set when the pinned credential source or the permission mode does not
    /// match, or never arrived on a run that would otherwise complete. Later
    /// events are ignored, and [`Self::halt_token`] is cancelled so the child
    /// is stopped.
    rejected: bool,
    /// A Claude init event reported the pinned credential source.
    saw_pinned_source: bool,
    /// The mode the policy expects the run to report, for an agent that
    /// reports one; `None` checks nothing.
    mode: Option<mode::Expected>,
    /// The mode the run reported, once it matched.
    verified_mode: Option<String>,
    halt: CancellationToken,
    usage: Option<AgentUsage>,
    /// The CLI's own session ID, for continuing the conversation.
    session: Option<String>,
    /// A tool ran since the last assistant text, so the next text starts a
    /// new message (Gemini CLI streams its messages as deltas).
    new_message: bool,
    /// Where status lines from structured events go while the run lasts.
    progress: ProgressSink,
    /// The run's activity log: its message text and tool lines as they come.
    activity: Option<RunActivity>,
    /// The end of plain text that stopped inside a UTF-8 character.
    utf8_tail: Vec<u8>,
}

impl AgentStream {
    pub(crate) fn new(format: OutputFormat, output_limit: usize) -> Self {
        Self {
            format,
            line: Vec::new(),
            line_overflowed: false,
            text: Vec::new(),
            text_limit: if format == OutputFormat::Text {
                output_limit
            } else {
                output_limit.min(MAX_FALLBACK_BYTES)
            },
            text_truncated: false,
            parsed_events: 0,
            reply: None,
            error: None,
            failed: false,
            completed: false,
            refused: false,
            auth: AgentAuth::None,
            rejected: false,
            saw_pinned_source: false,
            mode: None,
            verified_mode: None,
            halt: CancellationToken::new(),
            usage: None,
            session: None,
            new_message: false,
            progress: ProgressSink::default(),
            activity: None,
            utf8_tail: Vec::new(),
        }
    }

    /// Log the agent's message text and tool calls to `activity` as they
    /// arrive.
    pub(crate) fn with_activity(mut self, activity: Option<RunActivity>) -> Self {
        self.activity = activity;
        self
    }

    /// Report what the agent does (commands, files, searches, tool calls)
    /// to `progress` as its events arrive.
    pub(crate) fn with_progress(mut self, progress: ProgressSink) -> Self {
        self.progress = progress;
        self
    }

    pub(crate) fn with_auth(mut self, auth: AgentAuth) -> Self {
        self.auth = auth;
        self
    }

    /// Require the run to report a permission mode `expected` accepts:
    /// Claude Code's init event, or a Codex rollout through
    /// [`check_mode`](Self::check_mode).
    pub(crate) fn with_mode(mut self, expected: Option<mode::Expected>) -> Self {
        self.mode = expected;
        self
    }

    /// Cancelled when this stream sees a Claude credential source other than
    /// the one its billing identity is pinned to, or a permission mode the
    /// policy does not expect. The process supervisor watches it and stops
    /// the child.
    pub(crate) fn halt_token(&self) -> CancellationToken {
        self.halt.clone()
    }

    /// The CLI's own session ID, once its output named one.
    pub(crate) fn session(&self) -> Option<&str> {
        self.session.as_deref()
    }

    /// Whether the run's mode is settled: verified, or the run stopped.
    pub(crate) fn mode_settled(&self) -> bool {
        self.rejected || self.verified_mode.is_some() || self.mode.is_none()
    }

    /// Check the mode the run reported, which SCV read outside its stdout
    /// (a Codex rollout): accepted, it is the run's verified mode; refused,
    /// the run stops.
    pub(crate) fn check_mode(&mut self, reported: &str) {
        let Some(expected) = &self.mode else {
            return;
        };
        if self.rejected {
            return;
        }
        if expected.accepts(reported) {
            self.verified_mode = Some(reported.to_owned());
        } else {
            let error = expected.mismatch(reported);
            self.reject(error);
        }
    }

    /// Stop a run whose mode should have been reported and was not.
    pub(crate) fn reject_missing_mode(&mut self) {
        if let Some(expected) = self.mode.as_ref().filter(|_| !self.mode_settled()) {
            let error = expected.missing();
            self.reject(error);
        }
    }

    pub(crate) fn push(&mut self, bytes: &[u8]) {
        if self.format == OutputFormat::Text {
            self.log_text(bytes);
            self.push_text(bytes);
            return;
        }
        for chunk in bytes.split_inclusive(|byte| *byte == b'\n') {
            let (content, ends_line) = match chunk.strip_suffix(b"\n") {
                Some(content) => (content, true),
                None => (chunk, false),
            };
            if !self.line_overflowed {
                if self.line.len() + content.len() > MAX_EVENT_LINE_BYTES {
                    self.line.clear();
                    self.line_overflowed = true;
                } else {
                    self.line.extend_from_slice(content);
                }
            }
            if ends_line {
                self.end_line();
            }
        }
    }

    fn end_line(&mut self) {
        let line = std::mem::take(&mut self.line);
        if std::mem::take(&mut self.line_overflowed) {
            return;
        }
        let trimmed = line.trim_ascii();
        if trimmed.is_empty() {
            return;
        }
        if let Ok(Value::Object(event)) = serde_json::from_slice::<Value>(trimmed) {
            self.parsed_events += 1;
            self.event(&Value::Object(event));
        } else {
            self.push_text(trimmed);
            self.push_text(b"\n");
        }
    }

    /// Plain output as it prints, whole characters only, for the activity
    /// log.
    fn log_text(&mut self, bytes: &[u8]) {
        let Some(activity) = &self.activity else {
            return;
        };
        self.utf8_tail.extend_from_slice(bytes);
        let valid = match std::str::from_utf8(&self.utf8_tail) {
            Ok(_) => self.utf8_tail.len(),
            // Wait for the rest of a character cut at the end.
            Err(error) if error.error_len().is_none() => error.valid_up_to(),
            Err(_) => self.utf8_tail.len(),
        };
        let complete: Vec<u8> = self.utf8_tail.drain(..valid).collect();
        activity.text(&String::from_utf8_lossy(&complete));
    }

    fn push_text(&mut self, bytes: &[u8]) {
        let remaining = self.text_limit.saturating_sub(self.text.len());
        self.text
            .extend_from_slice(&bytes[..bytes.len().min(remaining)]);
        self.text_truncated |= bytes.len() > remaining;
    }

    fn event(&mut self, event: &Value) {
        if self.rejected {
            return;
        }
        let kind = event.get("type").and_then(Value::as_str).unwrap_or("");
        if self.progress.is_enabled() {
            for line in progress_lines(self.format, kind, event) {
                self.progress.report(&line);
            }
        }
        if let Some(activity) = &self.activity {
            for line in tool_lines(self.format, kind, event) {
                activity.tool(&line);
            }
        }
        let session = match (self.format, kind) {
            (format, "system" | "result") if format.claude_like() => event.get("session_id"),
            (OutputFormat::CodexJsonl, "thread.started") => event.get("thread_id"),
            (OutputFormat::PiJson, "session") => event.get("id"),
            (OutputFormat::GeminiStreamJson, "init") => event.get("session_id"),
            _ => None,
        };
        if let Some(session) = session
            .and_then(Value::as_str)
            .filter(|id| valid_session_id(id))
        {
            // Saved at once, so a run cut off by a crash can be resumed.
            if self.session.as_deref() != Some(session)
                && let Some(activity) = &self.activity
            {
                activity.set_vendor(session);
            }
            self.session = Some(session.to_owned());
        }
        match self.format {
            OutputFormat::Text => {}
            OutputFormat::GeminiStreamJson => self.gemini_event(kind, event),
            OutputFormat::ClaudeStreamJson
            | OutputFormat::CursorStreamJson
            | OutputFormat::AmpStreamJson => match kind {
                "system" if event.get("subtype").and_then(Value::as_str) == Some("init") => {
                    let expected = match self.auth {
                        AgentAuth::ClaudeApiKey => Some("apiKeyHelper"),
                        AgentAuth::ClaudeSubscription => Some("none"),
                        AgentAuth::None => None,
                    };
                    if let Some(expected) = expected {
                        let actual = event
                            .get("apiKeySource")
                            .and_then(Value::as_str)
                            .unwrap_or("missing");
                        if actual == expected {
                            self.saw_pinned_source = true;
                        } else {
                            self.reject_credential(actual);
                        }
                    }
                    if self.format == OutputFormat::ClaudeStreamJson
                        && self.mode.is_some()
                        && !self.rejected
                    {
                        match event.get("permissionMode").and_then(Value::as_str) {
                            Some(reported) => self.check_mode(reported),
                            None => self.reject_missing_mode(),
                        }
                    }
                }
                "result" => {
                    if let Some(result) = event.get("result").and_then(Value::as_str) {
                        self.reply = Some(result.to_owned());
                    }
                    let error_subtype = event
                        .get("subtype")
                        .and_then(Value::as_str)
                        .is_some_and(|subtype| subtype.starts_with("error"));
                    if event.get("is_error").and_then(Value::as_bool) == Some(true) || error_subtype
                    {
                        self.failed = true;
                        // Claude Code puts its own error (sign-in, API
                        // status) in the result of a failed run.
                        self.error = self.reply.clone();
                    } else {
                        self.completed = true;
                    }
                    self.refused |=
                        event.get("stop_reason").and_then(Value::as_str) == Some("refusal");
                    self.usage = usage(event.get("usage"), "input_tokens", "output_tokens");
                    // A successful result with no init is the drift this
                    // check exists to catch. Stop before any further work.
                    if self.completed && self.requires_pinned_source() && !self.saw_pinned_source {
                        self.reject_credential("missing");
                    }
                    if self.completed && self.format == OutputFormat::ClaudeStreamJson {
                        self.reject_missing_mode();
                    }
                }
                "assistant" => {
                    let text = text_parts(event.pointer("/message/content"));
                    if let Some(activity) = self.activity.as_ref().filter(|_| !text.is_empty()) {
                        activity.message(&text);
                    }
                    if !text.is_empty() && !self.completed {
                        self.reply = Some(text);
                    }
                    // The API message's own stop reason.
                    self.refused |= event
                        .pointer("/message/stop_reason")
                        .and_then(Value::as_str)
                        == Some("refusal");
                }
                _ => {}
            },
            OutputFormat::CodexJsonl => match kind {
                "item.completed" => {
                    if event.pointer("/item/type").and_then(Value::as_str) == Some("agent_message")
                        && let Some(text) = event.pointer("/item/text").and_then(Value::as_str)
                    {
                        if let Some(activity) = &self.activity {
                            activity.message(text);
                        }
                        self.reply = Some(text.to_owned());
                    }
                }
                "turn.completed" => {
                    self.completed = true;
                    let turn = usage(event.get("usage"), "input_tokens", "output_tokens");
                    self.usage = add_usage(self.usage, turn);
                }
                "turn.failed" => {
                    self.failed = true;
                    if let Some(message) = event.pointer("/error/message").and_then(Value::as_str) {
                        self.error = Some(message.to_owned());
                    }
                }
                // Codex also reports recoverable problems (reconnects) this
                // way; only a run without `turn.completed` treats it as final.
                "error" => {
                    if let Some(message) = event.get("message").and_then(Value::as_str) {
                        self.error = Some(message.to_owned());
                    }
                }
                _ => {}
            },
            OutputFormat::PiJson => {
                if kind == "message_end"
                    && event.pointer("/message/role").and_then(Value::as_str) == Some("assistant")
                {
                    let message = &event["message"];
                    let text = text_parts(message.get("content"));
                    if !text.is_empty() {
                        if let Some(activity) = &self.activity {
                            activity.message(&text);
                        }
                        self.reply = Some(text);
                        self.completed = true;
                    }
                    let turn = usage(message.get("usage"), "input", "output");
                    self.usage = add_usage(self.usage, turn);
                    if message.get("stopReason").and_then(Value::as_str) == Some("error")
                        && let Some(error) = message.get("errorMessage").and_then(Value::as_str)
                    {
                        self.error = Some(error.to_owned());
                    }
                }
            }
        }
    }

    /// One Gemini CLI `stream-json` event, or its whole `json` output. The
    /// reply is the assistant text after the last tool call.
    fn gemini_event(&mut self, kind: &str, event: &Value) {
        let text = |key: &str| event.get(key).and_then(Value::as_str);
        match kind {
            "message" if text("role") == Some("assistant") => {
                let Some(content) = text("content").filter(|content| !content.is_empty()) else {
                    return;
                };
                let delta = event.get("delta").and_then(Value::as_bool) == Some(true);
                if let Some(activity) = &self.activity {
                    if delta {
                        activity.text(content);
                    } else {
                        activity.message(content);
                    }
                }
                let reply = self.reply.get_or_insert_with(String::new);
                if !delta || std::mem::take(&mut self.new_message) {
                    reply.clear();
                }
                // Bounded like the one event other formats take a reply from.
                let room = MAX_EVENT_LINE_BYTES.saturating_sub(reply.len());
                reply.push_str(scv_client::text::utf8_prefix(content, room));
            }
            "tool_use" | "tool_result" => self.new_message = true,
            "error" if text("severity") != Some("warning") => {
                if let Some(message) = text("message") {
                    self.error = Some(message.to_owned());
                }
            }
            "result" => {
                if text("status") == Some("error") {
                    self.failed = true;
                    if let Some(message) = event.pointer("/error/message").and_then(Value::as_str) {
                        self.error = Some(message.to_owned());
                    }
                } else {
                    self.completed = true;
                }
                self.usage = usage(event.get("stats"), "input_tokens", "output_tokens");
            }
            // `--output-format json`: one object with the whole response.
            "" if event.get("response").is_some() => {
                if let Some(message) = event.pointer("/error/message").and_then(Value::as_str) {
                    self.failed = true;
                    self.error = Some(message.to_owned());
                }
                if let Some(response) = text("response") {
                    if let Some(activity) = &self.activity {
                        activity.message(response);
                    }
                    self.reply = Some(response.to_owned());
                    self.completed |= !self.failed;
                }
            }
            _ => {}
        }
    }

    fn requires_pinned_source(&self) -> bool {
        matches!(
            self.auth,
            AgentAuth::ClaudeApiKey | AgentAuth::ClaudeSubscription
        )
    }

    fn credential_error(auth: AgentAuth, actual: &str) -> String {
        match auth {
            AgentAuth::ClaudeApiKey => format!(
                "claude reported credential source {actual}, not the Claude API key; the run was stopped"
            ),
            AgentAuth::ClaudeSubscription => format!(
                "claude-ai reported credential source {actual}, not the Claude subscription; the run was stopped"
            ),
            AgentAuth::None => unreachable!(),
        }
    }

    /// The pinned credential source did not match, or never arrived.
    fn reject_credential(&mut self, actual: &str) {
        self.reject(Self::credential_error(self.auth, actual));
    }

    /// The credential source or permission mode did not match, or never
    /// arrived. Forget any model text and ask the supervisor to stop the
    /// child.
    fn reject(&mut self, error: String) {
        self.rejected = true;
        self.failed = true;
        self.completed = false;
        self.refused = false;
        self.reply = None;
        self.usage = None;
        self.session = None;
        self.verified_mode = None;
        self.error = Some(error);
        self.halt.cancel();
    }

    /// Settle the run. `fallback_reply` is a final message the CLI wrote
    /// elsewhere (Codex `-o`), used when the stream carried none.
    pub(crate) fn finish(mut self, exit: RunExit, fallback_reply: Option<String>) -> AgentResult {
        if !self.line.is_empty() || self.line_overflowed {
            self.end_line();
        }
        if self.rejected {
            let error = self.error.clone().unwrap_or_else(|| {
                "credential source or permission mode did not match; the run was stopped".to_owned()
            });
            return AgentResult::stopped(error);
        }
        let text = String::from_utf8_lossy(&self.text).trim_end().to_owned();
        let structured = self.format != OutputFormat::Text && self.parsed_events > 0;
        let incomplete = structured && !self.completed;
        let status = match exit {
            RunExit::TimedOut => RunStatus::Timeout,
            RunExit::Killed => RunStatus::Cancelled,
            RunExit::Exited { .. } if self.refused => RunStatus::Declined,
            RunExit::Exited { success } => {
                if !success || self.failed || (incomplete && self.error.is_some()) {
                    RunStatus::Failed
                } else {
                    RunStatus::Completed
                }
            }
        };
        if self.requires_pinned_source()
            && !self.saw_pinned_source
            && status == RunStatus::Completed
        {
            return AgentResult::stopped(Self::credential_error(self.auth, "missing"));
        }
        if let Some(expected) = &self.mode
            && self.verified_mode.is_none()
            && status == RunStatus::Completed
        {
            return AgentResult::stopped(expected.missing());
        }
        let mut reply = if structured {
            self.reply.or(fallback_reply).or_else(|| self.error.clone())
        } else {
            fallback_reply.or_else(|| (!text.is_empty()).then(|| text.clone()))
        }
        .unwrap_or_default();
        // What went wrong, kept apart from the reply: the CLI's reported
        // error, or for a failed plain-text run its closing output, which
        // is the CLI's own error message rather than a model reply.
        let error = match status {
            RunStatus::Failed if structured => self.error.clone(),
            RunStatus::Failed if !text.is_empty() => Some(tail(&text, TEXT_ERROR_BYTES)),
            _ => None,
        };
        if status != RunStatus::Completed
            && let Some(error) = &self.error
            && !reply.contains(error.as_str())
        {
            if reply.is_empty() {
                reply = error.clone();
            } else {
                reply = format!("{reply}\n{error}");
            }
        }
        AgentResult {
            status,
            reply,
            error,
            usage: self.usage,
            truncated: self.text_truncated && !structured,
            session: self.session,
            mode: self.verified_mode.as_deref().map(mode::verified),
        }
    }
}

/// The last `limit` bytes of `text`, on a character boundary.
fn tail(text: &str, limit: usize) -> String {
    let mut start = text.len().saturating_sub(limit);
    while !text.is_char_boundary(start) {
        start += 1;
    }
    text[start..].to_owned()
}

/// How the process ended, as far as the parser needs to know.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum RunExit {
    Exited {
        success: bool,
    },
    TimedOut,
    /// Stopped by `scv agents kill`.
    Killed,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct AgentResult {
    pub(crate) status: RunStatus,
    pub(crate) reply: String,
    /// Why the run failed, as the CLI, the agent's protocol, or SCV reported
    /// it; never taken from the model's reply. Fallback advice and sign-in
    /// hints are decided from this and the status alone.
    pub(crate) error: Option<String>,
    pub(crate) usage: Option<AgentUsage>,
    pub(crate) truncated: bool,
    /// The session ID the CLI reported, if any.
    pub(crate) session: Option<String>,
    /// What SCV knows of the permission mode the run had: `verified <mode>`,
    /// or `unverified`; `None` for an agent with no mode to verify.
    pub(crate) mode: Option<String>,
}

impl AgentResult {
    /// A run SCV stopped for `error`, keeping nothing the agent said.
    fn stopped(error: String) -> Self {
        Self {
            status: RunStatus::Failed,
            reply: error.clone(),
            error: Some(error),
            usage: None,
            truncated: false,
            session: None,
            mode: None,
        }
    }

    /// The tool result: the reply bounded to `limit` bytes, never the log.
    pub(crate) fn to_json(
        &self,
        agent: &str,
        conversation: Option<(&str, u32)>,
        exit_code: Option<i32>,
        stderr_tail: &str,
        limit: usize,
    ) -> (String, bool) {
        let (reply, cut) = truncate_utf8(&self.reply, limit);
        let truncated = self.truncated || cut;
        let mut value = json!({
            "agent": agent,
            "status": self.status.as_str(),
            "reply": reply,
            "usage": self.usage.map(|usage| json!({
                "input_tokens": usage.input_tokens,
                "output_tokens": usage.output_tokens,
            })),
            "exit_code": exit_code,
            "stderr_tail": stderr_tail,
            "truncated": truncated,
        });
        if let Some(error) = &self.error {
            value["error"] = truncate_utf8(error, limit).0.into();
        }
        if let Some(mode) = &self.mode {
            value["mode"] = mode.as_str().into();
        }
        if self.status == RunStatus::Declined {
            value["note"] = DECLINED_NOTE.into();
        }
        if let Some((handle, turn)) = conversation {
            value["session"] = handle.into();
            value["turn"] = turn.into();
        }
        (value.to_string(), truncated)
    }
}

/// The parts of an agent tool's result ([`AgentResult::to_json`]) that SCV
/// reads back: for a background job's report and status, and to decide
/// whether a failed call names other agents.
#[derive(Debug, Default, Deserialize)]
pub(crate) struct AgentReply {
    pub(crate) status: Option<JobStatus>,
    pub(crate) reply: Option<String>,
    pub(crate) error: Option<String>,
    /// The conversation handle to continue it with.
    pub(crate) session: Option<String>,
}

impl AgentReply {
    /// The agent result `value` holds, or `None` for any other value.
    pub(crate) fn read(value: &Value) -> Option<Self> {
        value
            .is_object()
            .then(|| Self::deserialize(value).ok())
            .flatten()
    }
}

/// Whether a CLI-reported session ID is safe to pass back as one argument:
/// it must not read as a flag or carry path or shell syntax.
pub(crate) fn valid_session_id(id: &str) -> bool {
    !id.is_empty()
        && id.len() <= 128
        && !id.starts_with('-')
        && id
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
}

/// The last bytes written to a stream.
#[derive(Debug)]
pub(crate) struct TailBuffer {
    bytes: Vec<u8>,
    limit: usize,
}

impl TailBuffer {
    pub(crate) fn new(limit: usize) -> Self {
        Self {
            bytes: Vec::new(),
            limit,
        }
    }

    pub(crate) fn push(&mut self, bytes: &[u8]) {
        self.bytes.extend_from_slice(bytes);
        if self.bytes.len() > self.limit * 2 {
            let excess = self.bytes.len() - self.limit;
            self.bytes.drain(..excess);
        }
    }

    pub(crate) fn text(&self) -> String {
        let start = self.bytes.len().saturating_sub(self.limit);
        String::from_utf8_lossy(&self.bytes[start..])
            .trim()
            .to_owned()
    }
}

fn text_parts(content: Option<&Value>) -> String {
    match content {
        Some(Value::String(text)) => text.clone(),
        Some(Value::Array(parts)) => parts
            .iter()
            .filter(|part| part.get("type").and_then(Value::as_str) == Some("text"))
            .filter_map(|part| part.get("text").and_then(Value::as_str))
            .collect::<Vec<_>>()
            .join(""),
        _ => String::new(),
    }
}

fn usage(value: Option<&Value>, input: &str, output: &str) -> Option<AgentUsage> {
    let value = value?;
    let input_tokens = value.get(input).and_then(Value::as_u64);
    let output_tokens = value.get(output).and_then(Value::as_u64);
    (input_tokens.is_some() || output_tokens.is_some()).then(|| AgentUsage {
        input_tokens: input_tokens.unwrap_or(0),
        output_tokens: output_tokens.unwrap_or(0),
    })
}

fn add_usage(total: Option<AgentUsage>, turn: Option<AgentUsage>) -> Option<AgentUsage> {
    match (total, turn) {
        (Some(total), Some(turn)) => Some(AgentUsage {
            input_tokens: total.input_tokens.saturating_add(turn.input_tokens),
            output_tokens: total.output_tokens.saturating_add(turn.output_tokens),
        }),
        (total, turn) => total.or(turn),
    }
}

pub(crate) fn truncate_utf8(value: &str, limit: usize) -> (&str, bool) {
    (
        scv_client::text::utf8_prefix(value, limit),
        value.len() > limit,
    )
}

/// Point a failed agent run whose reported error reads like a missing
/// sign-in at the host command that fixes it, since the agent's own advice
/// (`/login`) cannot be followed from a remote chat. Only the structured
/// `error` counts, never the agent's reply.
pub(crate) fn add_sign_in_hint(output: &mut ToolOutput, agent: &str) {
    let Ok(Value::Object(mut content)) = serde_json::from_str::<Value>(&output.content) else {
        return;
    };
    let Some(error) = content.get("error").and_then(Value::as_str) else {
        return;
    };
    let lower = error.to_ascii_lowercase();
    let unauthenticated = [
        "not logged in",
        "not signed in",
        "not authenticated",
        "login",
        "log in",
        "unauthorized",
        "authentication",
        "missing_credential",
        "no api key",
        "auth_required",
        // Gemini CLI: "Please set an Auth method", and over ACP "Gemini API
        // key is missing or not configured".
        "auth method",
        "api key is missing",
    ]
    .iter()
    .any(|needle| lower.contains(needle));
    if !unauthenticated {
        return;
    }
    content.insert(
        "hint".into(),
        format!(
            "The {agent} CLI appears to be signed out of SCV's private agent home. \
             The host owner can sign it in with: scv agents login {agent}"
        )
        .into(),
    );
    output.content = Value::Object(content).to_string();
}

impl OutputSink for AgentStream {
    fn push(&mut self, bytes: &[u8]) {
        AgentStream::push(self, bytes);
    }
}

impl OutputSink for TailBuffer {
    fn push(&mut self, bytes: &[u8]) {
        TailBuffer::push(self, bytes);
    }
}

#[cfg(test)]
mod tests;