use std::io;
use std::net::TcpStream;
use std::path::PathBuf;
use std::sync::mpsc;
use std::sync::Arc;
use std::thread;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use running_process::pty::NativePtyProcess;
use running_process::{NativeProcess, TerminalCapabilities};
use serde::{Deserialize, Serialize};
use sysinfo::Signal;

use crate::command::LaunchPlan;
pub use crate::gc::RepoVisit;
use crate::process_identity::ProcessIdentity;

use super::process_utils::signal_process_tree;

pub(super) const ENV_FEATURE_FLAG: &str = "CLUD_EXPERIMENTAL_DAEMON";
pub(super) const ENV_STATE_DIR: &str = "CLUD_DAEMON_STATE_DIR";
pub(super) const ENV_BACKLOG_BYTES: &str = "CLUD_BACKLOG_BYTES";
/// Issue #135: opt out of the always-on daemon auto-spawn. Used by both
/// the CLI flag `--no-daemon` and the `clud gc *` precondition check.
pub const ENV_NO_DAEMON: &str = "CLUD_NO_DAEMON";
pub(super) const DEFAULT_BACKLOG_LIMIT_BYTES: usize = 256 * 1024;
pub(super) const BACKGROUND_PROMPT_TIMEOUT: Duration = Duration::from_secs(5);

/// pm2-style per-session log file. Soft cap at 10 MiB; exceeding rolls the
/// current file to `<id>.log.1` (overwriting any prior backup). Keeping only
/// one backup is deliberate — clud sessions are ephemeral and the on-disk
/// footprint shouldn't grow unboundedly for a stale session nobody
/// reattaches to.
pub(super) const LOG_ROTATE_BYTES: u64 = 10 * 1024 * 1024;

pub(super) fn default_attachable() -> bool {
    true
}

