//! Orchestration for souveraine-sessiond: the control socket, the shell
//! heartbeat, and the retake state machine.
//!
//! Sync, std-only, thread-per-connection — the same auditable shape as
//! machined's server. The daemon's whole job is to make sure that at every
//! moment either (a) it holds the session lock, (b) a live shell does, or
//! (c) the user deliberately unlocked at the fallback surface. Transitions
//! between those states are the only logic here.

use std::io::{BufRead, BufReader, Read, Write};
use std::os::unix::fs::PermissionsExt;
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Condvar, Mutex};
use std::time::Duration;

use anyhow::{Context, Result};
use tracing::{info, warn};

use crate::sessiond::device_state::{Action, DeviceState, DeviceStateMachine, SensorEvidence};
use crate::sessiond::idle;
use crate::sessiond::lock::{self, LockController, Msg, SessionOutcome};
use crate::sessiond::lockhint;
use crate::sessiond::protocol::{
    Envelope, InputTrigger, Phase, RefusalCode, Request, SensorSource, SensorValue,
    LOCKED_ACK_TIMEOUT_SECS, MAX_REQUEST_BYTES, VERBS,
};

/// How often the machine's clock runs. One second is fine: the shortest
/// budget it enforces is measured in seconds, and a locked idle phone should
/// not be woken more often than it must be.
const TICK_INTERVAL: Duration = Duration::from_secs(1);

/// How long a connection may sit without sending a request before it is closed.
///
/// Exempts the registered shell heartbeat, whose whole job is to stay silent.
/// Generous on purpose: this is a leak backstop, not a policy. Every real
/// client here either speaks immediately or is the heartbeat.
const IDLE_CONNECTION_TIMEOUT_SECS: u64 = 120;

/// The one userspace authority for panel DPMS on blueline. sessiond decides;
/// this serializes the transition and owns the FTS pacing lesson.
const DPMS_EXECUTOR: &str = "blueline-screen-toggle";

/// Restores brightness after a dim, with the 10% floor. Never a bare
/// `brightnessctl -r` — see Pixel3Arch b9f83f0.
const UNDIM_EXECUTOR: &str = "blueline-undim";

struct Daemon {
    phase: Phase,
    /// Present exactly while a lock-session thread is running.
    controller: Option<LockController>,
    /// Bumped on every shell registration; a heartbeat EOF only counts if
    /// its generation is still current (a restarted shell supersedes).
    heartbeat_gen: u64,
    shell_alive: bool,
    /// Write handle to the live shell's heartbeat connection, so the authority
    /// can push a directive down it instead of only answering requests. The
    /// shell owns the rich lock surface in steady state; this is how the
    /// daemon that owns the *decision* reaches the process that owns the
    /// *surface*.
    shell_directives: Option<UnixStream>,
    pam_service: String,
    /// The unified device state machine. The single authority for device
    /// power state — idle, lock, doze, sleep. All actors route through it.
    device_state: DeviceStateMachine,
}

pub struct Shared {
    state: Mutex<Daemon>,
    cond: Condvar,
}

impl Shared {
    fn lock(&self) -> std::sync::MutexGuard<'_, Daemon> {
        self.state.lock().unwrap_or_else(|e| e.into_inner())
    }
}

pub fn socket_path() -> Result<PathBuf> {
    let runtime = std::env::var("XDG_RUNTIME_DIR").context("XDG_RUNTIME_DIR is not set")?;
    Ok(PathBuf::from(runtime).join(crate::sessiond::protocol::SOCKET_RELPATH))
}

pub fn run(pam_service: String, initial_lock: bool, socket_path: &Path) -> Result<()> {
    if let Some(parent) = socket_path.parent() {
        std::fs::create_dir_all(parent)
            .with_context(|| format!("creating {}", parent.display()))?;
        std::fs::set_permissions(parent, std::fs::Permissions::from_mode(0o700)).ok();
    }
    if socket_path.exists() {
        if UnixStream::connect(socket_path).is_ok() {
            anyhow::bail!(
                "another souveraine-sessiond is already serving {}",
                socket_path.display()
            );
        }
        warn!("removing stale socket at {}", socket_path.display());
        std::fs::remove_file(socket_path)?;
    }
    let listener = UnixListener::bind(socket_path)
        .with_context(|| format!("binding {}", socket_path.display()))?;
    // Same-user only today. When agents get their own accounts, this widens
    // to a group grant deliberately, machined-style — never by default.
    std::fs::set_permissions(socket_path, std::fs::Permissions::from_mode(0o600))?;

    let shared = Arc::new(Shared {
        state: Mutex::new(Daemon {
            phase: Phase::Idle,
            controller: None,
            heartbeat_gen: 0,
            shell_alive: false,
            shell_directives: None,
            pam_service,
            device_state: DeviceStateMachine::new(),
        }),
        cond: Condvar::new(),
    });

    if initial_lock {
        spawn_lock_session(&shared);
    }
    spawn_clock(&shared);
    spawn_idle_watcher(&shared);
    spawn_lock_hint_watcher(&shared);
    info!("sessiond serving on {}", socket_path.display());

    for stream in listener.incoming() {
        match stream {
            Ok(s) => {
                let shared = Arc::clone(&shared);
                std::thread::spawn(move || handle_connection(s, shared));
            }
            Err(e) => warn!("accept failed: {e}"),
        }
    }
    Ok(())
}

