a4aa 0.5.0

//! Native agent-server lane (`lane: "serve"`).
//!
//! Some agents ship a first-party HTTP server that is richer than the
//! generic protocol bridges built on top of it — OpenCode's own `serve`
//! process is the canonical example: its `acp` binary is itself a client of
//! this server's event feed, and high-volume sessions can overflow that
//! internal subscriber queue. Driving the REST surface directly with
//! request/response polling never creates that queue, so this lane is the
//! preferred transport where the agent offers it and the protocol lanes
//! remain the portable fallback.
//!
//! Startup discovery: the child announces `server listening on <url>` and
//! `server password <secret>` on stdout; the lane reads those lines before
//! issuing requests. Sessions are created and driven over the versioned
//! REST API with HTTP basic auth:
//!
//! - `POST /api/session` — create a session in the server's working
//!   directory (the lane spawns the server with the session's cwd).
//! - `POST /api/session/:id/model` — apply a `provider/model` selection.
//! - `POST /api/session/:id/prompt` — submit one turn; the API returns the
//!   accepted user message immediately and the turn runs asynchronously.
//! - `GET  /api/session/:id/message` — poll until the terminal `idle`
//!   record reports the turn's outcome; assistant parts carry the text.
//! - `POST /api/session/:id/interrupt` — cancel the in-flight turn.
//!
//! MCP servers declared on the lane are merged into the server's
//! `OPENCODE_CONFIG_CONTENT` document (`mcp.servers`) before spawn — the
//! server connects them lazily per session, so the lane needs no MCP
//! protocol machinery of its own.

use std::collections::BTreeMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;

use serde_json::{json, Value};
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::sync::{mpsc, Mutex};

use crate::bounds;
use crate::env::SecretResolver;
use crate::error::AdapterError;
use crate::model::{AgentEvent, CommandSpec, SessionRequest, TokenUsage, TurnOutcome, TurnRequest};
use crate::observe::{Direction, SharedObserver};
use crate::supervision::SupervisedChild;
use crate::transport::{base64, stdio};

/// Lines the lane scans for discovery; the server prints exactly two
/// relevant lines among its startup banner.
const DISCOVERY_LINES: usize = 128;

/// One HTTP request's response cap — message histories are the largest
/// payload and stay well under this in practice.
const MAX_RESPONSE_BYTES: usize = 16 * 1024 * 1024;

/// OpenCode's own typed refusal for an exhausted quota or usage
/// allowance. Its server maps `AI.Error.QuotaExceeded` onto this
/// `error.type`, so a served turn can classify the refusal from the
/// agent's error identity instead of guessing from vendor wording.
///
/// Classification only labels the terminal failure; it issues no retry.
/// The exhausted provider has already refused the turn, so any further
/// metered request would come from the caller's configured failover chain,
/// not from this adapter.
pub const OPENCODE_QUOTA_ERROR_TYPE: &str = "provider.quota";

/// How a turn failure surfaces in the message feed's terminal record.
#[derive(Clone, Debug)]
struct IdleRecord {
    outcome: String,
    /// The server's typed error class when it supplies one
    /// (OpenCode emits `provider.quota`, `provider.auth`, …).
    error_type: Option<String>,
    error: Option<String>,
}

/// The spawned server's discovery result.
struct ServerEndpoint {
    base: String,
    auth: String,
}

/// Minimal HTTP/1.1 client for the loopback server: one fresh connection
/// per request keeps the code free of connection-pool state, which matters
/// because a hung server must never poison a later request.
struct HttpClient {
    base: String,
    auth: String,
}

impl HttpClient {
    async fn call(
        &self,
        method: &str,
        path: &str,
        body: Option<&Value>,
    ) -> Result<Value, AdapterError> {
        let address = self
            .base
            .strip_prefix("http://")
            .unwrap_or(&self.base)
            .to_owned();
        let mut stream = tokio::net::TcpStream::connect(&address)
            .await
            .map_err(|error| {
                AdapterError::io("agent_protocol_failed", "session/transport", error)
            })?;
        let payload = body.map(|body| body.to_string()).unwrap_or_default();
        let request = format!(
            "{method} {path} HTTP/1.1\r\nHost: {address}\r\nAuthorization: Basic {}\r\nAccept: application/json\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{payload}",
            self.auth,
            payload.len(),
        );
        stream
            .write_all(request.as_bytes())
            .await
            .map_err(|error| {
                AdapterError::io("agent_protocol_failed", "session/transport", error)
            })?;
        let mut raw = Vec::new();
        let mut chunk = [0u8; 16 * 1024];
        loop {
            let read = stream.read(&mut chunk).await.map_err(|error| {
                AdapterError::io("agent_protocol_failed", "session/transport", error)
            })?;
            if read == 0 {
                break;
            }
            raw.extend_from_slice(&chunk[..read]);
            if raw.len() > MAX_RESPONSE_BYTES {
                return Err(AdapterError::permanent(
                    "agent_output_overflow",
                    "session/transport",
                ));
            }
        }
        parse_response(&raw)
    }
}

