scsh 1.41.15

//! HTTP client for posting run events to the session browser daemon.

use std::io::{Read, Write};
use std::net::TcpStream;
use std::sync::mpsc::{self, RecvTimeoutError, Sender, TryRecvError};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

use super::model::{ProcKind, ProcStatus};
use super::paths::daemon_port;
use crate::json::quote;

/// Batch proc output before POSTing — ~2 flushes/sec, up to half a megabyte per payload.
const LINE_BATCH_INTERVAL: Duration = Duration::from_millis(500);
const LINE_BATCH_MAX_BYTES: usize = 512 * 1024;
/// Large parallel runs (e.g. code-review) can queue thousands of lines; wait long enough to drain.
const FLUSH_DRAIN_TIMEOUT: Duration = Duration::from_secs(120);
const POSTER_JOIN_TIMEOUT: Duration = Duration::from_secs(30);

enum PostJob {
  ProcLine { proc: usize, at: f64, line: String },
  Send { path: String, body: String },
  Flush { done: mpsc::Sender<()> },
}

struct ClientInner {
  port: u16,
  session_id: String,
  post_tx: Mutex<Option<Sender<PostJob>>>,
  poster: Mutex<Option<thread::JoinHandle<()>>>,
}

/// Posts events to the local daemon. Best-effort — failures are silent so runs never break.
pub struct Client {
  inner: Arc<ClientInner>,
}

impl Client {
  pub fn new(session_id: String) -> Client {
    let port = daemon_port();
    let (post_tx, post_rx) = mpsc::channel::<PostJob>();
    let session_for_poster = session_id.clone();
    let poster = thread::spawn(move || poster_loop(port, session_for_poster, post_rx));
    Client {
      inner: Arc::new(ClientInner {
        port,
        session_id,
        post_tx: Mutex::new(Some(post_tx)),
        poster: Mutex::new(Some(poster)),
      }),
    }
  }

  pub fn session_url(&self) -> String {
    super::paths::session_url(self.inner.port, &self.inner.session_id)
  }

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

  /// Keep a long quiet operation live in the daemon. Dropping the returned flag is not
  /// sufficient; set it to false before finishing the session.
  pub fn start_heartbeat(&self, every: std::time::Duration) -> Arc<std::sync::atomic::AtomicBool> {
    let active = Arc::new(std::sync::atomic::AtomicBool::new(true));
    let flag = Arc::clone(&active);
    let client = Client { inner: Arc::clone(&self.inner) };
    thread::spawn(move || {
      while flag.load(std::sync::atomic::Ordering::Relaxed) {
        client.ping();
        thread::sleep(every);
      }
    });
    active
  }

  pub fn register_session(
    &self, repo: &str, branch: &str, profile: Option<&str>, kind: &str, skills: &[(&str, &str)],
  ) -> bool {
    self.register_session_with_workflow(repo, branch, profile, kind, skills, None, None)
  }

  /// Like [`Self::register_session`], optionally attaching immutable workflow DAG metadata
  /// and/or a parent session id (e.g. annotate-cast catch-up under `(internal)`).
  #[allow(clippy::too_many_arguments)]
  pub fn register_session_with_workflow(
    &self, repo: &str, branch: &str, profile: Option<&str>, kind: &str, skills: &[(&str, &str)],
    workflow: Option<&super::workflow::WorkflowMeta>, parent_session: Option<&str>,
  ) -> bool {
    let skill_parts: Vec<String> = skills
      .iter()
      .map(|(name, harness)| format!("{{ \"name\": {}, \"harness\": {} }}", quote(name), quote(harness)))
      .collect();
    let profile_json = match profile {
      Some(p) => quote(p),
      None => "null".to_string(),
    };
    let workflow_part = match workflow {
      Some(w) => format!(", \"workflow\": {}", super::workflow::workflow_json(w)),
      None => String::new(),
    };
    let parent_part = match parent_session {
      Some(p) => format!(", \"parent_session\": {}", quote(p)),
      None => String::new(),
    };
    // An explicit `scsh run --retries N` (propagated as SCSH_RETRIES) declares the job's
    // restart budget at registration; omitted, the daemon applies its default — or keeps
    // the budget it already stamped on a session it created itself.
    let retries_part = std::env::var("SCSH_RETRIES")
      .ok()
      .and_then(|s| s.trim().parse::<u32>().ok())
      .map(|n| format!(", \"retries\": {n}"))
      .unwrap_or_default();
    let body = format!(
      "{{ \"session\": {}, \"repo\": {}, \"branch\": {}, \"profile\": {}, \"kind\": {}, \"skills\": [{}], \"run_pid\": {}{workflow_part}{parent_part}{retries_part} }}",
      quote(&self.inner.session_id),
      quote(repo),
      quote(branch),
      profile_json,
      quote(kind),
      skill_parts.join(", "),
      std::process::id(),
    );
    let start_ok = self.post_sync_after_flush("/api/v1/session/start", &body);
    let register_ok =
      self.post_sync_after_flush("/api/v1/register", &format!("{{ \"session\": {} }}", quote(&self.inner.session_id)));
    start_ok && register_ok
  }