/// The machine's clock. Ticks the state machine and carries out whatever it
/// decided.
///
/// The daemon had no periodic anything before this: every `Duration` in it was
/// a one-shot handshake deadline. That is the whole reason the state machine
/// never moved — it could answer requests, but nothing ever told it that time
/// had passed, so it could not be the proactive half of the nervous system.
fn spawn_clock(shared: &Arc<Shared>) {
    let shared = Arc::clone(shared);
    std::thread::spawn(move || loop {
        std::thread::sleep(TICK_INTERVAL);
        // Hold the lock only to decide, never while running a command.
        let actions = {
            let mut d = shared.lock();
            d.device_state.tick()
        };
        for action in actions {
            execute(&shared, action);
        }
    });
}

/// Feed compositor input truth into the machine.
///
/// Without this the machine has a clock but no idea whether anyone is
/// touching the device, and the only other actor holding that knowledge is
/// hypridle — which is why the lock screen's timeout lived in hypridle's
/// config with a single 600 s timer instead of in the authority that owns the
/// lock.
fn spawn_idle_watcher(shared: &Arc<Shared>) {
    let shared = Arc::clone(shared);
    idle::spawn(Arc::new(move |event| match event {
        idle::IdleEvent::Idled { since } => {
            shared.lock().device_state.note_idle_start(since);
        }
        idle::IdleEvent::Resumed => {
            let actions = {
                let mut d = shared.lock();
                d.device_state.note_input(InputTrigger::Unknown)
            };
            for action in actions {
                execute(&shared, action);
            }
        }
    }));
}

/// Feed logind's lock truth into the machine.
///
/// Without this the machine tracked `locked` itself and had no unlock ingress,
/// so after the first unlock it believed the session was locked forever — and
/// then blanked an in-use phone on proximity, because the proximity rule is
/// gated on `is_locked`. Doctrine §4: never hold state the protocol owns.
fn spawn_lock_hint_watcher(shared: &Arc<Shared>) {
    let shared = Arc::clone(shared);
    lockhint::spawn(Arc::new(move |locked| {
        let actions = {
            let mut d = shared.lock();
            d.device_state.set_session_locked(locked)
        };
        for action in actions {
            execute(&shared, action);
        }
    }));
}

/// Carry out one decision. Failures are recorded in the forensic trail — a
/// blank that did not happen is a state divergence, not a log line to lose.
/// Ask logind whether this session is locked. Fail closed: if the question
/// cannot be answered, the answer is "not locked", because the failure this
/// guards is a dark screen on an open session.
fn session_is_locked() -> bool {
    match crate::sessiond::lockhint::resolve_session_path()
        .and_then(|p| crate::sessiond::lockhint::read_locked_hint(&p))
    {
        Ok(locked) => locked,
        Err(e) => {
            warn!("could not read LockedHint ({e:#}); treating the session as unlocked");
            false
        }
    }
}