/// Split the raw response into status code and decoded body. Transfer
/// encodings the server uses — `chunked` — are decoded; a `close` framed
/// body arrives intact because the connection ends after the response.
fn parse_response(raw: &[u8]) -> Result<Value, AdapterError> {
    let split = raw
        .windows(4)
        .position(|window| window == b"\r\n\r\n")
        .ok_or_else(|| {
            AdapterError::permanent("agent_protocol_failed", "session/transport")
                .with_detail("response headers never terminated")
        })?;
    let head = String::from_utf8_lossy(&raw[..split]).to_string();
    let status: u16 = head
        .split_whitespace()
        .nth(1)
        .and_then(|code| code.parse().ok())
        .unwrap_or(0);
    let raw_body = &raw[split + 4..];
    let body = if head.to_lowercase().contains("transfer-encoding: chunked") {
        decode_chunked(raw_body)?
    } else {
        raw_body.to_vec()
    };
    if !(200..300).contains(&status) {
        let text = String::from_utf8_lossy(&body);
        let detail = format!(
            "server answered HTTP {status}: {}",
            bounds::bounded_prefix(&text, Some(4096))
        );
        // A 429/5xx body naming quota or capacity means the same thing a
        // lane error does: custody may fail over, but the effect stays
        // uncertain.
        if crate::error::quota_or_capacity_refusal(&text) || status == 429 {
            return Err(
                AdapterError::uncertain("agent_quota_exhausted", "session/transport")
                    .with_detail(detail),
            );
        }
        return Err(
            AdapterError::permanent("agent_protocol_failed", "session/transport")
                .with_detail(detail),
        );
    }
    if body.is_empty() {
        return Ok(Value::Null);
    }
    serde_json::from_slice(&body).map_err(|error| {
        AdapterError::permanent("agent_protocol_failed", "session/transport")
            .with_detail(format!("response body is not JSON: {error}"))
    })
}

/// Decode a chunked transfer body. Returns the unconsumed prefix when the
/// stream ends mid-chunk — the terminating `0`-chunk is the contract, and
/// anything else means the server dropped the connection early.
fn decode_chunked(raw: &[u8]) -> Result<Vec<u8>, AdapterError> {
    let mut decoded = Vec::with_capacity(raw.len());
    let mut cursor = raw;
    loop {
        let line_end = cursor
            .windows(2)
            .position(|window| window == b"\r\n")
            .ok_or_else(|| {
                AdapterError::permanent("agent_protocol_failed", "session/transport")
                    .with_detail("chunked body truncated inside a size line")
            })?;
        let size_text = String::from_utf8_lossy(&cursor[..line_end]);
        let size_text = size_text.split(';').next().unwrap_or("").trim();
        let size = usize::from_str_radix(size_text, 16).map_err(|_| {
            AdapterError::permanent("agent_protocol_failed", "session/transport")
                .with_detail(format!("invalid chunk size {size_text:?}"))
        })?;
        cursor = &cursor[line_end + 2..];
        if size == 0 {
            return Ok(decoded);
        }
        if cursor.len() < size + 2 {
            return Err(
                AdapterError::permanent("agent_protocol_failed", "session/transport")
                    .with_detail("chunked body truncated inside a chunk"),
            );
        }
        decoded.extend_from_slice(&cursor[..size]);
        cursor = &cursor[size + 2..];
    }
}

/// Merge the lane's declared MCP servers into the server configuration the
/// spec environment carries: `mcp.servers.<name>` gets a `local` entry with
/// the stdio command line and environment. ACP-style entries
/// (`{name, command, args, env:[{name,value}]}`) translate to the server
/// config shape; entries already in the server shape pass through.
fn merge_mcp_servers(env: &mut BTreeMap<String, std::ffi::OsString>, servers: &[Value]) {
    if servers.is_empty() {
        return;
    }
    let key = "OPENCODE_CONFIG_CONTENT";
    let mut document: Value = env
        .get(key)
        .and_then(|value| serde_json::from_str(&value.to_string_lossy()).ok())
        .unwrap_or_else(|| json!({}));
    let declared = document
        .pointer_mut("/mcp/servers")
        .and_then(Value::as_object_mut)
        .map(|_| true)
        .is_some();
    if !declared {
        document["mcp"] = json!({"servers": {}});
    }
    let registry = document["mcp"]["servers"].as_object_mut().unwrap();
    for server in servers {
        let Some(name) = server["name"].as_str() else {
            continue;
        };
        if server.get("type").and_then(Value::as_str) == Some("local") {
            registry.insert(name.to_owned(), server.clone());
            continue;
        }
        let mut command = vec![server["command"].as_str().unwrap_or_default().to_owned()];
        command.extend(
            server["args"]
                .as_array()
                .into_iter()
                .flatten()
                .filter_map(Value::as_str)
                .map(str::to_owned),
        );
        let environment: serde_json::Map<String, Value> = server["env"]
            .as_array()
            .into_iter()
            .flatten()
            .filter_map(|entry| {
                Some((
                    entry["name"].as_str()?.to_owned(),
                    Value::String(entry["value"].as_str()?.to_owned()),
                ))
            })
            .collect();
        registry.insert(
            name.to_owned(),
            json!({
                "type": "local",
                "command": command,
                "environment": environment,
                "enabled": true,
            }),
        );
    }
    env.insert(
        key.to_owned(),
        std::ffi::OsString::from(document.to_string()),
    );
}

