hotl 0.7.1

//! `hotl acp` — the ACP-shaped protocol surface (M4).
//!
//! A JSON-RPC 2.0 line protocol over stdio that drives the *same* engine the
//! REPL does: an editor or orchestrator is just another client of
//! `SessionHandle`. One session per connection (process-per-session — the
//! orchestrator pattern). Model output and tool status
//! arrive as `session/update` notifications carrying a `schema_version`
//! (Tier-1 stable, not a side-channel); permission asks are
//! `session/request_permission` round-trips to the client.

use std::collections::{HashMap, VecDeque};
use std::sync::Arc;

use hotl_engine::{AskReply, EngineEvent, Outcome, SessionHandle};
use serde_json::{json, Value};
use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader};
use tokio::sync::{mpsc, oneshot, Mutex};

/// Stable schema version of `session/update` / prompt-result payloads (an MD
/// Tier-1 contract — bump only on a breaking change).
pub const UPDATE_SCHEMA_VERSION: u32 = 1;
pub const PROTOCOL_VERSION: &str = "0.1";

type Writer = Arc<Mutex<Box<dyn AsyncWrite + Send + Unpin>>>;
type Pending = Arc<std::sync::Mutex<HashMap<u64, oneshot::Sender<AskReply>>>>;
/// In-flight `session/request_question` round-trips, parallel to `Pending`.
/// Disjoint ids from `Pending`: both draw from the same per-session `req_id`
/// counter in `drain_events`, so an id never means two different things.
type PendingQuestions =
    Arc<std::sync::Mutex<HashMap<u64, oneshot::Sender<hotl_types::QuestionAnswer>>>>;

/// Map an ACP client's permission `result` to an `AskReply` (T1/§2b): a client
/// may `{"allow":true}`, `{"allow":false,"message":"…"}`, provide edited
/// `{"input":{…}}` (AllowEdited), or answer as the tool `{"respond":"…"}`.
fn ask_reply_from_result(result: Option<&Value>) -> AskReply {
    let Some(r) = result else {
        return AskReply::Deny { message: None };
    };
    if let Some(content) = r.get("respond").and_then(Value::as_str) {
        return AskReply::Respond {
            content: content.to_string(),
        };
    }
    if let Some(input) = r.get("input") {
        return AskReply::AllowEdited {
            input: input.clone(),
        };
    }
    if r.get("allow").and_then(Value::as_bool) == Some(true) {
        return AskReply::Allow;
    }
    AskReply::Deny {
        message: r.get("message").and_then(Value::as_str).map(String::from),
    }
}

/// Map an ACP client's `session/request_question` `result` to a
/// `QuestionAnswer`: `{"selected":["label", ...]}` or `{"freeText":"..."}`.
/// Anything else (including a missing/malformed result, or a client that
/// hangs up before replying) resolves to `NoHuman` — never a hang, and
/// never anything that could be mistaken for a permission grant.
fn question_answer_from_result(result: Option<&Value>) -> hotl_types::QuestionAnswer {
    let Some(r) = result else {
        return hotl_types::QuestionAnswer::NoHuman;
    };
    if let Some(text) = r.get("freeText").and_then(Value::as_str) {
        return hotl_types::QuestionAnswer::FreeText(text.to_string());
    }
    if let Some(arr) = r.get("selected").and_then(Value::as_array) {
        let labels: Vec<String> = arr
            .iter()
            .filter_map(Value::as_str)
            .map(String::from)
            .collect();
        if !labels.is_empty() {
            return hotl_types::QuestionAnswer::Selected(labels);
        }
    }
    hotl_types::QuestionAnswer::NoHuman
}

/// JSON-RPC ids of in-flight prompt requests, answered in order on TurnDone
/// (the engine queues overlapping prompts and finishes them FIFO).
type PendingPrompt = Arc<std::sync::Mutex<VecDeque<Value>>>;

/// What a client asked the factory to produce.
pub enum SessionSpec {
    New {
        name: Option<String>,
    },
    Load {
        session_id: String,
        name: Option<String>,
    },
}