fn execute(shared: &Arc<Shared>, action: Action) {
    // The lock is not a shell command, so it does not go through the executor
    // table below. It is the authority acting as the authority: the shell owns
    // the rich lock surface in steady state and executes the directive, and
    // sessiond takes the lock itself when no shell is alive to do it. Either
    // way the decision is the daemon's — doctrine §11, the session authority
    // and the binary authority are the same authority, and it is not the UID
    // that makes it one (§1: everything running as you is you).
    if action == Action::Lock {
        request_session_lock(shared);
        return;
    }

    // Dim and restore carry a number, so they are not table entries. The
    // machine remembers the exact brightness the panel was at; the executor
    // reads the hardware and sets it back. Neither `brightnessctl -s/-r` nor
    // the floor in `blueline-undim` can tell a mistakenly-saved dim value from
    // a phone the user deliberately runs dark — that guess is what made a
    // tap-to-dismiss come back at a different level than it started.
    if action == Action::Dim {
        let before = read_brightness();
        if before.is_none() {
            warn!("could not read brightness before dimming — restore will use the floor");
        }
        shared
            .lock()
            .device_state
            .note_brightness_before_dim(before);
    }
    if action == Action::Restore {
        // Bind FIRST, then branch. `if let Some(v) = shared.lock()...` keeps the
        // temporary MutexGuard alive for the whole if-let body on edition 2021,
        // and the body calls run_executor, which locks again — a self-deadlock
        // on a non-reentrant mutex. It held the state lock forever, so every
        // later request thread piled up behind it: measured on the phone
        // 2026-07-26, 347 threads in __futex_wait and 691 of 1024 fds, with the
        // accept loop still healthy and not one request ever answered.
        let saved = shared.lock().device_state.take_brightness_before_dim();
        if let Some(value) = saved {
            run_executor(
                shared,
                "brightnessctl",
                &["set", &value.to_string()],
                "panel-restore",
                action,
            );
            return;
        }
        // No capture. This is the only case the floor is right for: the panel
        // may be sitting at the dim level with nothing that knows better.
        run_executor(shared, UNDIM_EXECUTOR, &[], "panel-restore", action);
        return;
    }

    let (program, args, label): (&str, Vec<&str>, &str) = match action {
        // sessiond guarantees this runs once per dim, which is what the old
        // hypridle `-s`/`-r` listener could not: a second save while already
        // dim is what pinned brightness at 10/255.
        Action::Dim => ("brightnessctl", vec!["set", "10"], "panel-dim"),
        Action::Restore => unreachable!("handled above; the restore carries a value"),
        Action::Blank => {
            // The invariant, enforced where it cannot be reasoned around: the
            // panel does not go dark unless logind says this session is
            // locked, right now. The machine's own `Locked` is a belief, and a
            // belief that entered by one route (locked_ack) and can only leave
            // by another (logind's falling edge) gets stuck — observed
            // 2026-07-26: machine `locked`, LockedHint `no`, panel off, and
            // double-tap waking straight to an unlocked screen.
            //
            // Checked at the actuator, not at the decision, because every path
            // to a dark panel ends here and only here.
            if !session_is_locked() {
                warn!("refusing to blank: logind says this session is not locked");
                shared.lock().device_state.record_error(
                    "device-state",
                    "panel-off",
                    "refused: session not locked per logind",
                );
                request_session_lock(shared);
                return;
            }
            (DPMS_EXECUTOR, vec!["off"], "panel-off")
        }
        Action::Lock => unreachable!("handled above; the lock is not a shell command"),
    };

    run_executor(shared, program, &args, label, action);
}

/// Act on the machine's decision that the session must lock before the panel
/// goes dark.
///
/// Two paths, one authority. While a live shell owns steady state it holds the
/// rich lock surface, so the directive goes down the heartbeat connection and
/// the shell executes it — the same relationship §6 of `DEVICE-STATE-MACHINE.md`
/// gives the DPMS executor ("receives commands from the unified state machine,
/// not from independent actors"). With no shell alive there is nothing to
/// delegate to and sessiond raises its own surface.
///
/// A failure here is never swallowed. If the directive cannot be written the
/// daemon takes the lock itself rather than letting the pending blank time out
/// into a dark unlocked screen, and the attempt is recorded either way.
fn request_session_lock(shared: &Arc<Shared>) {
    let directive = serde_json::json!({ "directive": "lock", "why": "blank-requires-lock" });
    let mut line = directive.to_string();
    line.push('\n');

    let mut d = shared.lock();
    if d.shell_alive {
        if let Some(conn) = d.shell_directives.as_mut() {
            match conn.write_all(line.as_bytes()).and_then(|_| conn.flush()) {
                Ok(()) => {
                    info!("[device-state] lock directive sent to the shell");
                    d.device_state.record_decision(
                        "lock-directive",
                        serde_json::json!({ "target": "shell" }),
                        "authority told the shell to raise its lock surface",
                    );
                    return;
                }
                Err(e) => {
                    warn!("could not push lock directive to the shell: {e}");
                    d.device_state
                        .record_error("device-state", "lock-directive", &e.to_string());
                }
            }
        } else {
            warn!("shell is alive but no directive channel — taking the lock here");
            d.device_state.record_error(
                "device-state",
                "lock-directive",
                "shell registered without a directive channel",
            );
        }
    }
    drop(d);

    if !spawn_lock_session(shared) {
        warn!("could not start a lock session for a pending blank");
        shared.lock().device_state.record_error(
            "device-state",
            "lock-for-blank",
            "no lock session could be started",
        );
    }
}