/// Read the server's startup output until both discovery lines arrive:
/// `server listening on <url>` and `server password <secret>`.
async fn discover<R>(stdout: R, observer: SharedObserver) -> Result<ServerEndpoint, AdapterError>
where
    R: tokio::io::AsyncRead + Unpin,
{
    let mut lines = BufReader::new(stdout).lines();
    let mut base = None;
    let mut password: Option<String> = None;
    for _ in 0..DISCOVERY_LINES {
        let line = match lines.next_line().await {
            Ok(Some(line)) => line,
            Ok(None) => break,
            Err(error) => {
                return Err(AdapterError::io(
                    "agent_protocol_failed",
                    "session/open",
                    error,
                ));
            }
        };
        // Any line carrying the announcement's credential — the password
        // line itself or a later echo — reaches observers only with the
        // secret replaced.
        let redacted = match &password {
            Some(secret) => line.replace(secret.as_str(), "<redacted>"),
            None => line.clone(),
        };
        let observed: &[u8] = if redacted.starts_with("server password ") {
            b"server password <redacted>"
        } else {
            redacted.as_bytes()
        };
        if let Some(observer) = observer.as_ref() {
            observer.record("stdout", Direction::Received, observed);
        }
        if let Some(url) = line.strip_prefix("server listening on ") {
            let announced = url.trim();
            let authority = announced
                .strip_prefix("http://")
                .unwrap_or(announced)
                .split('/')
                .next()
                .unwrap_or_default();
            if !crate::transport::is_loopback_authority(authority) {
                return Err(
                    AdapterError::permanent("agent_protocol_failed", "session/open")
                        .with_detail(format!("announced endpoint is not loopback: {announced}")),
                );
            }
            base = Some(announced.to_owned());
        } else if let Some(secret) = line.strip_prefix("server password ") {
            password = Some(secret.trim().to_owned());
        }
        if base.is_some() && password.is_some() {
            break;
        }
    }
    let (base, secret) = match (base, password) {
        (Some(base), Some(secret)) => (base, secret),
        _ => {
            return Err(
                AdapterError::permanent("agent_protocol_failed", "session/open")
                    .with_detail("server never announced its endpoint and credential"),
            )
        }
    };
    Ok(ServerEndpoint {
        auth: base64(format!("opencode:{secret}").as_bytes()),
        base,
    })
}

/// The serve lane: drives the agent's native HTTP server.
pub struct ServeLane {
    pub max_output_bytes: Option<usize>,
    pub command: CommandSpec,
    pub mcp_servers: Vec<Value>,
    pub client_name: String,
    pub secrets: Arc<dyn SecretResolver>,
    pub observer: SharedObserver,
}

impl ServeLane {
    /// Spawn the server, discover its endpoint, create the session and
    /// apply the session-phase selectors.
    pub async fn open(
        &self,
        request: SessionRequest,
        events: mpsc::UnboundedSender<AgentEvent>,
    ) -> Result<ServeSession, AdapterError> {
        let mut env = crate::env::resolve(&self.command.env, &request.env, &*self.secrets).await?;
        merge_mcp_servers(&mut env, &self.mcp_servers);
        // A requested permission preset has no per-session API on this
        // lane; it is honoured only when the configuration the spec itself
        // injects verifiably denies every mutating tool — the strictest
        // possible reading of a read-only request. The proof runs on the
        // fully merged document before the server is even spawned:
        // anything else fails closed so the caller can fall back to a lane
        // that can enforce it.
        if let Some(permission) = request.config.permission.as_deref() {
            let locked = env
                .get("OPENCODE_CONFIG_CONTENT")
                .and_then(|value| serde_json::from_str::<Value>(&value.to_string_lossy()).ok());
            if permission != "read-only" || !locked.is_some_and(|doc| config_denies_mutation(&doc))
            {
                return Err(AdapterError::permanent(
                    "agent_permission_unsupported",
                    "session/configuration",
                ));
            }
        }
        let mut command = SupervisedChild::command(&self.command, &env, request.cwd.as_deref());
        let mut child = SupervisedChild::spawn(&mut command, stdio::PROTOCOL_IO)?;
        let (stdin, stdout, stderr) = stdio::protocol_halves(&mut child)?;
        drop(stdin);
        if let Some(stderr) = stderr {
            tokio::spawn(bounds::drain_stderr(
                stderr,
                self.observer.clone(),
                bounds::StderrTail::default(),
            ));
        }
        let endpoint = match discover(stdout, self.observer.clone()).await {
            Ok(endpoint) => endpoint,
            Err(error) => {
                let _ = child.terminate().await;
                return Err(error);
            }
        };
        let http = HttpClient {
            base: endpoint.base,
            auth: endpoint.auth,
        };
        let created = match http.call("POST", "/api/session", Some(&json!({}))).await {
            Ok(created) => created,
            Err(error) => {
                let _ = child.terminate().await;
                return Err(admission_error(error));
            }
        };
        let session_id = created["data"]["id"]
            .as_str()
            .unwrap_or_default()
            .to_owned();
        if session_id.is_empty() {
            let _ = child.terminate().await;
            return Err(
                AdapterError::permanent("agent_protocol_failed", "session/open")
                    .with_detail("server returned no session id"),
            );
        }
        let session = ServeSession {
            http,
            session_id: session_id.clone(),
            child: Mutex::new(child),
            events,
            max_output_bytes: self.max_output_bytes,
            cancelled: AtomicBool::new(false),
        };
        if let Some(model) = request.config.model.as_deref() {
            session.apply_model(model).await?;
        }
        Ok(session)
    }
}