/// A session the factory opened: the handle plus its display name (the one
/// just given, or inherited from the resumed chain).
pub struct SessionOpen {
    pub handle: SessionHandle,
    pub name: Option<String>,
    /// This session's effective permission mode at open (inherited-and-coerced
    /// on resume, else the configured default).
    /// INVARIANT: equals what the engine's `Rules` will actually enforce.
    /// Enforced by `tests/acp_protocol.rs::the_session_reports_its_effective_mode`.
    pub mode: String,
}

/// Builds a session per the client's request. The real binary wires engine
/// deps here; tests inject a scripted-provider session.
pub type SessionFactory = Box<dyn FnMut(SessionSpec) -> Result<SessionOpen, String> + Send>;

/// One skill as `initialize` advertises it. The description is client-facing
/// only — front ends render it in the `/` completion menu. Content still
/// enters context only when asked for: the always-sent tool description
/// (`hotl_tools::skills::SkillTool`'s `describe`) omits it — the model sees
/// it only when it calls the skill tool itself.
#[derive(Debug, Clone)]
pub struct SkillInfo {
    pub name: String,
    pub description: String,
}

/// What `serve` advertises to a client before any session exists. Bundled
/// rather than passed positionally: `serve` is already at the argument count
/// where this workspace reaches for `#[allow(clippy::too_many_arguments)]`.
#[derive(Debug, Clone)]
pub struct ServerInfo {
    pub skills: Vec<SkillInfo>,
    /// The configured permission mode a new session starts in, already run
    /// through `enforced_mode`. Clients render it; they never infer it.
    pub default_mode: String,
    /// Model context window in tokens — what a UI's fullness gauge divides by.
    pub context_window: u64,
    /// The session's primary model — prices `turn_done`'s `usage.cost_usd`
    /// (Task 5 cost telemetry). One process serves one model for its whole
    /// lifetime (process-per-session), so this is fixed for every
    /// `session/new` and `session/load` this connection sees; a fallback
    /// model used mid-turn is still priced as this one (see
    /// `wire::usage_frame`).
    pub model: String,
}

/// Drive the protocol over one connection until the client hangs up.
pub async fn serve(
    read: impl AsyncRead + Send + Unpin + 'static,
    write: impl AsyncWrite + Send + Unpin + 'static,
    mut factory: SessionFactory,
    info: ServerInfo,
) {
    let writer: Writer = Arc::new(Mutex::new(Box::new(write)));
    let pending: Pending = Arc::new(std::sync::Mutex::new(HashMap::new()));
    let pending_questions: PendingQuestions = Arc::new(std::sync::Mutex::new(HashMap::new()));
    let pending_prompt: PendingPrompt = Arc::new(std::sync::Mutex::new(VecDeque::new()));
    let mut next_id: u64 = 1;
    let mut session: Option<SessionState> = None;

    let mut lines = BufReader::new(read).lines();
    while let Ok(Some(line)) = lines.next_line().await {
        let Ok(msg) = serde_json::from_str::<Value>(&line) else {
            continue; // unparseable frame, no id to reply to
        };
        // A client response to one of our permission requests, or one of our
        // questions (disjoint id spaces — a hit in one map means a miss in
        // the other, so trying both is safe and order-independent)?
        if msg.get("method").is_none() {
            if let Some(id) = msg.get("id").and_then(Value::as_u64) {
                if let Some(reply) = pending_questions
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .remove(&id)
                {
                    let _ = reply.send(question_answer_from_result(msg.get("result")));
                } else if let Some(reply) = pending
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .remove(&id)
                {
                    let _ = reply.send(ask_reply_from_result(msg.get("result")));
                }
            }
            continue;
        }
        handle_request(
            &msg,
            &writer,
            &mut factory,
            &mut session,
            &pending,
            &pending_questions,
            &pending_prompt,
            &mut next_id,
            &info,
        )
        .await;
    }
}

struct SessionState {
    id: String,
    handle: SessionHandle,
    drain: tokio::task::JoinHandle<()>,
}