pub(super) fn unix_millis_now() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as u64
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub(super) enum SessionKind {
    Subprocess,
    Pty,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub(super) struct DaemonInfo {
    pub(super) pid: u32,
    /// Issue #558: start time of `pid`, so a client that later signals the
    /// recorded daemon cannot hit an unrelated process that inherited the
    /// PID after the daemon died. `0` in files written by an older clud;
    /// [`ProcessIdentity`] falls back to a PID-only comparison there.
    #[serde(default)]
    pub(super) pid_start: u64,
    pub(super) port: u16,
    /// Issue #183: loopback port for the in-process HTTP dashboard. `None`
    /// when the dashboard listener failed to bind (logged once at daemon
    /// start; IPC keeps working). Older clud versions wrote this file
    /// without the field, so reads tolerate its absence.
    #[serde(default)]
    pub(super) dashboard_port: Option<u16>,
    /// Per-daemon capability needed to access the loopback dashboard. This is
    /// deliberately absent from dashboard responses; it is only persisted in
    /// the user-private daemon-info file so `clud ui` can bootstrap a browser.
    #[serde(default)]
    pub(super) dashboard_token: Option<String>,
    /// Issue #192: the `CARGO_PKG_VERSION` of the binary that launched
    /// this daemon. `ensure_daemon` uses this to detect a stale daemon
    /// after an in-place upgrade and restart it so bug-fix releases (e.g.
    /// the #190 registry merge) take effect on the next `clud` invocation.
    /// `None` for daemon.json files written by clud <= 2.0.14.
    #[serde(default)]
    pub(super) version: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub(super) struct SessionSnapshot {
    pub(super) id: String,
    pub(super) kind: SessionKind,
    #[serde(default)]
    pub(super) backend: Option<String>,
    #[serde(default)]
    pub(super) launch_mode: Option<String>,
    #[serde(default)]
    pub(super) repo_root: Option<String>,
    #[serde(default)]
    pub(super) command: Vec<String>,
    #[serde(default)]
    pub(super) cwd: Option<String>,
    #[serde(default)]
    pub(super) name: Option<String>,
    #[serde(default)]
    pub(super) created_at: Option<u64>,
    #[serde(default)]
    pub(super) detachable: bool,
    #[serde(default)]
    pub(super) background: bool,
    #[serde(default = "default_attachable")]
    pub(super) attachable: bool,
    #[serde(default)]
    pub(super) repeat_interval_secs: Option<u64>,
    #[serde(default)]
    pub(super) repeat_next_run_at: Option<u64>,
    #[serde(default)]
    pub(super) repeat_running: bool,
    pub(super) daemon_pid: u32,
    pub(super) worker_pid: u32,
    pub(super) worker_port: u16,
    pub(super) root_pid: Option<u32>,
    /// Issue #558: OS start times pinned to the three PIDs above, so a
    /// consumer of this snapshot can tell "that process is still running"
    /// from "something else now holds that number". `0` (i.e.
    /// [`UNKNOWN_START_TIME`]) in snapshots written by an older clud.
    #[serde(default)]
    pub(super) daemon_pid_start: u64,
    #[serde(default)]
    pub(super) worker_pid_start: u64,
    #[serde(default)]
    pub(super) root_pid_start: u64,
    pub(super) exit_code: Option<i32>,
    #[serde(default)]
    pub(super) exited_at: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(super) ctrl_c: Option<CtrlCProfile>,
}

impl DaemonInfo {
    /// PID plus start time, for callers that will later signal this daemon.
    pub(super) fn identity(&self) -> ProcessIdentity {
        ProcessIdentity::new(self.pid, self.pid_start)
    }
}

impl SessionSnapshot {
    /// Identity of the worker that owns this session.
    pub(super) fn worker_identity(&self) -> ProcessIdentity {
        ProcessIdentity::new(self.worker_pid, self.worker_pid_start)
    }

    /// Identity of the agent process this session launched, if one is
    /// currently recorded.
    pub(super) fn root_identity(&self) -> Option<ProcessIdentity> {
        self.root_pid
            .map(|pid| ProcessIdentity::new(pid, self.root_pid_start))
    }

    /// Identity of the daemon that supervises this session.
    pub(super) fn daemon_identity(&self) -> ProcessIdentity {
        ProcessIdentity::new(self.daemon_pid, self.daemon_pid_start)
    }
}

#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub(super) struct CtrlCProfile {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(super) cli_pid: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(super) cli_observed_at_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(super) cli_handoff_at_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(super) cli_return_ready_at_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(super) cli_handoff_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(super) daemon_received_at_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(super) daemon_kill_started_at_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(super) daemon_kill_finished_at_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(super) daemon_kill_ms: Option<u64>,
    #[serde(default, skip_serializing_if = "is_false")]
    pub(super) fast_path: bool,
}

fn is_false(value: &bool) -> bool {
    !*value
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub(super) struct WorkerLaunchSpec {
    pub(super) plan: LaunchPlan,
    pub(super) kind: SessionKind,
    #[serde(default)]
    pub(super) name: Option<String>,
    #[serde(default)]
    pub(super) detachable: bool,
    #[serde(default)]
    pub(super) background_on_launch: bool,
    #[serde(default = "default_attachable")]
    pub(super) attachable: bool,
    pub(super) rows: u16,
    pub(super) cols: u16,
    #[serde(default)]
    pub(super) repeat_interval_secs: Option<u64>,
    #[serde(default)]
    pub(super) repeat_run_command: Option<Vec<String>>,
    /// In-memory attach-replay backlog cap. `None` uses the compiled default
    /// (`DEFAULT_BACKLOG_LIMIT_BYTES`). Optional for wire compatibility with
    /// spec files written by older clud versions.
    #[serde(default)]
    pub(super) backlog_bytes: Option<usize>,
    /// Optional daemon-side transcript file. The worker tees every output
    /// chunk through running-process telemetry and drains the file sink
    /// before exiting.
    #[serde(default)]
    pub(super) transcript_path: Option<PathBuf>,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "op", rename_all = "snake_case")]
pub(super) enum DaemonRequest {
    Create {
        spec: Box<WorkerLaunchSpec>,
    },
    Session {
        session_id: String,
    },
    /// Return canonicalized CWDs for every live session snapshot.
    ListLiveCwds,
    Terminate {
        session_id: String,
    },
    Interrupt {
        session_id: String,
        profile: CtrlCProfile,
    },
    /// Fire-and-forget handoff: the CLI is exiting and wants the daemon
    /// to kill these process trees on a background thread so the user-
    /// visible Ctrl+C-to-shell latency stays sub-100ms. The daemon
    /// replies `AdoptKillAck` immediately (before doing the kill) so the
    /// CLI's `send_daemon_request` call returns ASAP. PIDs are typically
    /// the root spawned by `runner::run_plan_subprocess` /
    /// `run_plan_pty`.
    AdoptKill {
        pids: Vec<u32>,
        #[serde(default)]
        reason: Option<String>,
    },
    /// Issue #135: GC ops served by the registry worker thread (see
    /// `gc_service.rs`). Carry the original `gc.*` op inside a single
    /// enum variant so the wire format and the registry-worker dispatch
    /// share one definition. (Field is `payload` rather than `op` so it
    /// doesn't collide with the outer `#[serde(tag = "op")]`.)
    Gc {
        payload: GcOp,
    },
    /// Ask the daemon to exit after replying. Used by `clud daemon restart`.
    Shutdown,
    /// Fire-and-forget: ask the daemon to sweep every CLUD-tagged process
    /// whose originator is no longer alive and `kill_tree` each. Sent from
    /// the foreground clud's exit hook so the daemon catches descendants
    /// the foreground couldn't reap on its own (e.g., reparented away from
    /// `self_pid`, or left behind by a SIGKILL'd sibling clud). The daemon
    /// replies `ReapOrphansAck` immediately and does the sweep on a
    /// background thread.
    ReapOrphans,
    /// Return lightweight daemon process metrics for foreground clients
    /// that want ambient status without scraping dashboard JSON.
    Metrics,
    /// Return the daemon's most recent cached process-tree sample. The
    /// daemon must not perform synchronous process enumeration for this
    /// request; callers get whatever the background sampler last wrote.
    ProcSnapshot {
        #[serde(default)]
        include_dead_since_ms: u64,
    },
    AcquireClientLease {
        identity: ProcessIdentity,
    },
    ReleaseClientLease {
        identity: ProcessIdentity,
    },
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "op", rename_all = "snake_case")]
pub(super) enum DaemonResponse {
    Created {
        session: SessionSnapshot,
    },
    Session {
        session: SessionSnapshot,
    },
    LiveCwds {
        paths: Vec<PathBuf>,
    },
    Terminated {
        session: SessionSnapshot,
    },
    Interrupted {
        session: SessionSnapshot,
    },
    /// Ack for [`DaemonRequest::AdoptKill`]. Returned before the daemon
    /// actually performs the kill, so the CLI can exit immediately.
    AdoptKillAck {
        accepted: usize,
    },
    Gc {
        reply: GcReply,
    },
    /// Acknowledgement for `DaemonRequest::Shutdown`.
    ShutdownAck {
        pid: u32,
    },
    /// Ack for [`DaemonRequest::ReapOrphans`]. Returned before the daemon
    /// performs the kill (the sweep runs on a background thread), so the
    /// CLI's exit path is never blocked. `found` and `reaped` are 0 in the
    /// fast-path ack; populated only by the synchronous test path.
    ReapOrphansAck {
        found: u32,
        reaped: u32,
    },
    Metrics {
        pid: u32,
        cpu_pct: f32,
    },
    ProcSnapshot {
        snapshot: ProcTreeSnapshot,
    },
    ClientLeaseAcquired {
        lease_count: usize,
    },
    ClientLeaseReleased {
        lease_count: usize,
    },
    Error {
        message: String,
    },
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub(super) enum ProcTier {
    Hot,
    Warm,
    Cold,
    Frozen,
}

#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub(super) struct ProcTreeSummary {
    pub(super) process_count: usize,
    pub(super) originator_count: usize,
    pub(super) total_cpu_pct: f32,
    pub(super) total_rss_bytes: u64,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub(super) struct ProcRow {
    pub(super) pid: u32,
    pub(super) ppid: Option<u32>,
    pub(super) originator: String,
    pub(super) originator_pid: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(super) session_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(super) session_name: Option<String>,
    pub(super) cpu_pct: f32,
    pub(super) cpu_ewma_pct: f32,
    pub(super) rss_bytes: u64,
    pub(super) age_secs: u64,
    pub(super) command: String,
    pub(super) depth: u32,
    pub(super) tier: ProcTier,
    #[serde(default = "default_live_proc_row")]
    pub(super) live: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(super) exited_at_ms: Option<u64>,
}

fn default_live_proc_row() -> bool {
    true
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub(super) struct ProcTreeSnapshot {
    pub(super) schema_version: u32,
    pub(super) sampled_at_ms: u64,
    pub(super) sample_age_ms: u64,
    pub(super) sampler_pid: u32,
    pub(super) interval_ms: u64,
    pub(super) rows: Vec<ProcRow>,
    pub(super) summary: ProcTreeSummary,
}

impl ProcTreeSnapshot {
    pub(super) fn empty(interval_ms: u64) -> Self {
        Self {
            schema_version: 1,
            sampled_at_ms: unix_millis_now(),
            sample_age_ms: 0,
            sampler_pid: std::process::id(),
            interval_ms,
            rows: Vec::new(),
            summary: ProcTreeSummary::default(),
        }
    }

    pub(super) fn refresh_age(&mut self) {
        self.sample_age_ms = unix_millis_now().saturating_sub(self.sampled_at_ms);
    }

    pub(super) fn recompute_summary(&mut self) {
        let mut originators = std::collections::BTreeSet::new();
        let mut total_cpu = 0.0_f32;
        let mut total_rss = 0_u64;
        for row in &self.rows {
            originators.insert(row.originator.clone());
            total_cpu += row.cpu_pct;
            total_rss = total_rss.saturating_add(row.rss_bytes);
        }
        self.summary = ProcTreeSummary {
            process_count: self.rows.len(),
            originator_count: originators.len(),
            total_cpu_pct: total_cpu,
            total_rss_bytes: total_rss,
        };
    }
}

/// Issue #135: payload carried by `DaemonRequest::Gc`. Identical in
/// shape to the standalone `gc_daemon` protocol it replaces; only the
/// outer envelope changed.
#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "gc_op", rename_all = "snake_case")]
pub(crate) enum GcOp {
    /// Issue #545: register one non-recursive daemon-owned discovery root.
    /// Registration is idempotent; the daemon canonicalizes and deduplicates
    /// roots shared by concurrent foreground clients.
    Watch {
        kind: String,
        watch_dir: String,
        #[serde(default)]
        repo_root: Option<String>,
    },
    List {
        #[serde(default)]
        kind: Option<String>,
    },
    Purge {
        /// Duration string (e.g. `"7d"`) or `None` to purge ALL non-live-locked entries.
        #[serde(default)]
        duration: Option<String>,
        #[serde(default)]
        kind: Option<String>,
        #[serde(default)]
        dry_run: bool,
    },
    Reconcile {
        repo_root: String,
    },
    Insert {
        kind: String,
        path: String,
        #[serde(default)]
        repo_root: Option<String>,
        #[serde(default)]
        branch: Option<String>,
        #[serde(default)]
        agent_id: Option<String>,
        #[serde(default)]
        created_unix: Option<i64>,
    },
    /// Issue #183: upsert a `repo_visits` row, called by every clud
    /// startup from inside a git repo. The daemon increments the
    /// per-repo run counter and records the current cwd.
    RecordRepoVisit {
        repo_root: String,
        cwd: String,
        #[serde(default)]
        now_unix: Option<i64>,
    },
    /// Issue #183: enumerate the `repo_visits` table, newest first.
    /// Powers the `repos` array in `clud ui` / `/state.json`.
    ListRepoVisits,
    /// Issue #183: surgically delete a single tracked entry by its
    /// `id`. Used by the dashboard's per-row Delete button. Runs the
    /// same on-disk removal as `Purge` (worktree-aware) but targets
    /// exactly one row regardless of how many siblings share its kind.
    DeleteById {
        id: i64,
    },
}

/// Issue #135: payload carried by `DaemonResponse::Gc`. Mirrors what the
/// registry worker emits.
#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "gc_op", rename_all = "snake_case")]
pub(crate) enum GcReply {
    /// `gc.watch` acknowledgement. The initial scan is deliberately queued
    /// after this reply so foreground launch never waits on discovery.
    WatchOk,
    ListOk {
        rows: Vec<ListRow>,
    },
    /// Synchronous purge result. Used by `Purge { dry_run: true }` and
    /// `DeleteById` — both finish on the registry worker thread before
    /// replying, so the caller learns the exact outcome inline.
    PurgeOk {
        removed: usize,
        skipped: usize,
    },
    /// Issue #268: a non-dry-run bulk `Purge` returns the moment the
    /// worker has dispatched every purgeable entry onto the parallel
    /// purge pool. The filesystem `remove_dir_all` and the matching
    /// `registry.delete(id)` happen asynchronously; this reply only
    /// reports how many tasks were dispatched (`dispatched`) and how
    /// many candidates the worker filtered out as live or non-purgeable
    /// (`skipped`). Completions stream into the worker's mpsc as
    /// `RegistryMsg::PurgeCompletion(..)` items and are logged to the
    /// daemon log. The redb registry is eventually consistent with the
    /// filesystem — stale rows during/after a partial purge are
    /// acceptable and get reconciled by the next list/purge.
    PurgeStarted {
        dispatched: usize,
        skipped: usize,
    },
    ReconcileOk {
        inserted: usize,
    },
    InsertOk {
        /// `false` is the conservative compatibility default for an old
        /// daemon that did not report whether the registry changed.
        #[serde(default)]
        inserted: bool,
    },
    /// Issue #183: ack for a successful `GcOp::RecordRepoVisit` upsert.
    RepoVisitOk,
    /// Issue #183: payload for `GcOp::ListRepoVisits`.
    RepoVisitsOk {
        rows: Vec<RepoVisit>,
    },
    Error {
        message: String,
    },
}

/// One logical GC-discovery registration. This lives on the JSON IPC wire so
/// foreground clients can hand the daemon all three legacy scanner roots in a
/// bounded, fire-and-forget burst.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct GcWatchRoot {
    pub kind: String,
    pub watch_dir: String,
    pub repo_root: Option<String>,
}

/// Row shape returned by `gc.list`. Stable JSON schema for the CLI; the
/// `clud gc list --json` output is this struct serialized as an array.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ListRow {
    pub id: i64,
    pub kind: String,
    pub path: String,
    pub repo_root: Option<String>,
    pub branch: Option<String>,
    pub agent_id: Option<String>,
    pub created_unix: i64,
    pub live_locked: bool,
    /// Issue #896: `reclaimable` | `pinned` | `dangling`.
    ///
    /// This and the fields below are `#[serde(default)]` because `ListRow`
    /// crosses the daemon wire: a new client may talk to an already-running
    /// old daemon (fields absent → defaults) just as an old client may
    /// receive them from a new one (unknown fields ignored). `live_locked`
    /// is deliberately kept rather than folded into `state`, so existing
    /// consumers keep working.
    #[serde(default = "default_state")]
    pub state: String,
    /// Short human-readable explanation, present only for non-default
    /// states (e.g. `pinned: uncommitted or unpushed work`).
    #[serde(default)]
    pub reason: Option<String>,
    /// Whether GC will take this row on a future tick.
    #[serde(default = "default_reclaimable")]
    pub reclaimable: bool,
    /// When the purge tick last evaluated this row, if ever. Lets a
    /// consumer judge how stale `state`/`reason` are — worst case one tick
    /// (1 h by default), since `gc list` never re-probes (see #946).
    #[serde(default)]
    pub evaluated_unix: Option<i64>,
}