/// The injected server config denies mutation when its permission rules
/// reject the editing actions, its tool table disables the mutation and
/// shell surfaces, and no MCP server stays enabled — a live MCP entry is a
/// mutation-capable surface the permission table cannot reach, so `enabled`
/// absent counts as enabled and fails closed — the audit posture expressed
/// as a verifiable document.
fn config_denies_mutation(document: &Value) -> bool {
    let permissions = [
        "edit",
        "bash",
        "webfetch",
        "doom_loop",
        "external_directory",
    ]
    .iter()
    .all(|rule| document["permission"][rule].as_str() == Some("deny"));
    let tools = [
        "write",
        "edit",
        "patch",
        "bash",
        "webfetch",
        "websearch",
        "task",
        "skill",
        "question",
    ]
    .iter()
    .all(|tool| document["tools"][tool].as_bool() == Some(false));
    let mcp = document["mcp"]["servers"]
        .as_object()
        .into_iter()
        .flatten()
        .all(|(_, server)| server["enabled"].as_bool() == Some(false));
    permissions && tools && mcp
}

/// One open session on a native server.
pub struct ServeSession {
    http: HttpClient,
    session_id: String,
    child: Mutex<SupervisedChild>,
    events: mpsc::UnboundedSender<AgentEvent>,
    max_output_bytes: Option<usize>,
    /// Set when a turn observed an interrupt; the poll loop treats the
    /// terminal record that follows as a cancellation.
    cancelled: AtomicBool,
}

impl ServeSession {
    pub fn session_id(&self) -> &str {
        &self.session_id
    }

    /// `provider/model` → `{model:{id, providerID}}`.
    async fn apply_model(&self, model: &str) -> Result<(), AdapterError> {
        let (provider, id) = model.split_once('/').unwrap_or(("opencode", model));
        let body = json!({"model": {"id": id, "providerID": provider}});
        self.http
            .call(
                "POST",
                &format!("/api/session/{}/model", self.session_id),
                Some(&body),
            )
            .await
            .map_err(|mut error| {
                error.stage = "session/set_model";
                admission_error(error)
            })?;
        Ok(())
    }

    /// Submit one turn: apply the requested model, post the prompt, then
    /// poll the message feed until the terminal `idle` record reports the
    /// outcome. Assistant text collected along the way becomes the turn's
    /// output.
    pub async fn prompt(&self, request: TurnRequest) -> Result<TurnOutcome, AdapterError> {
        if let Some(model) = request.model.as_deref() {
            self.apply_model(model).await?;
        }
        let marker = self
            .http
            .call(
                "POST",
                &format!("/api/session/{}/prompt", self.session_id),
                Some(&json!({"text": request.text})),
            )
            .await
            .map_err(admission_error)?;
        // Bound this turn by the prompt record's creation time, not its id:
        // the message feed paginates and the user record ages out of the
        // newest page on long turns.
        let marker_ms = marker["data"]["time"]["created"].as_u64().unwrap_or(0);
        // The poll is the turn: no internal deadline, the caller's own
        // supervision decides how long a viable turn may run.
        loop {
            tokio::time::sleep(std::time::Duration::from_millis(400)).await;
            let feed = self
                .http
                .call(
                    "GET",
                    &format!("/api/session/{}/message", self.session_id),
                    None,
                )
                .await?;
            let messages = feed["data"].as_array().cloned().unwrap_or_default();
            match terminal(&messages, marker_ms) {
                Some(IdleRecord { outcome, .. }) if outcome == "succeeded" => {
                    let (text, usage) = collect_output(&messages, marker_ms);
                    // No stream exists on this lane: deliver the completed
                    // text as one chunk and the terminal usage record so
                    // observers see the same event shape other lanes emit.
                    if !text.is_empty() {
                        let _ = self
                            .events
                            .send(AgentEvent::MessageChunk { text: text.clone() });
                    }
                    if let Some(usage) = usage.clone() {
                        let _ = self.events.send(AgentEvent::Usage { usage, raw: None });
                    }
                    return Ok(self.outcome(text, usage));
                }
                Some(IdleRecord { .. }) if self.cancelled.load(Ordering::Acquire) => {
                    return Err(AdapterError::permanent(
                        "agent_session_cancelled",
                        "session/prompt",
                    ));
                }
                Some(record) => {
                    // Only an assistant record that actually carries output
                    // or usage proves the model ran. The server creates an
                    // empty assistant envelope when it accepts the prompt,
                    // before a provider refusal can stop the turn, so
                    // counting the envelope alone would mark every quota
                    // refusal uncertain-effect and freeze the failover chain.
                    return Err(turn_failure(
                        record,
                        turn_produced_output(&messages, marker_ms),
                    ));
                }
                None => {}
            }
        }
    }