/// Drive the machine to a tier. Kept separate from the lock plumbing so the
/// call sites read as one line each.
fn set_tier(d: &mut Daemon, next: DeviceState, why: &str) {
    if d.device_state.state() == next {
        return;
    }
    if !d.device_state.transition(next) {
        warn!("[device-state] {why}: refused {:?}", next);
    }
}

/// Start a lock session unless one is already running. Returns whether a
/// session is running when we leave (either ours or a pre-existing one).
fn spawn_lock_session(shared: &Arc<Shared>) -> bool {
    let mut d = shared.lock();
    if d.controller.is_some() {
        return true;
    }
    let (controller, rx, wake_read) = match lock::channel() {
        Ok(t) => t,
        Err(e) => {
            warn!("cannot create lock channel: {e:#}");
            return false;
        }
    };
    // The auth worker feeds verdicts back into the SAME session's channel.
    let auth_controller = match controller.try_clone() {
        Ok(c) => c,
        Err(e) => {
            warn!("cannot clone lock controller: {e:#}");
            return false;
        }
    };
    d.controller = Some(controller);
    d.phase = Phase::Holding;
    // sessiond itself now holds the lock — boot lock, retake after a shell
    // death, or an explicit `lock` request. All three are the locked tier.
    set_tier(&mut d, DeviceState::Locked, "sessiond holds the lock");
    drop(d);

    let shared = Arc::clone(shared);
    std::thread::spawn(move || {
        let pam_service = shared.lock().pam_service.clone();
        let outcome = lock::run(rx, wake_read, auth_controller, pam_service);
        let mut d = shared.lock();
        d.controller = None;
        match outcome {
            Ok(SessionOutcome::Unlocked) => {
                info!("session unlocked at fallback surface");
                d.phase = if d.shell_alive {
                    Phase::Released
                } else {
                    Phase::Idle
                };
                // PAM succeeded. Only this path may leave the locked tier.
                set_tier(&mut d, DeviceState::Active, "unlocked at fallback surface");
            }
            Ok(SessionOutcome::Released) => {
                d.phase = Phase::AwaitingShellLock;
                let gen = d.heartbeat_gen;
                drop(d);
                start_ack_timer(&shared, gen);
                shared.cond.notify_all();
                return;
            }
            Ok(SessionOutcome::Denied) => {
                info!("lock denied/revoked — another locker owns the session");
                d.phase = if d.shell_alive {
                    Phase::Released
                } else {
                    Phase::Idle
                };
            }
            Err(e) => {
                warn!("lock session failed: {e:#}");
                d.phase = if d.shell_alive {
                    Phase::Released
                } else {
                    Phase::Idle
                };
            }
        }
        drop(d);
        shared.cond.notify_all();
    });
    true
}

/// If the shell never acks its lock after a handoff, take the lock back.
/// A retake against a shell that DID lock is refused by the compositor
/// (Denied outcome), so this is safe against a late or lost ack.
fn start_ack_timer(shared: &Arc<Shared>, gen: u64) {
    let shared = Arc::clone(shared);
    std::thread::spawn(move || {
        std::thread::sleep(Duration::from_secs(LOCKED_ACK_TIMEOUT_SECS));
        let d = shared.lock();
        if d.phase == Phase::AwaitingShellLock && d.heartbeat_gen == gen {
            warn!("shell never confirmed its lock after handoff — retaking");
            drop(d);
            spawn_lock_session(&shared);
        }
    });
}

/// A refusal a caller can branch on.
///
/// `code` is the class, `reason` is this particular one. Doctrine §13: an
/// agent composing verbs needs to know *which kind* of no it got — not now,
/// never, or wrong arguments are three different next moves and a sentence
/// does not distinguish them.
/// The panel's current brightness, or `None` if it cannot be read.
///
/// Read rather than remembered from the last set: the user, the shell, or an
/// agent may have changed it since, and restoring a value the machine assumed
/// is how the panel ends up somewhere nobody asked for.
fn read_brightness() -> Option<u32> {
    let out = std::process::Command::new("brightnessctl")
        .arg("get")
        .output()
        .ok()?;
    if !out.status.success() {
        return None;
    }
    String::from_utf8(out.stdout).ok()?.trim().parse().ok()
}