fn default_state() -> String {
    "reclaimable".to_string()
}

fn default_reclaimable() -> bool {
    true
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "op", rename_all = "snake_case")]
pub(super) enum WorkerClientMessage {
    Attach {
        #[serde(default, skip_serializing_if = "Option::is_none")]
        terminal: Option<TerminalCapabilities>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        rows: Option<u16>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        cols: Option<u16>,
    },
    Input {
        data_b64: String,
        submit: bool,
    },
    Resize {
        rows: u16,
        cols: u16,
    },
    Interrupt {
        #[serde(default, skip_serializing_if = "Option::is_none")]
        profile: Option<CtrlCProfile>,
    },
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "op", rename_all = "snake_case")]
pub(super) enum WorkerServerMessage {
    Attached { session: Box<SessionSnapshot> },
    Output { data_b64: String },
    Exited { exit_code: i32 },
    Error { message: String },
}

#[derive(Clone)]
pub(super) enum SessionRuntime {
    Subprocess(Arc<NativeProcess>),
    Pty(Arc<NativePtyProcess>),
}

impl SessionRuntime {
    pub(super) fn root_pid(&self) -> Option<u32> {
        match self {
            Self::Subprocess(process) => process.pid(),
            Self::Pty(process) => process.pid().ok().flatten(),
        }
    }