#[allow(clippy::too_many_arguments)]
async fn handle_request(
    msg: &Value,
    writer: &Writer,
    factory: &mut SessionFactory,
    session: &mut Option<SessionState>,
    pending: &Pending,
    pending_questions: &PendingQuestions,
    pending_prompt: &PendingPrompt,
    next_id: &mut u64,
    info: &ServerInfo,
) {
    let id = msg.get("id").cloned().unwrap_or(Value::Null);
    match msg.get("method").and_then(Value::as_str).unwrap_or("") {
        "initialize" => {
            // `skills` lets a front end resolve `/<skill>` itself and build a
            // completion menu — the roster is server-side knowledge, so the
            // client never has to walk the config dirs to know what a slash
            // could mean. Descriptions ride this response only; they are
            // never part of any prompt.
            let skills: Vec<Value> = info
                .skills
                .iter()
                .map(|s| json!({"name": s.name, "description": s.description}))
                .collect();
            // `defaultMode` and `contextWindow` are server-side truth a client
            // renders rather than guesses: the badge used to read "ask" while
            // the shipped default ran "auto" (evaluation §5.7).
            reply_ok(
                writer,
                id,
                json!({
                    "protocolVersion": PROTOCOL_VERSION,
                    "schemaVersion": UPDATE_SCHEMA_VERSION,
                    "skills": skills,
                    "defaultMode": info.default_mode,
                    "contextWindow": info.context_window,
                }),
            )
            .await;
        }
        method @ ("session/new" | "session/load") => {
            let name = match msg.pointer("/params/name") {
                None | Some(Value::Null) => None,
                Some(v) => match v.as_str().and_then(hotl_types::normalize_session_name) {
                    Some(n) => Some(n),
                    None => {
                        return reply_err(
                            writer,
                            id,
                            "params.name must be 1–64 chars after trimming",
                        )
                        .await
                    }
                },
            };
            let spec = if method == "session/load" {
                match msg.pointer("/params/sessionId").and_then(Value::as_str) {
                    Some(sid) => SessionSpec::Load {
                        session_id: sid.to_string(),
                        name,
                    },
                    None => {
                        return reply_err(writer, id, "session/load requires params.sessionId")
                            .await
                    }
                }
            } else {
                SessionSpec::New { name }
            };
            match factory(spec) {
                Ok(open) => {
                    // Captured before `open.handle` moves into `start_session`.
                    let mode = open.mode;
                    // Replacing a session: interrupt its in-flight turn (its
                    // events are about to stop rendering anywhere — it must
                    // not keep running tools invisibly in the shared cwd),
                    // stop its drain task, and drop its parked state — a
                    // dropped ask sender reads as a deny to the old engine,
                    // and a stale prompt id must never be answered with the
                    // new session's first TurnDone.
                    if let Some(old) = session.take() {
                        old.handle.interrupt();
                        old.drain.abort();
                        pending
                            .lock()
                            .unwrap_or_else(std::sync::PoisonError::into_inner)
                            .clear();
                        pending_questions
                            .lock()
                            .unwrap_or_else(std::sync::PoisonError::into_inner)
                            .clear();
                        pending_prompt
                            .lock()
                            .unwrap_or_else(std::sync::PoisonError::into_inner)
                            .clear();
                    }
                    let state = start_session(
                        open.handle,
                        writer.clone(),
                        pending.clone(),
                        pending_questions.clone(),
                        pending_prompt.clone(),
                        next_id,
                        info.model.clone(),
                    );
                    // Resume auto-continuation (M4/#8): a loaded projection
                    // that ends mid-turn (user prompt or unanswered tool
                    // results) picks the work back up; the engine no-ops when
                    // there is nothing to continue.
                    if method == "session/load" {
                        state.handle.continue_turn().await;
                    }
                    let sid = state.id.clone();
                    *session = Some(state);
                    // The session's own effective mode rides the open result:
                    // a client's handshake runs before it takes the screen and
                    // wants the seed synchronously, rather than waiting for a
                    // notification that only fires on a *change*.
                    reply_ok(
                        writer,
                        id,
                        json!({"sessionId": sid, "name": open.name, "mode": mode}),
                    )
                    .await;
                }
                Err(e) => reply_err(writer, id, &e).await,
            }
        }
        "session/prompt" => {
            let Some(state) = session.as_ref() else {
                return reply_err(writer, id, "no session — call session/new first").await;
            };
            let Some(text) = msg.pointer("/params/text").and_then(Value::as_str) else {
                return reply_err(writer, id, "session/prompt requires params.text").await;
            };
            // Stash the id; the drain task answers it on TurnDone so the read
            // loop stays free to service permission responses meanwhile.
            pending_prompt
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .push_back(id);
            state.handle.prompt(text.to_string()).await;
        }
        "session/rename" => {
            let Some(state) = session.as_ref() else {
                return reply_err(writer, id, "no session — call session/new first").await;
            };
            let Some(name) = msg
                .pointer("/params/name")
                .and_then(Value::as_str)
                .and_then(hotl_types::normalize_session_name)
            else {
                return reply_err(
                    writer,
                    id,
                    "session/rename requires params.name (1–64 chars after trimming)",
                )
                .await;
            };
            state.handle.rename(name).await;
            reply_ok(writer, id, json!({"ok": true})).await;
        }
        "session/set_mode" => {
            let Some(state) = session.as_ref() else {
                return reply_err(writer, id, "no session — call session/new first").await;
            };
            let Some(mode) = msg
                .pointer("/params/mode")
                .and_then(Value::as_str)
                .and_then(hotl_tools::rules::PermissionMode::from_str)
            else {
                return reply_err(
                    writer,
                    id,
                    "session/set_mode requires params.mode (ask | auto | plan | dontask)",
                )
                .await;
            };
            // The *effective* mode, post-coercion: a `security-enforced` build
            // forces Auto→Ask inside `Rules::with_mode`, so acking the
            // requested mode would leave a client's badge reading `auto` for a
            // session actually running `ask` — the §5.7 lie in reverse.
            let effective = hotl_tools::rules::enforced_mode(mode);
            state.handle.set_mode(mode).await;
            let session_id = state.id.clone();
            reply_ok(writer, id, json!({"ok": true, "mode": effective.as_str()})).await;
            // Broadcast, not just ack: a mode changed by *any* client (or by
            // resume inheritance) has to reach every attached surface, and
            // `session/update` needs no id-plumbing in a client's loop.
            notify(
                writer,
                &session_id,
                json!({"type": "mode_changed", "mode": effective.as_str()}),
            )
            .await;
        }
        "session/steer" => {
            let Some(state) = session.as_ref() else {
                return reply_err(writer, id, "no session — call session/new first").await;
            };
            let Some(text) = msg.pointer("/params/text").and_then(Value::as_str) else {
                return reply_err(writer, id, "session/steer requires params.text").await;
            };
            state.handle.steer(text.to_string()).await;
            reply_ok(writer, id, json!({"queued": true})).await;
        }
        "session/cancel" => {
            if let Some(state) = session.as_ref() {
                state.handle.interrupt();
            }
            reply_ok(writer, id, json!({"cancelled": true})).await;
        }
        other => reply_err(writer, id, &format!("unknown method `{other}`")).await,
    }
}