  pub fn deregister(&self) {
    let body = format!("{{ \"session\": {} }}", quote(&self.inner.session_id));
    if !self.post_sync("/api/v1/deregister", &body) {
      log_post_failure("/api/v1/deregister", None);
    }
  }

  /// Drain the poster, deregister synchronously, then stop the poster thread.
  pub fn finish_session(&self) {
    let drained = self.flush_poster();
    self.deregister();
    if !drained {
      crate::failure::log_session_proc(
        &self.inner.session_id,
        crate::failure::reason::DAEMON_DRAIN_TIMEOUT,
        "(session)",
        "event poster did not drain before session end — some proc output or finish events may be missing from the browser UI",
      );
    }
    self.close_poster();
  }

  /// Close the async poster after queued work is drained (see `finish_session`).
  pub fn close_poster(&self) {
    let mut post_tx = self.inner.post_tx.lock().unwrap_or_else(|e| e.into_inner());
    *post_tx = None;
    let mut poster = self.inner.poster.lock().unwrap_or_else(|e| e.into_inner());
    if let Some(handle) = poster.take() {
      let (done_tx, done_rx) = mpsc::channel();
      thread::spawn(move || {
        let _ = handle.join();
        let _ = done_tx.send(());
      });
      let _ = done_rx.recv_timeout(POSTER_JOIN_TIMEOUT);
    }
  }

  /// Drain buffered proc lines and close the async poster.
  pub fn flush(&self) {
    self.flush_poster();
    self.close_poster();
  }

  #[allow(clippy::too_many_arguments)]
  pub fn proc_add(
    &self, proc_index: usize, label: &str, kind: ProcKind, skill_name: Option<&str>, harness: Option<&str>,
    model: Option<&str>, skill_source: Option<&str>, route: Option<&str>, annotate_target: Option<&str>,
    previous_attempt: Option<usize>,
  ) {
    let mut extras = Vec::new();
    if let Some(s) = skill_name {
      extras.push(format!("\"skill_name\": {}", quote(s)));
    }
    if let Some(h) = harness {
      extras.push(format!("\"harness\": {}", quote(h)));
    }
    if let Some(m) = model {
      extras.push(format!("\"model\": {}", quote(m)));
    }
    if let Some(s) = skill_source {
      extras.push(format!("\"skill_source\": {}", quote(s)));
    }
    if let Some(r) = route {
      extras.push(format!("\"route\": {}", quote(r)));
    }
    // The cast an Annotate proc summarizes — how the daemon later answers "which job is
    // summarizing this recording?" for the chapters-pending link on the job page.
    if let Some(t) = annotate_target {
      extras.push(format!("\"annotate_target\": {}", quote(t)));
    }
    if let Some(index) = previous_attempt {
      extras.push(format!("\"previous_attempt\": {index}"));
    }
    let tail = if extras.is_empty() { String::new() } else { format!(", {}", extras.join(", ")) };
    let body = format!(
      "{{ \"session\": {}, \"proc\": {}, \"label\": {}, \"kind\": {}{tail} }}",
      quote(&self.inner.session_id),
      proc_index,
      quote(label),
      quote(kind.as_str()),
    );
    self.post("/api/v1/proc/add", &body);
  }

  pub fn proc_start(&self, proc_index: usize) {
    let body = format!("{{ \"session\": {}, \"proc\": {} }}", quote(&self.inner.session_id), proc_index);
    self.post("/api/v1/proc/start", &body);
  }

  pub fn proc_note(&self, proc_index: usize, note: &str) {
    let body = format!(
      "{{ \"session\": {}, \"proc\": {}, \"note\": {} }}",
      quote(&self.inner.session_id),
      proc_index,
      quote(note)
    );
    self.post("/api/v1/proc/note", &body);
  }

  /// Tell the daemon where this proc's asciinema `.cast` lives on the host: the live
  /// run-dir file while the container runs, then the durable copy after the skill ends.
  pub fn proc_cast(&self, proc_index: usize, path: &str) {
    let body = format!(
      "{{ \"session\": {}, \"proc\": {}, \"path\": {} }}",
      quote(&self.inner.session_id),
      proc_index,
      quote(path)
    );
    // Synchronous so a browser force-stop can suppress annotation on the durable cast
    // before the run reaches its end-of-job annotation sweep.
    if !self.post_sync_after_flush("/api/v1/proc/cast", &body) {
      log_post_failure("/api/v1/proc/cast", Some(proc_index));
    }
  }