    /// Bound the collected text to the spec's output budget and attach the
    /// last reported usage.
    fn outcome(&self, text: String, usage: Option<TokenUsage>) -> TurnOutcome {
        let mut truncated = false;
        let text = match self.max_output_bytes {
            Some(budget) => {
                let prefix = bounds::bounded_prefix(&text, Some(budget)).to_owned();
                truncated = prefix.len() < text.len();
                prefix
            }
            None => text,
        };
        TurnOutcome {
            stop_reason: "end_turn".to_owned(),
            text,
            truncated,
            exit_code: None,
            usage,
        }
    }

    /// Interrupt the in-flight turn; the poll loop reports the terminal
    /// record that follows as the failed outcome it is.
    pub async fn cancel(&self) -> Result<(), AdapterError> {
        self.cancelled.store(true, Ordering::Release);
        self.http
            .call(
                "POST",
                &format!("/api/session/{}/interrupt", self.session_id),
                Some(&json!({})),
            )
            .await?;
        Ok(())
    }

    /// The server process owns the session store; terminate it.
    pub async fn close(&self) -> Result<(), AdapterError> {
        let mut child = self.child.lock().await;
        child.terminate().await?;
        Ok(())
    }
}

/// A record's creation time in milliseconds; records without a timestamp
/// count as the newest so they stay visible.
fn created_ms(message: &Value) -> u64 {
    message["time"]["created"].as_u64().unwrap_or(u64::MAX)
}

/// Whether a turn produced billable output before its terminal record.
///
/// The server opens an assistant envelope as soon as it accepts the prompt.
/// A provider refusal can leave that envelope empty and without token usage:
/// nothing was inferred, so it must not be read as a served turn. Replay
/// safety depends on this distinction — `uncertain_effect` is only for a
/// turn whose output may already exist.
fn turn_produced_output(messages: &[Value], marker_ms: u64) -> bool {
    messages.iter().any(|message| {
        created_ms(message) >= marker_ms
            && message["type"].as_str() == Some("assistant")
            && assistant_carries_output(message)
    })
}

/// One assistant record's evidence of work: a non-empty text part, a tool
/// or reasoning part, or reported token usage — on the record or on any of
/// its parts. The detector owns OpenCode's token shape, so a present usage
/// object counts as reported usage. An empty text part, a null usage table
/// and bookkeeping parts such as `step-start`/`step-finish` markers are the
/// refusal envelope, not a completed inference.
fn assistant_carries_output(message: &Value) -> bool {
    if let Some(text) = message["content"].as_str() {
        if !text.is_empty() {
            return true;
        }
    }
    for part in message["content"].as_array().into_iter().flatten() {
        match part["type"].as_str() {
            // Step markers and other bookkeeping parts are not evidence on
            // their own — a marker-only envelope is the refusal shape.
            Some("step-start") | Some("step-finish") => {}
            // An empty text part carries no output either.
            Some("text") if part["text"].as_str().is_none_or(|text| text.is_empty()) => {}
            // Every other part type (tool, reasoning, patch, file, agent,
            // subtask, or one this build does not know) is evidence by
            // default: treating produced work as "no output" would let
            // custody replay a request over real side effects.
            Some(_) => return true,
            None => {}
        }
        // Any part the server attached reported usage to is evidence.
        if token_usage_detector::agent::from_opencode_tokens(&part["tokens"]).is_some() {
            return true;
        }
    }
    token_usage_detector::agent::from_opencode_tokens(&message["tokens"]).is_some()
}

/// A quota refusal that answered an admission call — session create,
/// model select or the prompt POST — refused before any turn could run.
/// Clearing the effect doubt lets custody replay the request on the next
/// candidate; every other error is returned unchanged.
fn admission_error(mut error: AdapterError) -> AdapterError {
    if error.code == "agent_quota_exhausted" {
        error.uncertain_effect = false;
    }
    error
}

/// Whether a served turn's failure names a quota, capacity or
/// model-availability refusal.
///
/// OpenCode's typed `provider.quota` is authoritative — the provider's own
/// `AI.Error.QuotaExceeded` refusal arrives with that class even when the
/// message wording changes. Every other served refusal keeps the shared
/// text matcher so custody still fails over on untyped quota wording.
pub fn is_quota_refusal(error_type: Option<&str>, detail: &str) -> bool {
    error_type == Some(OPENCODE_QUOTA_ERROR_TYPE) || crate::error::quota_or_capacity_refusal(detail)
}

/// Build the typed error for a failed turn. `served` says whether the
/// turn produced assistant output before the terminal refusal: a typed or
/// textual quota refusal with no output is an admission rejection —
/// nothing ran, so custody may replay it on the next candidate. A refusal
/// after output cannot prove no billable work ran, so it stays
/// uncertain-effect. This adapter sends no retry of its own; the
/// exhausted provider has already refused the turn, and any next
/// candidate comes from the caller's configured chain.
fn turn_failure(record: IdleRecord, served: bool) -> AdapterError {
    let detail = match (record.error_type.as_deref(), record.error.as_deref()) {
        (Some(kind), Some(message)) if !message.is_empty() => format!("{kind}: {message}"),
        (Some(kind), _) => kind.to_owned(),
        (None, Some(message)) => message.to_owned(),
        (None, None) => "turn failed".to_owned(),
    };
    if is_quota_refusal(record.error_type.as_deref(), &detail) {
        return if served {
            AdapterError::uncertain("agent_quota_exhausted", "session/prompt").with_detail(detail)
        } else {
            AdapterError::retryable("agent_quota_exhausted", "session/prompt").with_detail(detail)
        };
    }
    AdapterError::permanent("agent_protocol_failed", "session/prompt").with_detail(detail)
}