/// Run one executor and record what it did. Shared by the paths that build
/// their arguments dynamically and the table below.
fn run_executor(
    shared: &Arc<Shared>,
    program: &str,
    args: &[&str],
    label: &str,
    action: Action,
) {
    match std::process::Command::new(program).args(args).status() {
        Ok(status) if status.success() => {
            if action == Action::Blank {
                let follow_up = shared.lock().device_state.set_panel(false);
                for a in follow_up {
                    execute(shared, a);
                }
            }
        }
        Ok(status) => {
            let code = status.code().unwrap_or(-1).to_string();
            warn!("{program} {args:?} exited {code}");
            shared
                .lock()
                .device_state
                .record_error("device-state", label, &code);
        }
        Err(e) => {
            warn!("could not run {program}: {e}");
            shared
                .lock()
                .device_state
                .record_error("device-state", label, &e.to_string());
        }
    }
}

fn refuse(code: RefusalCode, reason: &str) -> serde_json::Value {
    serde_json::json!({ "ok": false, "code": code.as_str(), "reason": reason })
}

fn respond(writer: &mut UnixStream, value: serde_json::Value) -> std::io::Result<()> {
    writer.write_all(value.to_string().as_bytes())?;
    writer.write_all(b"\n")?;
    writer.flush()
}

fn handle_connection(stream: UnixStream, shared: Arc<Shared>) {
    let _ = stream.set_write_timeout(Some(Duration::from_secs(10)));
    // A read timeout as well as a write one. Without it a client that connects
    // and says nothing parks this thread and its two fds forever, and 691 of
    // them reached two thirds of the fd limit before anyone noticed. Idle
    // connections are cheap to re-establish; leaked ones are not.
    let _ = stream.set_read_timeout(Some(Duration::from_secs(IDLE_CONNECTION_TIMEOUT_SECS)));
    let mut writer = match stream.try_clone() {
        Ok(w) => w,
        Err(e) => {
            warn!("could not clone stream: {e}");
            return;
        }
    };
    let mut reader = BufReader::new(stream);
    // Whether THIS connection is the registered shell heartbeat, and under
    // which generation.
    let mut heartbeat: Option<u64> = None;

    loop {
        let mut line = String::new();
        match (&mut reader).take(MAX_REQUEST_BYTES).read_line(&mut line) {
            Ok(0) => break, // EOF
            Ok(_) => {}
            // A read timeout is not an error for the shell's heartbeat. That
            // connection is SUPPOSED to sit silent — its silence is the
            // liveness signal and its EOF is shell death, so dropping it here
            // would manufacture exactly the false shell-death lock the
            // registration path guards against. Any other silent connection is
            // a leak and gets closed.
            Err(e)
                if matches!(
                    e.kind(),
                    std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
                ) =>
            {
                if heartbeat.is_some() {
                    continue;
                }
                warn!(
                    "closing idle connection after {IDLE_CONNECTION_TIMEOUT_SECS}s with no request"
                );
                break;
            }
            Err(e) => {
                warn!("read error on control socket: {e}");
                break;
            }
        }
        if line.len() as u64 >= MAX_REQUEST_BYTES {
            let _ = respond(&mut writer, refuse(RefusalCode::InvalidArgument, "request exceeds size limit"));
            break;
        }
        let line = line.trim();
        if line.is_empty() {
            continue;
        }
        let response = match serde_json::from_str::<Envelope>(line) {
            Ok(env) => {
                // The intent is set for exactly this request and cleared after,
                // so an entry recorded by the clock thread between two requests
                // is never mislabelled with a chain it had nothing to do with.
                shared
                    .lock()
                    .device_state
                    .forensic
                    .set_intent(env.intent.clone());
                let out = handle_request(env.request, &shared, Some(&writer), &mut heartbeat);
                shared.lock().device_state.forensic.set_intent(None);
                out
            }
            Err(e) => refuse(RefusalCode::UnsupportedOp, &format!("malformed request: {e}")),
        };
        if respond(&mut writer, response).is_err() {
            break;
        }
    }

    // Connection gone. If it was the live heartbeat, the shell died —
    // fail closed: take the session lock regardless of prior lock state.
    if let Some(gen) = heartbeat {
        let mut d = shared.lock();
        if d.heartbeat_gen == gen && d.shell_alive {
            d.shell_alive = false;
            d.shell_directives = None;
            let holding = d.controller.is_some();
            drop(d);
            warn!("shell heartbeat lost — locking the session");
            if !holding {
                spawn_lock_session(&shared);
            }
        }
    }
}