    pub(super) fn write(&self, data: &[u8], submit: bool) {
        if let Self::Pty(process) = self {
            let _ = process.write_impl(data, submit);
        }
    }

    pub(super) fn resize(&self, rows: u16, cols: u16) {
        if let Self::Pty(process) = self {
            let _ = process.resize_impl(rows, cols);
        }
    }

    pub(super) fn cleanup_tree(&self) {
        if let Some(pid) = self.root_pid() {
            signal_process_tree(pid, Signal::Term);
            thread::sleep(Duration::from_millis(150));
            signal_process_tree(pid, Signal::Kill);
        }
        match self {
            Self::Subprocess(process) => {
                let _ = process.kill();
            }
            Self::Pty(process) => {
                let _ = process.terminate_tree_impl();
                thread::sleep(Duration::from_millis(150));
                let _ = process.kill_tree_impl();
                let _ = process.close_impl();
            }
        }
    }
}

pub(super) type AttachClientResult = (
    u64,
    mpsc::Receiver<WorkerServerMessage>,
    SessionSnapshot,
    Vec<Vec<u8>>,
);

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum LocalAttachResult {
    Completed(i32),
    InterruptRequested(LocalInterruptProfile),
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) struct LocalInterruptProfile {
    pub(super) observed_at_ms: u64,
    observed_at: Instant,
}