  /// Tell the daemon where the packed commits-diff page for this step lives on the host
  /// (the packdiff HTML under `$SCSH_HOME/sessions/<session>/diffs/`). Posted after the
  /// run integrates the step's commits — i.e. after proc/finish — so it must be
  /// synchronous: the run is about to deregister and close the poster.
  pub fn proc_diff(&self, proc_index: usize, path: &str) {
    let body = format!(
      "{{ \"session\": {}, \"proc\": {}, \"path\": {} }}",
      quote(&self.inner.session_id),
      proc_index,
      quote(path)
    );
    if !self.post_sync_after_flush("/api/v1/proc/diff", &body) {
      log_post_failure("/api/v1/proc/diff", Some(proc_index));
    }
  }

  /// Tell the daemon where the durable skill-result JSON for this proc lives
  /// (`$SCSH_HOME/sessions/<session>/results/…`). Posted after collect, like `proc_diff`.
  pub fn proc_result(&self, proc_index: usize, path: &str) {
    let body = format!(
      "{{ \"session\": {}, \"proc\": {}, \"path\": {} }}",
      quote(&self.inner.session_id),
      proc_index,
      quote(path)
    );
    if !self.post_sync_after_flush("/api/v1/proc/result", &body) {
      log_post_failure("/api/v1/proc/result", Some(proc_index));
    }
  }

  pub fn proc_line(&self, proc_index: usize, at: f64, line: &str) {
    if let Some(tx) = lock_post_tx(&self.inner) {
      let _ = tx.send(PostJob::ProcLine { proc: proc_index, at, line: line.to_string() });
    }
  }

  pub fn proc_finish(
    &self, proc_index: usize, status: ProcStatus, fail_reason: Option<&str>, detail: Option<&str>, elapsed: f64,
  ) {
    let detail_json = match detail {
      Some(d) => quote(d),
      None => "null".to_string(),
    };
    let fail_reason_json = match fail_reason {
      Some(r) => quote(r),
      None => "null".to_string(),
    };
    let body = format!(
      "{{ \"session\": {}, \"proc\": {}, \"status\": {}, \"fail_reason\": {}, \"detail\": {}, \"elapsed\": {} }}",
      quote(&self.inner.session_id),
      proc_index,
      quote(status.as_str()),
      fail_reason_json,
      detail_json,
      elapsed
    );
    if !self.post_sync_after_flush("/api/v1/proc/finish", &body) {
      log_post_failure("/api/v1/proc/finish", Some(proc_index));
      if status == ProcStatus::Fail {
        crate::failure::log_session_proc(
          &self.inner.session_id,
          crate::failure::reason::DAEMON_POST_FAILED,
          &format!("proc {proc_index}"),
          detail.unwrap_or("(proc/finish not accepted by daemon)"),
        );
      }
    }
  }

  pub fn container_event(&self, proc_index: usize, action: &str, name: &str, runtime: &str) {
    let body = format!(
      "{{ \"session\": {}, \"proc\": {}, \"action\": {}, \"name\": {}, \"runtime\": {} }}",
      quote(&self.inner.session_id),
      proc_index,
      quote(action),
      quote(name),
      quote(runtime),
    );
    if action == "stop" {
      if !self.post_sync_after_flush("/api/v1/container", &body) {
        log_post_failure("/api/v1/container", Some(proc_index));
      }
    } else {
      self.post("/api/v1/container", &body);
    }
  }

  /// Backup schedule for a run dir — the client deletes first; the daemon retries if it still exists.
  pub fn schedule_run_dir_prune(&self, run_dir: &str, container_name: &str, runtime: &str, outcome_ok: bool) {
    let outcome = if outcome_ok { "ok" } else { "fail" };
    let body = format!(
      "{{ \"run_dir\": {}, \"container_name\": {}, \"runtime\": {}, \"outcome\": {} }}",
      quote(run_dir),
      quote(container_name),
      quote(runtime),
      quote(outcome),
    );
    self.post("/api/v1/prune/schedule", &body);
  }

  pub fn ping(&self) {
    self.post("/api/v1/ping", &format!("{{ \"session\": {} }}", quote(&self.inner.session_id)));
  }

  /// True when the session browser daemon serves its HTTP API on localhost.
  pub fn daemon_alive() -> bool {
    super::paths::daemon_api_responds(daemon_port())
  }