/// The terminal `idle` record covering this turn, if it has arrived yet.
/// Messages stream newest-first; only records at or after `marker_ms` (the
/// prompt's creation time) belong to this turn — a stale `idle` from an
/// earlier turn cannot close the poll.
fn terminal(messages: &[Value], marker_ms: u64) -> Option<IdleRecord> {
    for message in messages {
        if created_ms(message) < marker_ms {
            break;
        }
        if message["type"].as_str() == Some("idle") {
            let error = &message["error"];
            // Current servers emit a typed flat object; older records
            // wrapped the same fields in a named-error `data` object.
            let error_type = error["type"]
                .as_str()
                .or_else(|| error["name"].as_str())
                .map(str::to_owned);
            let detail = error["message"]
                .as_str()
                .or_else(|| error["data"]["message"].as_str())
                .or_else(|| error.as_str())
                .map(str::to_owned);
            return Some(IdleRecord {
                outcome: message["outcome"].as_str().unwrap_or("failed").to_owned(),
                error_type,
                error: detail,
            });
        }
    }
    None
}

/// Accumulate assistant text and the last usage record from the completed
/// turn — the assistant records created at or after `marker_ms`.
fn collect_output(messages: &[Value], marker_ms: u64) -> (String, Option<TokenUsage>) {
    let mut text = String::new();
    let mut usage = None;
    // Feed order is newest-first; walk it in reverse for chronological
    // output. Records older than the marker belong to earlier turns.
    let mut parts: Vec<&Value> = messages
        .iter()
        .filter(|message| {
            created_ms(message) >= marker_ms && message["type"].as_str() == Some("assistant")
        })
        .collect();
    parts.reverse();
    for message in parts {
        for part in message["content"].as_array().into_iter().flatten() {
            if part["type"].as_str() == Some("text") {
                if let Some(chunk) = part["text"].as_str() {
                    text.push_str(chunk);
                }
            }
        }
        if let Some(reported) =
            token_usage_detector::agent::from_opencode_tokens(&message["tokens"])
        {
            usage = Some(reported);
        }
    }
    (text, usage)
}

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

    #[test]
    fn chunked_bodies_decode() {
        let raw =
            b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n7\r\n{\"a\":1}\r\n0\r\n\r\n";
        let value = parse_response(raw).unwrap();
        assert_eq!(value, json!({"a": 1}));
    }

    #[test]
    fn chunked_truncation_fails_closed() {
        let raw = b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n8\r\n{\"a\"";
        assert!(parse_response(raw).is_err());
    }

    #[test]
    fn non_2xx_is_a_protocol_failure() {
        let raw = b"HTTP/1.1 404 Not Found\r\nContent-Length: 9\r\n\r\nnot found";
        let error = parse_response(raw).unwrap_err();
        assert_eq!(error.code, "agent_protocol_failed");
    }

    #[test]
    fn base64_matches_rfc_vectors() {
        assert_eq!(base64(b""), "");
        assert_eq!(base64(b"f"), "Zg==");
        assert_eq!(base64(b"fo"), "Zm8=");
        assert_eq!(base64(b"foo"), "Zm9v");
        assert_eq!(base64(b"opencode:secret"), "b3BlbmNvZGU6c2VjcmV0");
    }

    #[tokio::test]
    async fn discovery_never_forwards_the_password_to_observers() {
        struct Seen(std::sync::Mutex<Vec<Vec<u8>>>);
        impl crate::observe::Observer for Seen {
            fn record(&self, _source: &'static str, _dir: Direction, bytes: &[u8]) {
                self.0.lock().unwrap().push(bytes.to_vec());
            }
        }
        let seen = Arc::new(Seen(std::sync::Mutex::new(Vec::new())));
        let loopback = format!("{}.{}.{}.{}", 127, 0, 0, 1);
        let feed = format!(
            "server password s3cret-value\nnote: s3cret-value echoed\nserver listening on http://{loopback}:{}\n",
            9000
        );
        let endpoint = discover(feed.as_bytes(), Some(seen.clone())).await.unwrap();
        assert_eq!(endpoint.auth, base64(b"opencode:s3cret-value"));
        for line in seen.0.lock().unwrap().iter() {
            assert!(
                !line.windows(12).any(|w| w == b"s3cret-value"),
                "observer received the raw password line"
            );
        }
        assert!(seen
            .0
            .lock()
            .unwrap()
            .iter()
            .any(|l| l == b"note: <redacted> echoed"));
        assert!(seen
            .0
            .lock()
            .unwrap()
            .iter()
            .any(|l| l == b"server password <redacted>"));
    }

    #[test]
    fn read_only_proof_fails_closed_on_live_mcp_servers() {
        let locked = json!({
            "permission": {
                "edit": "deny", "bash": "deny", "webfetch": "deny",
                "doom_loop": "deny", "external_directory": "deny",
            },
            "tools": {
                "write": false, "edit": false, "patch": false, "bash": false,
                "webfetch": false, "websearch": false, "task": false,
                "skill": false, "question": false,
            },
        });
        assert!(config_denies_mutation(&locked));
        // An enabled MCP entry is a mutation surface the tool table never
        // reaches — explicit or defaulted, it voids the read-only proof.
        let mut live = locked.clone();
        live["mcp"] = json!({"servers": {"audit": {"enabled": true}}});
        assert!(!config_denies_mutation(&live));
        let mut defaulted = locked.clone();
        defaulted["mcp"] = json!({"servers": {"audit": {"type": "local"}}});
        assert!(!config_denies_mutation(&defaulted));
        let mut off = locked.clone();
        off["mcp"] = json!({"servers": {"audit": {"enabled": false}}});
        assert!(config_denies_mutation(&off));
    }

    #[test]
    fn acp_style_mcp_entries_translate_to_server_shape() {
        let mut env = BTreeMap::new();
        env.insert(
            "OPENCODE_CONFIG_CONTENT".to_owned(),
            std::ffi::OsString::from(r#"{"enabled_providers":["opencode"]}"#),
        );
        merge_mcp_servers(
            &mut env,
            &[json!({
                "type": "stdio", "name": "audit",
                "command": "/bin/tool", "args": ["serve", "--run", "r1"],
                "env": [{"name": "K", "value": "V"}],
            })],
        );
        let document: Value =
            serde_json::from_str(&env["OPENCODE_CONFIG_CONTENT"].to_string_lossy()).unwrap();
        assert_eq!(document["enabled_providers"][0], "opencode");
        let entry = &document["mcp"]["servers"]["audit"];
        assert_eq!(entry["type"], "local");
        assert_eq!(
            entry["command"],
            json!(["/bin/tool", "serve", "--run", "r1"])
        );
        assert_eq!(entry["environment"]["K"], "V");
        assert_eq!(entry["enabled"], true);
    }

    const OPENCODE_QUOTA_FIXTURE: &str = include_str!(
        "../../crates/a4aa-opencode/tests/fixtures/opencode-quota-refusal-synthetic.json"
    );

    #[test]
    fn synthetic_opencode_quota_refusal_is_agent_quota_exhausted() {
        let fixture: Value = serde_json::from_str(OPENCODE_QUOTA_FIXTURE).expect("fixture json");
        assert_eq!(
            fixture["synthetic"], true,
            "fixture must be labeled synthetic"
        );
        let messages = vec![fixture["record"].clone()];
        let record = terminal(&messages, 0).expect("synthetic terminal record");
        assert_eq!(
            record.error_type.as_deref(),
            Some(OPENCODE_QUOTA_ERROR_TYPE)
        );
        // With no assistant output this turn, the refusal is an admission
        // rejection: retryable and certain — custody replays it.
        let error = turn_failure(record.clone(), false);
        assert_eq!(error.code, "agent_quota_exhausted");
        assert!(error.retryable);
        assert!(!error.uncertain_effect);
        // After billable output the same refusal stays uncertain-effect.
        let served = turn_failure(record, true);
        assert_eq!(served.code, "agent_quota_exhausted");
        assert!(served.uncertain_effect);
        assert!(error
            .detail
            .as_deref()
            .is_some_and(|detail| detail.starts_with("provider.quota: ")));
    }

    #[test]
    fn empty_assistant_envelope_keeps_a_quota_refusal_certain_and_retryable() {
        // Captured shape: OpenCode's free tier accepts the prompt, creates an
        // assistant envelope with no content and no usage, then posts the
        // terminal idle record carrying the typed 429 refusal.
        let marker = 1_793_006_888_649u64;
        let messages = vec![
            json!({
                "type": "idle", "outcome": "failed",
                "time": {"created": marker + 1495},
                "error": {
                    "type": "provider.quota",
                    "message": "Free usage limit reached."
                }
            }),
            json!({
                "type": "assistant",
                "time": {"created": marker + 43},
                "content": [],
                "tokens": null,
                "error": {
                    "type": "provider.quota",
                    "message": "Free usage limit reached."
                }
            }),
            json!({"type": "user", "time": {"created": marker}}),
        ];
        assert!(!turn_produced_output(&messages, marker));
        let record = terminal(&messages, marker).expect("terminal record");
        let error = turn_failure(record, turn_produced_output(&messages, marker));
        assert_eq!(error.code, "agent_quota_exhausted");
        assert!(error.retryable, "custody must be allowed to replay");
        assert!(!error.uncertain_effect, "nothing ran before the refusal");
    }

    #[test]
    fn assistant_output_or_usage_still_marks_the_turn_served() {
        let marker = 100u64;
        let text = vec![json!({
            "type": "assistant", "time": {"created": 110},
            "content": [{"type": "text", "text": "OK"}]
        })];
        assert!(turn_produced_output(&text, marker));
        let tool = vec![json!({
            "type": "assistant", "time": {"created": 110},
            "content": [{"type": "tool", "tool": "bash"}]
        })];
        assert!(turn_produced_output(&tool, marker));
        let usage = vec![json!({
            "type": "assistant", "time": {"created": 110},
            "content": [], "tokens": {"input": 9, "output": 0}
        })];
        assert!(turn_produced_output(&usage, marker));
        // Step markers and other bookkeeping parts are not output: an
        // envelope the server populates with step bookkeeping before a
        // provider refusal must not read as a served turn.
        let markers = vec![json!({
            "type": "assistant", "time": {"created": 110},
            "content": [{"type": "step-start"}, {"type": "step-finish"}],
            "tokens": null
        })];
        assert!(!turn_produced_output(&markers, marker));
        // A part type this build does not enumerate (patch, snapshot,
        // file, agent, subtask, …) is evidence by default — assuming "no
        // output" would let custody replay over real side effects.
        let patch_only = vec![json!({
            "type": "assistant", "time": {"created": 110},
            "content": [{"type": "patch", "patch": "@@"}], "tokens": null
        })];
        assert!(turn_produced_output(&patch_only, marker));
        // A part the server attached reported usage to is evidence
        // whatever its type — the detector owns the token shape.
        let part_usage = vec![json!({
            "type": "assistant", "time": {"created": 110},
            "content": [{"type": "step-finish", "tokens": {"input": 9}}]
        })];
        assert!(turn_produced_output(&part_usage, marker));
        let empty_text = vec![json!({
            "type": "assistant", "time": {"created": 110},
            "content": [{"type": "text", "text": ""}], "tokens": null
        })];
        assert!(!turn_produced_output(&empty_text, marker));
        let before_marker = vec![json!({
            "type": "assistant", "time": {"created": 90},
            "content": [{"type": "text", "text": "earlier turn"}]
        })];
        assert!(!turn_produced_output(&before_marker, marker));
    }

    #[test]
    fn synthetic_opencode_auth_failure_keeps_the_protocol_classification() {
        let messages = vec![json!({
            "type": "idle", "outcome": "failed",
            "error": {"type": "provider.auth", "message": "This API key has expired."}
        })];
        let record = terminal(&messages, 0).unwrap();
        let error = turn_failure(record, false);
        assert_eq!(error.code, "agent_protocol_failed");
        assert!(!error.retryable);
        assert!(!error.uncertain_effect);
    }

    #[test]
    fn untyped_quota_wording_still_classifies() {
        let messages = vec![json!({
            "type": "idle", "outcome": "failed",
            "error": {"message": "AI.Error: Free usage limit reached."}
        })];
        let record = terminal(&messages, 0).unwrap();
        assert_eq!(turn_failure(record, false).code, "agent_quota_exhausted");
    }

    #[test]
    fn named_error_data_message_is_read() {
        let messages = vec![json!({
            "type": "idle", "outcome": "failed",
            "error": {
                "name": "AI.Error.QuotaExceeded",
                "data": {"message": "Free usage limit reached."}
            }
        })];
        let record = terminal(&messages, 0).unwrap();
        assert_eq!(record.error_type.as_deref(), Some("AI.Error.QuotaExceeded"));
        assert_eq!(turn_failure(record, false).code, "agent_quota_exhausted");
    }

    #[test]
    fn the_terminal_record_is_the_idle_at_or_after_the_prompt_marker() {
        // Newest-first feed; marker 100 separates this turn from the last.
        let messages = vec![
            json!({"type": "idle", "outcome": "succeeded", "time": {"created": 120}}),
            json!({"id": "m1", "type": "assistant", "time": {"created": 110},
                   "content": [{"type": "text", "text": "hi"}]}),
            json!({"id": "u1", "type": "user", "time": {"created": 100}}),
            json!({"type": "idle", "outcome": "failed", "time": {"created": 90}}),
        ];
        let record = terminal(&messages, 100).unwrap();
        assert_eq!(record.outcome, "succeeded");
        // No idle at or after the marker: this turn is still running, and
        // the older turn's terminal record must not close the poll.
        let running = vec![
            json!({"id": "a1", "type": "assistant", "time": {"created": 105}}),
            json!({"id": "u2", "type": "user", "time": {"created": 100}}),
            json!({"type": "idle", "outcome": "succeeded", "time": {"created": 90}}),
        ];
        assert!(terminal(&running, 100).is_none());
        // A paginated page whose user record aged out still yields the
        // turn's output and usage.
        let (text, usage) = collect_output(&messages[..2], 100);
        assert_eq!(text, "hi");
        assert!(usage.is_none());
        let with_tokens = vec![
            json!({"type": "idle", "outcome": "succeeded", "time": {"created": 120}}),
            json!({"id": "m1", "type": "assistant", "time": {"created": 110},
                   "content": [{"type": "text", "text": "hi"}],
                   "tokens": {"input": 10, "output": 4, "reasoning": 1,
                              "cache": {"read": 6, "write": 2}}}),
            json!({"id": "u1", "type": "user", "time": {"created": 100}}),
        ];
        let (_, usage) = collect_output(&with_tokens, 100);
        let usage = usage.unwrap();
        assert_eq!(usage.input, Some(10));
        assert_eq!(usage.cached_input, Some(6));
    }
}