impl LocalInterruptProfile {
    pub(super) fn now() -> Self {
        Self {
            observed_at_ms: unix_millis_now(),
            observed_at: Instant::now(),
        }
    }

    pub(super) fn elapsed_ms(&self) -> u64 {
        self.observed_at.elapsed().as_millis() as u64
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum BackgroundPromptDecision {
    ContinueInBackground,
    EndSession,
}

pub(super) struct RawTerminalGuard;

impl RawTerminalGuard {
    pub(super) fn enter() -> io::Result<Self> {
        crossterm::terminal::enable_raw_mode()?;
        Ok(Self)
    }
}

impl Drop for RawTerminalGuard {
    fn drop(&mut self) {
        let _ = crossterm::terminal::disable_raw_mode();
    }
}

#[derive(Debug, PartialEq, Eq)]
pub(super) enum KeyAction {
    Forward(Vec<u8>),
    Interrupt,
    /// F3 went down. In centralized mode the attach pump fires
    /// `InteractiveHooks::on_f3_press`; in local-PTY mode the runner's
    /// raw pump observes the matching byte sequence directly. Both paths
    /// converge on `VoiceMode` so hold-to-record behaves the same way.
    F3Press,
    /// F3 came back up — emitted only by terminals that report key
    /// releases (kitty protocol; Windows ConPTY only when keyboard
    /// enhancement flags negotiate). Other terminals stop recording via
    /// VAD auto-stop in `VoiceMode::on_tick`.
    F3Release,
    Ignore,
}

pub(super) struct AttachedClient {
    pub(super) id: u64,
    pub(super) sender: mpsc::Sender<WorkerServerMessage>,
    pub(super) shutdown: TcpStream,
    pub(super) attached_at: Instant,
}

#[derive(Default)]
pub(super) struct BacklogState {
    pub(super) chunks: std::collections::VecDeque<Vec<u8>>,
    pub(super) total_bytes: usize,
}

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

    #[test]
    fn shutdown_request_serializes_as_tagged_op() {
        let wire = serde_json::to_string(&DaemonRequest::Shutdown).unwrap();
        assert_eq!(wire, r#"{"op":"shutdown"}"#);
    }

    #[test]
    fn shutdown_request_roundtrips() {
        let wire = serde_json::to_string(&DaemonRequest::Shutdown).unwrap();
        let parsed: DaemonRequest = serde_json::from_str(&wire).unwrap();
        assert!(matches!(parsed, DaemonRequest::Shutdown));
    }

    #[test]
    fn list_live_cwds_request_serializes_as_tagged_op() {
        let wire = serde_json::to_string(&DaemonRequest::ListLiveCwds).unwrap();
        assert_eq!(wire, r#"{"op":"list_live_cwds"}"#);
    }

    #[test]
    fn insert_ok_reply_deserializes_without_inserted_field() {
        let reply: GcReply = serde_json::from_str(r#"{"gc_op":"insert_ok"}"#).unwrap();
        assert!(matches!(reply, GcReply::InsertOk { inserted: false }));
    }

    #[test]
    fn insert_ok_reply_roundtrips_inserted_field() {
        let wire = serde_json::to_string(&GcReply::InsertOk { inserted: true }).unwrap();
        assert!(wire.contains(r#""gc_op":"insert_ok""#));
        assert!(wire.contains(r#""inserted":true"#));
    }

    #[test]
    fn watch_op_serializes_as_tagged_gc_op() {
        let op = GcOp::Watch {
            kind: "worktree".to_string(),
            watch_dir: "/tmp/worktrees".to_string(),
            repo_root: Some("/tmp/repo".to_string()),
        };
        let wire = serde_json::to_string(&op).unwrap();
        assert!(wire.contains(r#""gc_op":"watch""#));
        let parsed: GcOp = serde_json::from_str(&wire).unwrap();
        assert!(matches!(parsed, GcOp::Watch { .. }));
    }

    #[test]
    fn metrics_request_serializes_as_tagged_op() {
        let wire = serde_json::to_string(&DaemonRequest::Metrics).unwrap();
        assert_eq!(wire, r#"{"op":"metrics"}"#);
    }

    #[test]
    fn proc_snapshot_request_roundtrips_dead_window() {
        let request = DaemonRequest::ProcSnapshot {
            include_dead_since_ms: 5_000,
        };
        let wire = serde_json::to_string(&request).unwrap();
        assert!(wire.contains(r#""op":"proc_snapshot""#));
        assert!(wire.contains(r#""include_dead_since_ms":5000"#));
        let parsed: DaemonRequest = serde_json::from_str(&wire).unwrap();
        match parsed {
            DaemonRequest::ProcSnapshot {
                include_dead_since_ms,
            } => assert_eq!(include_dead_since_ms, 5_000),
            other => panic!("expected ProcSnapshot, got {other:?}"),
        }
    }

    #[test]
    fn metrics_response_roundtrips() {
        let response = DaemonResponse::Metrics {
            pid: 142500,
            cpu_pct: 72.5,
        };
        let wire = serde_json::to_string(&response).unwrap();
        assert!(wire.contains(r#""op":"metrics""#));
        assert!(wire.contains(r#""pid":142500"#));

        let parsed: DaemonResponse = serde_json::from_str(&wire).unwrap();
        match parsed {
            DaemonResponse::Metrics { pid, cpu_pct } => {
                assert_eq!(pid, 142500);
                assert!((cpu_pct - 72.5).abs() < f32::EPSILON);
            }
            other => panic!("expected Metrics, got {other:?}"),
        }
    }

    #[test]
    fn proc_snapshot_response_roundtrips() {
        let mut snapshot = ProcTreeSnapshot::empty(2_000);
        snapshot.rows.push(ProcRow {
            pid: 42,
            ppid: Some(7),
            originator: "CLUD:7".to_string(),
            originator_pid: Some(7),
            session_id: Some("sess-test".to_string()),
            session_name: None,
            cpu_pct: 12.0,
            cpu_ewma_pct: 8.0,
            rss_bytes: 1024,
            age_secs: 5,
            command: "backend".to_string(),
            depth: 1,
            tier: ProcTier::Hot,
            live: true,
            exited_at_ms: None,
        });
        snapshot.recompute_summary();
        let response = DaemonResponse::ProcSnapshot {
            snapshot: snapshot.clone(),
        };
        let wire = serde_json::to_string(&response).unwrap();
        assert!(wire.contains(r#""op":"proc_snapshot""#));
        assert!(wire.contains(r#""schema_version":1"#));

        let parsed: DaemonResponse = serde_json::from_str(&wire).unwrap();
        match parsed {
            DaemonResponse::ProcSnapshot { snapshot: got } => assert_eq!(got, snapshot),
            other => panic!("expected ProcSnapshot, got {other:?}"),
        }
    }

    #[test]
    fn live_cwds_response_roundtrips_with_paths() {
        let response = DaemonResponse::LiveCwds {
            paths: vec![PathBuf::from("/tmp/live-a"), PathBuf::from("/tmp/live-b")],
        };
        let wire = serde_json::to_string(&response).unwrap();
        assert!(wire.contains(r#""op":"live_cwds""#));
        assert!(wire.contains(r#"/tmp/live-a"#));

        let parsed: DaemonResponse = serde_json::from_str(&wire).unwrap();
        match parsed {
            DaemonResponse::LiveCwds { paths } => {
                assert_eq!(
                    paths,
                    vec![PathBuf::from("/tmp/live-a"), PathBuf::from("/tmp/live-b")]
                );
            }
            other => panic!("expected LiveCwds, got {other:?}"),
        }
    }

    #[test]
    fn attach_request_accepts_missing_terminal_metadata() {
        let parsed: WorkerClientMessage = serde_json::from_str(r#"{"op":"attach"}"#).unwrap();
        assert!(matches!(
            parsed,
            WorkerClientMessage::Attach {
                terminal: None,
                rows: None,
                cols: None
            }
        ));
    }

    #[test]
    fn interrupt_request_accepts_legacy_unit_shape() {
        let parsed: WorkerClientMessage = serde_json::from_str(r#"{"op":"interrupt"}"#).unwrap();
        assert!(matches!(
            parsed,
            WorkerClientMessage::Interrupt { profile: None }
        ));
    }

    #[test]
    fn interrupt_request_serializes_ctrl_c_profile() {
        let wire = serde_json::to_string(&WorkerClientMessage::Interrupt {
            profile: Some(CtrlCProfile {
                cli_pid: Some(42),
                cli_observed_at_ms: Some(1000),
                cli_handoff_at_ms: Some(1010),
                cli_return_ready_at_ms: Some(1010),
                cli_handoff_ms: Some(10),
                fast_path: true,
                ..CtrlCProfile::default()
            }),
        })
        .unwrap();
        assert!(wire.contains(r#""op":"interrupt""#));
        assert!(wire.contains(r#""cli_handoff_ms":10"#));
        assert!(wire.contains(r#""fast_path":true"#));
    }

    #[test]
    fn adopt_kill_request_roundtrips_pids_and_reason() {
        let request = DaemonRequest::AdoptKill {
            pids: vec![1234, 5678],
            reason: Some("ctrl_c_handoff".to_string()),
        };
        let wire = serde_json::to_string(&request).unwrap();
        assert!(wire.contains(r#""op":"adopt_kill""#));
        assert!(wire.contains(r#""pids":[1234,5678]"#));
        let parsed: DaemonRequest = serde_json::from_str(&wire).unwrap();
        match parsed {
            DaemonRequest::AdoptKill { pids, reason } => {
                assert_eq!(pids, vec![1234, 5678]);
                assert_eq!(reason.as_deref(), Some("ctrl_c_handoff"));
            }
            other => panic!("expected AdoptKill, got {other:?}"),
        }
    }

    #[test]
    fn adopt_kill_request_omits_reason_when_none() {
        let request = DaemonRequest::AdoptKill {
            pids: vec![99],
            reason: None,
        };
        let wire = serde_json::to_string(&request).unwrap();
        assert!(wire.contains(r#""pids":[99]"#));
        // `reason` is `#[serde(default)]`; absence on the wire round-trips.
        let parsed: DaemonRequest =
            serde_json::from_str(r#"{"op":"adopt_kill","pids":[99]}"#).unwrap();
        match parsed {
            DaemonRequest::AdoptKill { pids, reason } => {
                assert_eq!(pids, vec![99]);
                assert!(reason.is_none());
            }
            other => panic!("expected AdoptKill, got {other:?}"),
        }
    }

    #[test]
    fn adopt_kill_ack_response_roundtrips() {
        let response = DaemonResponse::AdoptKillAck { accepted: 2 };
        let wire = serde_json::to_string(&response).unwrap();
        assert!(wire.contains(r#""op":"adopt_kill_ack""#));
        assert!(wire.contains(r#""accepted":2"#));
        let parsed: DaemonResponse = serde_json::from_str(&wire).unwrap();
        assert!(matches!(
            parsed,
            DaemonResponse::AdoptKillAck { accepted: 2 }
        ));
    }

    #[test]
    fn shutdown_ack_response_roundtrips_with_pid() {
        let response = DaemonResponse::ShutdownAck { pid: 142500 };
        let wire = serde_json::to_string(&response).unwrap();
        assert!(wire.contains(r#""op":"shutdown_ack""#));
        assert!(wire.contains(r#""pid":142500"#));

        let parsed: DaemonResponse = serde_json::from_str(&wire).unwrap();
        match parsed {
            DaemonResponse::ShutdownAck { pid } => assert_eq!(pid, 142500),
            other => panic!("expected ShutdownAck, got {other:?}"),
        }
    }

    /// `ListRow` crosses the daemon wire, and a new client routinely talks
    /// to an already-running older daemon (the daemon outlives a `clud`
    /// upgrade). Issue #896's new fields must therefore be optional on the
    /// way in, or `clud gc list` breaks entirely against an old daemon.
    #[test]
    fn list_row_deserializes_from_a_pre_896_daemon() {
        let old_wire = r#"{
            "id": 7,
            "kind": "extern-repo",
            "path": "/tmp/checkout",
            "repo_root": null,
            "branch": null,
            "agent_id": null,
            "created_unix": 1700000000,
            "live_locked": false
        }"#;
        let row: ListRow = serde_json::from_str(old_wire).expect("old wire must still parse");
        assert_eq!(row.id, 7);
        // Absent state must read as the neutral default, never as a
        // fabricated "pinned"/"dangling" verdict.
        assert_eq!(row.state, "reclaimable");
        assert!(row.reclaimable);
        assert_eq!(row.reason, None);
        assert_eq!(row.evaluated_unix, None);
    }

    /// Guards against a future `deny_unknown_fields` being added: serde
    /// ignores unknown keys by default, and that default is what lets an
    /// older client keep parsing rows from a newer daemon. (This
    /// deserializes into the current `ListRow`, so it exercises the
    /// tolerance, not a genuinely old client.)
    #[test]
    fn list_row_tolerates_unknown_future_fields() {
        let future_wire = r#"{
            "id": 1, "kind": "worktree", "path": "/tmp/w",
            "repo_root": null, "branch": null, "agent_id": null,
            "created_unix": 1, "live_locked": false,
            "state": "pinned", "reason": "pinned: something",
            "reclaimable": false, "evaluated_unix": 42,
            "a_field_from_the_future": {"nested": true}
        }"#;
        let row: ListRow = serde_json::from_str(future_wire).expect("unknown fields are ignored");
        assert_eq!(row.state, "pinned");
        assert_eq!(row.reason.as_deref(), Some("pinned: something"));
        assert!(!row.reclaimable);
        assert_eq!(row.evaluated_unix, Some(42));
    }
}