  fn post(&self, path: &str, body: &str) {
    if let Some(tx) = lock_post_tx(&self.inner) {
      let _ = tx.send(PostJob::Send { path: path.to_string(), body: body.to_string() });
    }
  }

  fn post_sync(&self, path: &str, body: &str) -> bool {
    send_post(self.inner.port, path, body)
  }

  /// Drain buffered proc lines in the poster before a synchronous lifecycle POST.
  fn post_sync_after_flush(&self, path: &str, body: &str) -> bool {
    self.flush_poster();
    self.post_sync(path, body)
  }

  fn flush_poster(&self) -> bool {
    if let Some(tx) = lock_post_tx(&self.inner) {
      let (done_tx, done_rx) = mpsc::channel();
      if tx.send(PostJob::Flush { done: done_tx }).is_ok() {
        return done_rx.recv_timeout(FLUSH_DRAIN_TIMEOUT).is_ok();
      }
    }
    true
  }
}

/// One synchronous POST to a daemon on `port`, outside any session (e.g. `scsh prune --now`).
pub fn post_once(port: u16, path: &str, body: &str) -> bool {
  send_post(port, path, body)
}

fn log_daemon_warn(msg: &str) {
  eprintln!("session browser: {msg}");
}

fn log_post_failure(path: &str, proc: Option<usize>) {
  match proc {
    Some(i) => log_daemon_warn(&format!("POST {path} for proc {i} failed (daemon unreachable or rejected request)")),
    None => log_daemon_warn(&format!("POST {path} failed (daemon unreachable or rejected request)")),
  }
}

fn lock_post_tx(inner: &ClientInner) -> Option<Sender<PostJob>> {
  inner.post_tx.lock().unwrap_or_else(|e| e.into_inner()).clone()
}

fn poster_loop(port: u16, session_id: String, rx: mpsc::Receiver<PostJob>) {
  let mut line_buf: Vec<(usize, f64, String)> = Vec::new();
  let mut line_bytes = 0usize;
  let mut conn = DaemonConn::default();
  loop {
    match rx.recv_timeout(LINE_BATCH_INTERVAL) {
      Ok(PostJob::ProcLine { proc, at, line }) => {
        line_bytes += line.len();
        line_buf.push((proc, at, line));
        if line_bytes >= LINE_BATCH_MAX_BYTES {
          flush_lines(&mut conn, port, &session_id, &mut line_buf, &mut line_bytes);
        }
      }
      Ok(PostJob::Send { path, body }) => {
        flush_lines(&mut conn, port, &session_id, &mut line_buf, &mut line_bytes);
        if !conn.post(port, &path, &body) {
          log_post_failure(&path, None);
        }
      }
      Ok(PostJob::Flush { done }) => {
        drain_poster_burst(&rx, &mut conn, port, &session_id, &mut line_buf, &mut line_bytes);
        let _ = done.send(());
      }
      Err(RecvTimeoutError::Timeout) => {
        flush_lines(&mut conn, port, &session_id, &mut line_buf, &mut line_bytes);
      }
      Err(RecvTimeoutError::Disconnected) => {
        flush_lines(&mut conn, port, &session_id, &mut line_buf, &mut line_bytes);
        break;
      }
    }
  }
}

/// After the explicit flush marker, drain any jobs already queued behind it before acking.
fn drain_poster_burst(
  rx: &mpsc::Receiver<PostJob>, conn: &mut DaemonConn, port: u16, session_id: &str,
  line_buf: &mut Vec<(usize, f64, String)>, line_bytes: &mut usize,
) {
  flush_lines(conn, port, session_id, line_buf, line_bytes);
  loop {
    match rx.try_recv() {
      Ok(PostJob::ProcLine { proc, at, line }) => {
        *line_bytes += line.len();
        line_buf.push((proc, at, line));
        if *line_bytes >= LINE_BATCH_MAX_BYTES {
          flush_lines(conn, port, session_id, line_buf, line_bytes);
        }
      }
      Ok(PostJob::Send { path, body }) => {
        flush_lines(conn, port, session_id, line_buf, line_bytes);
        if !conn.post(port, &path, &body) {
          log_post_failure(&path, None);
        }
      }
      Ok(PostJob::Flush { done }) => {
        flush_lines(conn, port, session_id, line_buf, line_bytes);
        let _ = done.send(());
      }
      Err(TryRecvError::Empty) => break,
      Err(TryRecvError::Disconnected) => break,
    }
  }
  flush_lines(conn, port, session_id, line_buf, line_bytes);
}