#[allow(clippy::too_many_arguments)]
fn start_session(
    mut handle: SessionHandle,
    writer: Writer,
    pending: Pending,
    pending_questions: PendingQuestions,
    pending_prompt: PendingPrompt,
    next_id: &mut u64,
    model: String,
) -> SessionState {
    let id = format!("acp-{}", *next_id);
    // Permission/question request ids for this session are disjoint from
    // every other id.
    let req_id_seed = *next_id * 1_000_000;
    *next_id += 1;
    let events = std::mem::replace(&mut handle.events, mpsc::channel(1).1);
    let sid = id.clone();
    let drain = tokio::spawn(drain_events(
        events,
        writer,
        pending,
        pending_questions,
        pending_prompt,
        sid,
        req_id_seed,
        model,
    ));
    SessionState { id, handle, drain }
}

/// Map engine events to `session/update` notifications, turn permission asks
/// into `session/request_permission` requests, questions into
/// `session/request_question` requests, and answer the pending prompt on
/// TurnDone.
#[allow(clippy::too_many_arguments)]
async fn drain_events(
    mut events: mpsc::Receiver<EngineEvent>,
    writer: Writer,
    pending: Pending,
    pending_questions: PendingQuestions,
    pending_prompt: PendingPrompt,
    session_id: String,
    mut req_id: u64,
    model: String,
) {
    while let Some(event) = events.recv().await {
        match event {
            EngineEvent::Ask {
                summary,
                protected_why,
                reply,
            } => {
                req_id += 1;
                pending
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .insert(req_id, reply);
                // INVARIANT (unimplemented — see
                // specs/exec-plans/active/0020-remediation-surface.md RQ-2):
                // an `edit`/`write` ask also carries `"diff"`, the proposed
                // change, so the human approving a write can see it. The
                // generator (`crate::diffgen::for_tool`) and the client's
                // renderer are both built and tested; this call site cannot
                // use them because `EngineEvent::Ask` carries no `tool`/
                // `input`. Once it does, this becomes:
                //     let diff = crate::diffgen::for_tool(tool, input);
                // plus `"diff": diff.map(|d| d.iter().map(DiffLine::to_json)…)`
                // in the params below.
                send(&writer, &json!({
                    "jsonrpc": "2.0", "id": req_id, "method": "session/request_permission",
                    "params": {"sessionId": session_id, "summary": summary, "protectedWhy": protected_why},
                }))
                .await;
            }
            EngineEvent::Question {
                question, reply, ..
            } => {
                // NOT a permission gate: the reply is plain text the model
                // reads, never an authorization (SECURITY invariant).
                req_id += 1;
                pending_questions
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .insert(req_id, reply);
                send(
                    &writer,
                    &json!({
                        "jsonrpc": "2.0", "id": req_id, "method": "session/request_question",
                        "params": {
                            "sessionId": session_id,
                            "header": question.header,
                            "prompt": question.prompt,
                            "options": question.options,
                            "multi": question.multi,
                        },
                    }),
                )
                .await;
            }
            EngineEvent::TurnDone { outcome, usage } => {
                // A turn that ended without its asks/questions being answered
                // left dead reply channels behind — drop them so they can't
                // leak.
                pending
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .retain(|_, tx| !tx.is_closed());
                pending_questions
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .retain(|_, tx| !tx.is_closed());
                notify(
                    &writer,
                    &session_id,
                    json!({"type": "turn_done", "outcome": outcome_tag(&outcome)}),
                )
                .await;
                // Take the id and drop the guard *before* awaiting (a
                // std::sync guard held across .await would make this non-Send).
                let prompt_id = pending_prompt
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .pop_front();
                if let Some(id) = prompt_id {
                    reply_ok(
                        &writer,
                        id,
                        json!({
                            "schemaVersion": UPDATE_SCHEMA_VERSION,
                            "outcome": outcome_tag(&outcome),
                            "usage": crate::wire::usage_frame(&model, &usage),
                        }),
                    )
                    .await;
                }
            }
            other => {
                if let Some(update) = update_payload(&other) {
                    notify(&writer, &session_id, update).await;
                }
            }
        }
    }
}