fn handle_request(
    req: Request,
    shared: &Arc<Shared>,
    directives: Option<&UnixStream>,
    heartbeat: &mut Option<u64>,
) -> serde_json::Value {
    match req {
        Request::Describe => {
            serde_json::json!({
                "ok": true,
                "verbs": VERBS.iter().map(|v| serde_json::json!({
                    "op": v.op,
                    "mutates": v.mutates,
                    "summary": v.summary,
                    "refuses": v.refuses.iter().map(|c| c.as_str()).collect::<Vec<_>>(),
                    "example": v.example,
                })).collect::<Vec<_>>(),
            })
        }
        Request::Status => {
            let d = shared.lock();
            serde_json::json!({
                "ok": true,
                "phase": d.phase,
                "shell_alive": d.shell_alive,
            })
        }
        Request::ShellReady => {
            let mut d = shared.lock();

            // Session authority is an exclusive lease. A utility QML process
            // or duplicate shell must never supersede the live supervised
            // shell merely by sending shell_ready: doing so turns that
            // process's EOF into a false shell-death lock. The current
            // authority must disconnect first; its connection handler clears
            // shell_alive before a replacement can register.
            if d.shell_alive {
                return refuse(RefusalCode::RefusedByState, "shell authority is already registered");
            }
            d.heartbeat_gen += 1;
            d.shell_alive = true;
            d.shell_directives = directives.and_then(|s| s.try_clone().ok());
            let gen = d.heartbeat_gen;
            *heartbeat = Some(gen);

            let holding = d.controller.is_some();
            if holding {
                if let Some(ctl) = &d.controller {
                    ctl.send(Msg::Release);
                }
                // Wait until the lock-session thread actually dropped its
                // Wayland connection; only then may the shell lock (the
                // compositor refuses a second locker while ours is alive).
                let deadline = Duration::from_secs(5);
                let (_guard, timed_out) = shared
                    .cond
                    .wait_timeout_while(d, deadline, |d| d.controller.is_some())
                    .unwrap_or_else(|e| e.into_inner());
                if timed_out.timed_out() {
                    return refuse(RefusalCode::Unavailable, "lock session did not release in time");
                }
                info!("handoff: lock released to shell (gen {gen})");
                serde_json::json!({ "ok": true, "held": true, "must_lock": true })
            } else {
                info!("shell registered (gen {gen}), no lock held");
                if d.phase == Phase::Idle {
                    d.phase = Phase::Released;
                }
                serde_json::json!({ "ok": true, "held": false, "must_lock": false })
            }
        }
        Request::LockedAck => {
            let mut d = shared.lock();
            if heartbeat.is_none() {
                return refuse(RefusalCode::RefusedByState, "locked_ack from a connection that never sent shell_ready");
            }
            d.phase = Phase::Released;
            info!("shell lock confirmed");
            // The compositor has acknowledged a secure lock surface. THIS is
            // the transition nothing ever made: without it the machine stayed
            // in Active for the life of the daemon, so every locked-state rule
            // below it was unreachable.
            set_tier(&mut d, DeviceState::Locked, "shell lock confirmed");
            // The handoff is a deliberate connection drop, but Hyprland
            // announces any dying lock client as a crashed lockscreen.
            // The shell's lock is confirmed, so clear that banner.
            let _ = std::process::Command::new("hyprctl")
                .arg("dismissnotify")
                .spawn();
            serde_json::json!({ "ok": true })
        }
        Request::Lock => {
            let d = shared.lock();
            match d.phase {
                Phase::Holding => serde_json::json!({ "ok": true, "already": true }),
                Phase::Released | Phase::AwaitingShellLock if d.shell_alive => {
                    refuse(RefusalCode::RefusedByState, "a live shell owns the session lock; use the shell's lock IPC")
                }
                _ => {
                    drop(d);
                    if spawn_lock_session(shared) {
                        serde_json::json!({ "ok": true })
                    } else {
                        refuse(RefusalCode::Unavailable, "could not start a lock session")
                    }
                }
            }
        }
        Request::DeviceState => {
            let d = shared.lock();
            let mut resp = d.device_state.to_ipc_json();
            resp["ok"] = serde_json::Value::Bool(true);
            resp["phase"] = serde_json::to_value(d.phase).unwrap_or_default();
            resp["shell_alive"] = serde_json::Value::Bool(d.shell_alive);
            resp
        }
        Request::SensorInput(input) => {
            let mut d = shared.lock();
            let evidence = &mut d.device_state.sensor_evidence;
            match (&input.source, &input.value) {
                (SensorSource::Proximity, SensorValue::Near(v)) => {
                    evidence.proximity_near = *v;
                }
                (SensorSource::Accelerometer, SensorValue::Moving(v)) => {
                    evidence.accel_moving = *v;
                }
                (SensorSource::Light, SensorValue::Changing(v)) => {
                    evidence.light_changing = *v;
                }
                (SensorSource::Touch, SensorValue::Active(v)) => {
                    evidence.touch_active = *v;
                }
                _ => {
                    return refuse(RefusalCode::InvalidArgument, "sensor source/value mismatch");
                }
            }
            let conf = evidence.confidence();
            let promote = evidence.should_promote_idle_faster();
            // Copy the readings out, ending the borrow before the machine is
            // touched again.
            let evidence_copy = SensorEvidence {
                proximity_near: evidence.proximity_near,
                accel_moving: evidence.accel_moving,
                light_changing: evidence.light_changing,
                touch_active: evidence.touch_active,
            };
            // Stamp the source as fresh. Without this every reading expires
            // on the next tick, because "never reported" and "reported long
            // ago" are the same thing to the staleness rule.
            d.device_state.mark_evidence_seen(input.source);
            // Let the state machine evaluate Observed transitions.
            d.device_state
                .update_sensors_from(input.source, evidence_copy);
            // The tap-to-wake answer specifically; a power button is never
            // refused. Asked after the update, not before.
            let suppress = d
                .device_state
                .suppress_wake(InputTrigger::DoubleTapToWake);
            info!(
                "[device-state] sensor {:?} = {:?} (confidence={:.2}, suppress_dpms={}, promote_idle={})",
                input.source, input.value, conf, suppress, promote
            );
            serde_json::json!({
                "ok": true,
                "confidence": conf,
                "suppress_dpms_wake": suppress,
                "promote_idle_faster": promote,
                "device_state": d.device_state.state(),
            })
        }
        Request::Input { trigger } => {
            let trigger = trigger.unwrap_or(InputTrigger::Unknown);
            let actions = {
                let mut d = shared.lock();
                d.device_state.note_input(trigger)
            };
            let cancelled = !actions.is_empty();
            for action in actions {
                execute(shared, action);
            }
            serde_json::json!({ "ok": true, "dim_cancelled": cancelled })
        }
        Request::Panel { on } => {
            let actions = {
                let mut d = shared.lock();
                d.device_state.set_panel(on)
            };
            for action in actions {
                execute(shared, action);
            }
            serde_json::json!({ "ok": true, "panel_on": on })
        }
        Request::GetPolicy => {
            let d = shared.lock();
            let p = &d.device_state.policy;
            serde_json::json!({
                "ok": true,
                "lock_blank_after_secs": p.lock_blank_after.map(|v| v.as_secs()).unwrap_or(0),
                "lock_blank_after_held_secs":
                    p.lock_blank_after_held.map(|v| v.as_secs()).unwrap_or(0),
                "dim_grace_secs": p.dim_grace.as_secs(),
                "evidence_ttl_secs": p.evidence_ttl.as_secs(),
                "dim_warning": p.dim_warning,
                "lock_ack_budget_secs": p.lock_ack_budget.as_secs(),
                "unlocked_blank_after_secs":
                    p.unlocked_blank_after.map(|v| v.as_secs()).unwrap_or(0),
            })
        }
        Request::SetPolicy {
            lock_blank_after_secs,
            lock_blank_after_held_secs,
            dim_grace_secs,
            evidence_ttl_secs,
            dim_warning,
            lock_ack_budget_secs,
            unlocked_blank_after_secs,
        } => {
            // 0 means "never" — the desk-clock case, and iOS's Auto-Lock has
            // the same option for the same reason.
            let as_budget = |secs: u64| {
                if secs == 0 {
                    None
                } else {
                    Some(Duration::from_secs(secs))
                }
            };
            let mut d = shared.lock();
            let p = &mut d.device_state.policy;
            if let Some(v) = lock_blank_after_secs {
                p.lock_blank_after = as_budget(v);
            }
            if let Some(v) = lock_blank_after_held_secs {
                p.lock_blank_after_held = as_budget(v);
            }
            if let Some(v) = dim_grace_secs {
                p.dim_grace = Duration::from_secs(v);
            }
            if let Some(v) = evidence_ttl_secs {
                p.evidence_ttl = Duration::from_secs(v);
            }
            if let Some(v) = dim_warning {
                p.dim_warning = v;
            }
            // A zero ack budget would blank before the lock could possibly
            // land, which is precisely the ordering this closes. Refuse it.
            if let Some(v) = lock_ack_budget_secs {
                if v == 0 {
                    drop(d);
                    return refuse(RefusalCode::InvalidArgument, "lock_ack_budget_secs must be at least 1");
                }
                p.lock_ack_budget = Duration::from_secs(v);
            }
            if let Some(v) = unlocked_blank_after_secs {
                p.unlocked_blank_after = as_budget(v);
            }
            // A dim warning longer than the budget it warns about is not a
            // warning — `dim_at` saturates to zero and the panel dims the
            // instant it goes idle, with no lit period at all. Refuse rather
            // than accept a setting whose behaviour contradicts its label.
            if let Some(budget) = p.lock_blank_after {
                if p.dim_warning && p.dim_grace >= budget {
                    drop(d);
                    return refuse(RefusalCode::InvalidArgument, "dim_grace_secs must be shorter than lock_blank_after_secs");
                }
            }
            let applied = serde_json::json!({
                "lock_blank_after_secs": p.lock_blank_after.map(|v| v.as_secs()).unwrap_or(0),
                "lock_blank_after_held_secs":
                    p.lock_blank_after_held.map(|v| v.as_secs()).unwrap_or(0),
                "dim_grace_secs": p.dim_grace.as_secs(),
                "evidence_ttl_secs": p.evidence_ttl.as_secs(),
                "dim_warning": p.dim_warning,
                "lock_ack_budget_secs": p.lock_ack_budget.as_secs(),
                "unlocked_blank_after_secs":
                    p.unlocked_blank_after.map(|v| v.as_secs()).unwrap_or(0),
            });
            info!("[device-state] policy set: {applied}");
            d.device_state.record_decision(
                "policy-set",
                applied.clone(),
                "settings changed the policy",
            );

            // Persist, and say so if it fails. A setting that took effect for
            // this run but will not survive a reboot is a success-shaped
            // switch (TASK-19's rule), so the failure goes back to the caller
            // rather than only into the log.
            if let Err(e) = d.device_state.policy.save() {
                warn!("[device-state] policy applied but NOT saved: {e}");
                d.device_state.record_error(
                    "device-state",
                    "policy-save-failed",
                    &format!("the policy is live for this run only: {e}"),
                );
                return serde_json::json!({
                    "ok": false,
                    "reason": format!("applied for this session, but not saved: {e}"),
                    "policy": applied,
                });
            }
            serde_json::json!({ "ok": true, "policy": applied })
        }
        Request::ForensicLog { count } => {
            let d = shared.lock();
            let entries = d.device_state.forensic.recent(count.unwrap_or(50));
            serde_json::json!({
                "ok": true,
                "count": entries.len(),
                "entries": entries,
            })
        }
    }
}

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

    fn idle_shared() -> Arc<Shared> {
        Arc::new(Shared {
            state: Mutex::new(Daemon {
                phase: Phase::Idle,
                controller: None,
                heartbeat_gen: 0,
                shell_alive: false,
                shell_directives: None,
                pam_service: "test".to_string(),
                device_state: DeviceStateMachine::new(),
            }),
            cond: Condvar::new(),
        })
    }

    #[test]
    fn second_shell_ready_cannot_steal_authority_lease() {
        let shared = idle_shared();
        let mut authority_heartbeat = None;
        let first = handle_request(Request::ShellReady, &shared, None, &mut authority_heartbeat);
        assert_eq!(first["ok"], true);
        assert_eq!(authority_heartbeat, Some(1));

        let mut utility_heartbeat = None;
        let second = handle_request(Request::ShellReady, &shared, None, &mut utility_heartbeat);
        assert_eq!(second["ok"], false);
        assert_eq!(second["reason"], "shell authority is already registered");
        assert_eq!(utility_heartbeat, None);

        let state = shared.lock();
        assert!(state.shell_alive);
        assert_eq!(state.heartbeat_gen, 1);
        assert_eq!(state.phase, Phase::Released);
    }
}