fn group_line_buf(buf: &[(usize, f64, String)]) -> Vec<(usize, Vec<(f64, String)>)> {
  let mut groups = Vec::new();
  let mut i = 0;
  while i < buf.len() {
    let proc = buf[i].0;
    let mut j = i + 1;
    while j < buf.len() && buf[j].0 == proc {
      j += 1;
    }
    let chunk: Vec<(f64, String)> = buf[i..j].iter().map(|(_, at, line)| (*at, line.clone())).collect();
    groups.push((proc, chunk));
    i = j;
  }
  groups
}

fn flush_lines(
  conn: &mut DaemonConn, port: u16, session_id: &str, buf: &mut Vec<(usize, f64, String)>, bytes: &mut usize,
) {
  if buf.is_empty() {
    return;
  }
  for (proc, chunk) in group_line_buf(buf) {
    if !send_lines_bulk(conn, port, session_id, proc, &chunk) {
      log_post_failure("/api/v1/proc/lines", Some(proc));
    }
  }
  buf.clear();
  *bytes = 0;
}

fn send_lines_bulk(
  conn: &mut DaemonConn, port: u16, session_id: &str, proc_index: usize, lines: &[(f64, String)],
) -> bool {
  if lines.is_empty() {
    return true;
  }
  let entries: Vec<String> =
    lines.iter().map(|(at, text)| format!("{{ \"at\": {at}, \"line\": {} }}", quote(text))).collect();
  let body =
    format!("{{ \"session\": {}, \"proc\": {}, \"lines\": [{}] }}", quote(session_id), proc_index, entries.join(", "));
  conn.post(port, "/api/v1/proc/lines", &body)
}

fn send_post(port: u16, path: &str, body: &str) -> bool {
  let Ok(mut stream) =
    TcpStream::connect_timeout(&format!("127.0.0.1:{port}").parse().unwrap(), Duration::from_millis(500))
  else {
    return false;
  };
  stream.set_read_timeout(Some(Duration::from_secs(2))).ok();
  stream.set_write_timeout(Some(Duration::from_secs(2))).ok();
  let req = format!(
    "POST {path} HTTP/1.1\r\n\
Host: 127.0.0.1\r\n\
Content-Type: application/json\r\n\
Content-Length: {len}\r\n\
Connection: close\r\n\r\n\
{body}",
    len = body.len()
  );
  if stream.write_all(req.as_bytes()).is_err() {
    return false;
  }
  let mut resp = String::new();
  if stream.read_to_string(&mut resp).is_err() {
    return false;
  }
  resp.starts_with("HTTP/1.1 200") || resp.starts_with("HTTP/1.0 200")
}

/// The poster thread's reusable keep-alive connection to the daemon. A running job posts
/// events and line batches constantly; a connection per POST parked thousands of loopback
/// sockets in TIME_WAIT and could exhaust the machine's ephemeral ports. One connection
/// carries every post; the daemon idle-closes it after a few quiet seconds and the next
/// post reconnects.
#[derive(Default)]
struct DaemonConn {
  stream: Option<TcpStream>,
}

impl DaemonConn {
  fn post(&mut self, port: u16, path: &str, body: &str) -> bool {
    let req = format!(
      "POST {path} HTTP/1.1\r\n\
Host: 127.0.0.1\r\n\
Content-Type: application/json\r\n\
Content-Length: {len}\r\n\r\n\
{body}",
      len = body.len()
    );
    // A post on a reused connection can race the daemon's idle close, and only a race the
    // server provably never processed may retry on a fresh connection:
    //  - the WRITE failed — the request is incomplete on the wire, an incomplete request
    //    cannot be routed, and dropping our socket guarantees it never completes;
    //  - the READ died before ANY response byte with a disconnect error (see
    //    `unprocessed_disconnect` for the per-OS shapes this takes).
    // A fresh-connection failure, a mid-response error, or a timeout is never retried — a
    // slow daemon may still process the request, and a duplicate would corrupt the store.
    // Known residual window, accepted deliberately: a daemon that DIES between reading the
    // full request and writing the first response byte is indistinguishable from the idle
    // close, so that one post could be applied twice across the daemon generations. The
    // idle-close race happens constantly (every >5s gap between posts); a daemon death
    // inside a sub-millisecond handler is vanishingly rare, and without the retry every
    // idle-close race would LOSE a post instead. See DAEMON-KEEPALIVE.md.
    for _ in 0..2 {
      let reused = self.stream.is_some();
      let Some(stream) = self.connect(port) else {
        return false;
      };
      if stream.write_all(req.as_bytes()).is_err() {
        self.stream = None;
        if reused {
          continue;
        }
        return false;
      }
      match read_keep_alive_response(stream) {
        Ok((ok, close)) => {
          if close {
            self.stream = None;
          }
          return ok;
        }
        Err(e) => {
          self.stream = None;
          if !(reused && unprocessed_disconnect(&e)) {
            return false;
          }
        }
      }
    }
    false
  }