/// The `session/update` payload for an event, or `None` when the event is not
/// a stream frame. Thin alias over [`crate::wire::update_frame`] — there used
/// to be three copies of this mapping and they had already drifted (§7); one
/// renderer means a new `EngineEvent` variant cannot reach one surface and
/// silently miss another.
pub(crate) fn update_payload(event: &EngineEvent) -> Option<Value> {
    crate::wire::update_frame(event)
}

pub(crate) fn outcome_tag(outcome: &Outcome) -> Value {
    crate::wire::outcome_frame(outcome)
}

async fn notify(writer: &Writer, session_id: &str, update: Value) {
    send(writer, &json!({
        "jsonrpc": "2.0", "method": "session/update",
        "params": {"schemaVersion": UPDATE_SCHEMA_VERSION, "sessionId": session_id, "update": update},
    }))
    .await;
}

async fn reply_ok(writer: &Writer, id: Value, result: Value) {
    send(
        writer,
        &json!({"jsonrpc": "2.0", "id": id, "result": result}),
    )
    .await;
}

async fn reply_err(writer: &Writer, id: Value, message: &str) {
    send(
        writer,
        &json!({"jsonrpc": "2.0", "id": id, "error": {"code": -32600, "message": message}}),
    )
    .await;
}

async fn send(writer: &Writer, msg: &Value) {
    let mut line = msg.to_string();
    line.push('\n');
    let mut w = writer.lock().await;
    let _ = w.write_all(line.as_bytes()).await;
    let _ = w.flush().await;
}