  fn connect(&mut self, port: u16) -> Option<&mut TcpStream> {
    if self.stream.is_none() {
      let stream =
        TcpStream::connect_timeout(&format!("127.0.0.1:{port}").parse().unwrap(), Duration::from_millis(500)).ok()?;
      stream.set_read_timeout(Some(Duration::from_secs(2))).ok();
      stream.set_write_timeout(Some(Duration::from_secs(2))).ok();
      self.stream = Some(stream);
    }
    self.stream.as_mut()
  }
}

/// True for read errors proving the server closed the connection before sending any
/// response byte. The same idle-close race surfaces differently per OS: a clean FIN reads
/// as zero bytes (mapped to `UnexpectedEof`), but Linux and Windows commonly answer a write
/// on a server-closed socket with an RST, so the read fails `ConnectionReset` (or
/// `ConnectionAborted`/`BrokenPipe`) instead. Timeouts are deliberately absent — a slow
/// daemon may still process the request. Callers only see these kinds when zero response
/// bytes arrived (`read_keep_alive_response` demotes later failures to `InvalidData`).
fn unprocessed_disconnect(e: &std::io::Error) -> bool {
  matches!(
    e.kind(),
    std::io::ErrorKind::UnexpectedEof
      | std::io::ErrorKind::ConnectionReset
      | std::io::ErrorKind::ConnectionAborted
      | std::io::ErrorKind::BrokenPipe
  )
}

/// Read exactly one HTTP response off a keep-alive connection: headers, then the body the
/// Content-Length header promises — never to EOF, which would block forever on a live
/// connection. Returns (status-is-200, server-asked-close). Raw error kinds pass through
/// only while ZERO response bytes have arrived (the retryable stale-connection shapes);
/// once any byte arrived the server may have processed the request, so every later failure
/// is demoted to `InvalidData`, which is never retried.
fn read_keep_alive_response(stream: &mut TcpStream) -> std::io::Result<(bool, bool)> {
  use std::io::Read;
  let mut buf = Vec::new();
  let mut chunk = [0u8; 4096];
  let header_end = loop {
    if let Some(end) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
      break end;
    }
    if buf.len() > 64 * 1024 {
      return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "response headers too large"));
    }
    let n = match stream.read(&mut chunk) {
      Ok(n) => n,
      Err(e) if buf.is_empty() => return Err(e),
      Err(_) => return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "connection failed mid-response")),
    };
    if n == 0 {
      let kind = if buf.is_empty() { std::io::ErrorKind::UnexpectedEof } else { std::io::ErrorKind::InvalidData };
      return Err(std::io::Error::new(kind, "connection closed mid-response"));
    }
    buf.extend_from_slice(&chunk[..n]);
  };
  let head = String::from_utf8_lossy(&buf[..header_end]).to_ascii_lowercase();
  let content_length = head
    .lines()
    .find_map(|line| line.strip_prefix("content-length:"))
    .and_then(|v| v.trim().parse::<usize>().ok())
    .unwrap_or(0);
  let close = head.lines().any(|line| line.trim() == "connection: close");
  let mut have = buf.len() - (header_end + 4);
  while have < content_length {
    let n = stream
      .read(&mut chunk)
      .map_err(|_| std::io::Error::new(std::io::ErrorKind::InvalidData, "connection failed mid-body"))?;
    if n == 0 {
      return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "connection closed mid-body"));
    }
    have += n;
  }
  let ok = head.starts_with("http/1.1 200") || head.starts_with("http/1.0 200");
  Ok((ok, close))
}

/// The real scsh binary to re-exec (`__daemon-serve`, and the daemon's `build-images` spawn).
pub(crate) fn scsh_executable() -> std::io::Result<std::path::PathBuf> {
  if let Ok(path) = std::env::var("SCSH_BIN") {
    return Ok(std::path::PathBuf::from(path));
  }
  let exe = std::env::current_exe()?;
  let lossy = exe.to_string_lossy();
  // `cargo test` runs a harness at `target/{debug,release}/deps/scsh-<hash>`; spawn the real binary.
  if lossy.contains("/deps/scsh-") {
    let mut candidate = exe.clone();
    if candidate.pop() && candidate.pop() {
      candidate.push("scsh");
      if candidate.is_file() {
        return Ok(candidate);
      }
    }
  }
  Ok(exe)
}

/// Spawn the daemon child process (re-exec `scsh __daemon-serve`).
#[cfg(not(unix))]
pub fn spawn_daemon(mode: super::model::DaemonMode) -> std::io::Result<()> {
  let exe = scsh_executable()?;
  let port = daemon_port();
  let mut cmd = std::process::Command::new(exe);
  cmd.args(["__daemon-serve", "--mode", mode.as_str(), "--port", &port.to_string()]);
  cmd.stdin(std::process::Stdio::null());
  cmd.stdout(std::process::Stdio::null());
  cmd.stderr(std::process::Stdio::null());
  cmd.spawn()?;
  wait_for_daemon(Duration::from_secs(5))
}

#[cfg(unix)]
pub fn spawn_daemon(mode: super::model::DaemonMode) -> std::io::Result<()> {
  let exe = scsh_executable()?;
  let port = daemon_port();
  let mut cmd = std::process::Command::new(exe);
  cmd.args(["__daemon-serve", "--mode", mode.as_str(), "--port", &port.to_string()]);
  cmd.stdin(std::process::Stdio::null());
  cmd.stdout(std::process::Stdio::null());
  cmd.stderr(std::process::Stdio::null());
  {
    use std::os::unix::process::CommandExt;
    // Double-fork so the daemon survives the parent `scsh` process exiting; `setsid` detaches
    // from the terminal (async-signal-safe syscalls only in pre_exec).
    unsafe {
      cmd.pre_exec(super::paths::daemon_detach_child);
    }
  }
  cmd.spawn()?;
  std::thread::sleep(Duration::from_millis(50));
  wait_for_daemon(Duration::from_secs(5))
}

fn wait_for_daemon(timeout: Duration) -> std::io::Result<()> {
  let deadline = std::time::Instant::now() + timeout;
  while std::time::Instant::now() < deadline {
    if Client::daemon_alive() {
      return Ok(());
    }
    std::thread::sleep(Duration::from_millis(50));
  }
  Err(std::io::Error::new(std::io::ErrorKind::TimedOut, "daemon did not start"))
}

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

  #[test]
  fn client_session_url_uses_port() {
    struct PinDaemonPort {
      previous: Option<String>,
    }
    impl PinDaemonPort {
      fn set() -> (u16, Self) {
        let port = std::net::TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
        let previous = std::env::var("SCSH_DAEMON_PORT").ok();
        std::env::set_var("SCSH_DAEMON_PORT", port.to_string());
        (port, Self { previous })
      }
    }
    impl Drop for PinDaemonPort {
      fn drop(&mut self) {
        match &self.previous {
          Some(v) => std::env::set_var("SCSH_DAEMON_PORT", v),
          None => std::env::remove_var("SCSH_DAEMON_PORT"),
        }
      }
    }
    let (port, _pin) = PinDaemonPort::set();
    let c = Client::new("abcdef".into());
    assert!(c.session_url().contains("abcdef"));
    assert!(c.session_url().contains(&port.to_string()));
  }

  #[test]
  fn poster_batches_lines_by_proc() {
    let (tx, rx) = mpsc::channel();
    let handle = thread::spawn(move || poster_loop(1, "sess01".into(), rx));
    tx.send(PostJob::ProcLine { proc: 1, at: 1.0, line: "b".into() }).unwrap();
    tx.send(PostJob::ProcLine { proc: 0, at: 0.5, line: "a".into() }).unwrap();
    drop(tx);
    let _ = handle.join();
    // No daemon on port 1 — exercise is flush/group without panic; send_post fails silently.
  }

  #[test]
  fn group_line_buf_preserves_insertion_order() {
    let buf =
      vec![(1, 1.0, "first".into()), (2, 2.0, "other".into()), (1, 3.0, "second".into()), (1, 4.0, "third".into())];
    let groups = group_line_buf(&buf);
    assert_eq!(groups.len(), 3);
    assert_eq!(groups[0].0, 1);
    assert_eq!(groups[0].1.len(), 1);
    assert_eq!(groups[0].1[0].1, "first");
    assert_eq!(groups[1].0, 2);
    assert_eq!(groups[2].0, 1);
    assert_eq!(groups[2].1.len(), 2);
    assert_eq!(groups[2].1[0].1, "second");
    assert_eq!(groups[2].1[1].1, "third");
  }

  /// Read one HTTP request off a test-server socket (headers, then the Content-Length body).
  fn read_test_request(s: &mut TcpStream) -> Vec<u8> {
    use std::io::Read;
    let mut buf = Vec::new();
    let mut chunk = [0u8; 4096];
    loop {
      if let Some(end) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
        let head = String::from_utf8_lossy(&buf[..end]).to_ascii_lowercase();
        let need = head
          .lines()
          .find_map(|l| l.strip_prefix("content-length:"))
          .and_then(|v| v.trim().parse::<usize>().ok())
          .unwrap_or(0);
        if buf.len() >= end + 4 + need {
          return buf;
        }
      }
      let n = s.read(&mut chunk).expect("read request");
      assert!(n > 0, "client closed mid-request");
      buf.extend_from_slice(&chunk[..n]);
    }
  }

  const KEEP_ALIVE_200: &[u8] =
    b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 2\r\nConnection: keep-alive\r\n\r\nok";

  #[test]
  fn keep_alive_responses_are_framed_and_the_connection_is_reused() {
    use std::io::Write;
    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let port = listener.local_addr().unwrap().port();
    let server = std::thread::spawn(move || {
      // ONE connection serves two posts back to back; a client reading to EOF would hang.
      let (mut s, _) = listener.accept().unwrap();
      for _ in 0..2 {
        read_test_request(&mut s);
        s.write_all(KEEP_ALIVE_200).unwrap();
      }
    });
    let mut conn = DaemonConn::default();
    assert!(conn.post(port, "/api/v1/ping", "{}"), "first post served");
    assert!(conn.stream.is_some(), "keep-alive response leaves the connection cached");
    assert!(conn.post(port, "/api/v1/ping", "{}"), "second post rides the same connection");
    server.join().unwrap();
  }

  #[test]
  fn stale_idle_closed_connection_is_replaced_with_one_safe_retry() {
    use std::io::Write;
    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let port = listener.local_addr().unwrap().port();
    let server = std::thread::spawn(move || {
      // First connection: serve one post, then idle-close (the daemon's 5s timeout).
      let (mut s, _) = listener.accept().unwrap();
      read_test_request(&mut s);
      s.write_all(KEEP_ALIVE_200).unwrap();
      drop(s);
      // The retry arrives on a fresh connection and is served.
      let (mut s, _) = listener.accept().unwrap();
      read_test_request(&mut s);
      s.write_all(KEEP_ALIVE_200).unwrap();
    });
    let mut conn = DaemonConn::default();
    assert!(conn.post(port, "/api/v1/ping", "{}"), "first post served");
    // The cached connection is now dead server-side; the next post must detect the EOF
    // (nothing was processed) and succeed via exactly one fresh-connection retry.
    assert!(conn.post(port, "/api/v1/ping", "{}"), "stale connection replaced transparently");
    server.join().unwrap();
  }

  #[test]
  fn unprocessed_disconnect_covers_each_os_reset_shape_but_never_timeouts() {
    use std::io::{Error, ErrorKind};
    // A clean FIN reads as zero bytes (macOS commonly); Linux and Windows answer a write
    // on a server-closed socket with an RST, surfacing as a reset instead.
    for kind in
      [ErrorKind::UnexpectedEof, ErrorKind::ConnectionReset, ErrorKind::ConnectionAborted, ErrorKind::BrokenPipe]
    {
      assert!(unprocessed_disconnect(&Error::new(kind, "x")), "{kind:?} proves nothing was processed");
    }
    // A slow daemon may still process the request — retrying a timeout could double-post.
    for kind in [ErrorKind::WouldBlock, ErrorKind::TimedOut, ErrorKind::InvalidData] {
      assert!(!unprocessed_disconnect(&Error::new(kind, "x")), "{kind:?} must never retry");
    }
  }

  #[test]
  fn a_connection_lost_mid_response_is_not_retried() {
    use std::io::Write;
    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let port = listener.local_addr().unwrap().port();
    let server = std::thread::spawn(move || {
      // Serve one full response (primes the cached connection), then die mid-headers on
      // the second post: the server may already have processed it, so no retry.
      let (mut s, _) = listener.accept().unwrap();
      read_test_request(&mut s);
      s.write_all(KEEP_ALIVE_200).unwrap();
      read_test_request(&mut s);
      s.write_all(b"HTTP/1.1 2").unwrap();
    });
    let mut conn = DaemonConn::default();
    assert!(conn.post(port, "/api/v1/ping", "{}"));
    assert!(!conn.post(port, "/api/v1/ping", "{}"), "a partial response fails without a retry");
    assert!(conn.stream.is_none(), "the broken connection is dropped");
    server.join().unwrap();
  }

  #[test]
  fn a_close_labeled_response_drops_the_cached_connection() {
    use std::io::Write;
    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let port = listener.local_addr().unwrap().port();
    let server = std::thread::spawn(move || {
      let (mut s, _) = listener.accept().unwrap();
      read_test_request(&mut s);
      s.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok").unwrap();
    });
    let mut conn = DaemonConn::default();
    assert!(conn.post(port, "/api/v1/ping", "{}"));
    assert!(conn.stream.is_none(), "a Connection: close response is not reused");
    server.join().unwrap();
  }
}