//! In-agent socket server (loose-ends v2 + harness-local reminders +
//! self-service compact). Binds the harness-owned `HIVE_AGENT_SOCKET` and
//! serves the `hive-agent-sock` protocol to the in-container producers —
//! matrix / bash daemons and forge-notify are the built-in ones, but any
//! user-configured MCP server can dial the same socket and push its own
//! todos — and to `hive-agent-mcp`'s `remind`/`get_loose_ends`/
//! `cancel_loose_end`/`compact` tool impls. Todo ops hit the harness-local
//! [`Todos`] store; a new-or-changed upsert fires an in-process [`Notify`]
//! so the serve loop drives a turn. Reminder ops hit the harness-local
//! [`Reminders`] store (`None` when the store failed to open — every
//! reminder op then returns `Response::Err`); a reminder *firing* is a
//! separate path (`reminder_timer`), not driven through this socket.
//! `Request::Compact` is the odd one out — it doesn't touch any store, just
//! the harness's [`Bus`] (gate-checked context usage, then the same
//! deferred `compact_pending` flag the operator dashboard's `/compact`
//! button sets). No hive-c0re round-trip, no broker long-poll, no marker
//! files.
//!
//! One request/response line per connection, matching the producers'
//! existing best-effort JSON-line clients (they just change which socket
//! they dial, not the payload).

use std::os::unix::fs::PermissionsExt;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use anyhow::{Context, Result};
use hive_agent_sock::{Request, Response};
use hive_sh4re::inbox::{LooseEnd, saturating_age};
use hive_sock_client::{LineDeadlines, REQUEST_LINE_DEADLINE, RequestLine, read_request_line};
use tokio::io::{AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};
use tokio::sync::Notify;

use crate::events::Bus;
use crate::reminders::{Reminder, Reminders};
use crate::todos::{Todo, Todos};

/// Context-usage floor for `Request::Compact`: below this fraction of the
/// effective context window, an agent-initiated compact is refused (nothing
/// meaningful to reclaim yet, and it'd just burn a compaction pass on a
/// near-empty session). Same number a human operator would eyeball off the
/// dashboard's `ctx·Nk` badge before hitting the `/compact` button.
const COMPACT_MIN_USAGE_FRACTION: f64 = 0.66;

/// Pause after a failed `accept`, so a failure that persists (fd exhaustion)
/// does not turn the accept loop into a busy spin.
const ACCEPT_RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(1);

/// Most connections served at once. The socket is mode 0666, so any process
/// in the container can dial it; past this the accept loop stops accepting
/// (the kernel queues the rest) instead of spawning a task and holding an fd
/// per stalled client.
const MAX_CONNECTIONS: usize = 64;

/// Resolve the in-agent socket path from `HIVE_AGENT_SOCKET`. `None` when
/// unset/empty (e.g. a standalone dev run) — the server then stays off.
fn socket_path() -> Option<PathBuf> {
    std::env::var_os("HIVE_AGENT_SOCKET")
        .filter(|v| !v.is_empty())
        .map(PathBuf::from)
}

/// Dial this same in-agent socket from elsewhere IN THIS PROCESS — the
/// harness's own `web_ui` endpoints (`api_todos`, `/api/stats` reminder
/// rollup) and `Surface::post_turn_counts` all need read access to the
/// `Todos`/`Reminders` stores this module owns behind an `Arc` on a
/// different spawned task, and a loopback dial is simpler than threading
/// those `Arc`s through every caller. One-shot, best-effort (no retry,
/// unlike the broker client): a connect failure means the socket server
/// task itself isn't up, which a retry within one call wouldn't fix.
pub(crate) async fn dial(req: &Request) -> Option<Response> {
    let path = socket_path()?;
    if !path.exists() {
        return None;
    }
    tokio::time::timeout(
        std::time::Duration::from_secs(3),
        hive_sock_client::request::<_, Response>(&path, req, hive_sock_client::Retry::None),
    )
    .await
    .ok()?
    .ok()
}

/// Run the in-agent socket server: bind + accept loop, one request/response
/// line per connection. A no-op (returns `Ok`) when `HIVE_AGENT_SOCKET` is
/// unset, so a standalone harness without producers just skips it.
///
/// # Errors
///
/// Returns an error if the socket path is set but can't be bound.
pub async fn run(
    store: Arc<Todos>,
    wake: Arc<Notify>,
    reminders: Option<Arc<Reminders>>,
    bus: Bus,
) -> Result<()> {
    let Some(path) = socket_path() else {
        tracing::info!("HIVE_AGENT_SOCKET unset — in-agent todo socket disabled");
        return Ok(());
    };
    let listener = bind(&path)?;
    tracing::info!(socket = %path.display(), "in-agent todo socket listening");
    accept_loop(listener, MAX_CONNECTIONS, move |stream| {
        let (store, wake, reminders, bus) =
            (store.clone(), wake.clone(), reminders.clone(), bus.clone());
        async move {
            if let Err(e) = handle_conn(stream, &store, &wake, reminders.as_deref(), &bus).await {
                tracing::warn!(error = ?e, "in-agent todo connection failed");
            }
        }
    })
    .await
}

/// Accept connections for ever, serving at most `max` at a time with
/// `handle`. A permit is taken before `accept`, so at the limit the loop
/// waits there and the kernel's listen queue holds the rest.
async fn accept_loop<F, Fut>(listener: UnixListener, max: usize, handle: F) -> Result<()>
where
    F: Fn(UnixStream) -> Fut,
    Fut: std::future::Future<Output = ()> + Send + 'static,
{
    let permits = Arc::new(tokio::sync::Semaphore::new(max));
    loop {
        let permit = permits
            .clone()
            .acquire_owned()
            .await
            .context("connection limit closed")?;
        match listener.accept().await {
            Ok((stream, _)) => {
                let served = handle(stream);
                tokio::spawn(async move {
                    served.await;
                    drop(permit);
                });
            }
            Err(e) => {
                tracing::warn!(error = ?e, "in-agent todo accept failed");
                // A persistent failure (fd exhaustion) fails again at once;
                // without a pause this loop spins and floods the log.
                tokio::time::sleep(ACCEPT_RETRY_DELAY).await;
            }
        }
    }
}

/// Bind a `UnixListener` at `path`, creating the parent dir and unlinking a
/// stale socket left by a prior boot (which would otherwise block `bind`
/// with `EADDRINUSE`).
///
/// The mode is set **here, at creation** — not corrected afterwards by
/// whoever notices. `connect(2)` on a unix socket requires *write*
/// permission, and `bind` leaves `0777 & ~umask` (typically `0755`), which
/// silently excludes every uid but the harness's own. `hive-c0re` pushes
/// todos in over this socket from the host, so that default locks it out.
/// Same two lines the sibling `web.sock` has carried all along
/// (`web_ui::serve`); access control is the containing directory's job, not
/// the socket's.
fn bind(path: &Path) -> Result<UnixListener> {
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent)
            .with_context(|| format!("create in-agent socket dir {}", parent.display()))?;
    }
    if path.exists() {
        let _ = std::fs::remove_file(path);
    }
    let listener = UnixListener::bind(path)
        .with_context(|| format!("bind in-agent socket {}", path.display()))?;
    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o666))
        .with_context(|| format!("set perms on in-agent socket {}", path.display()))?;
    Ok(listener)
}

/// Handle one connection: read a single JSON request line, apply it to the
/// store, write the JSON response line back.
async fn handle_conn(
    stream: UnixStream,
    store: &Todos,
    wake: &Notify,
    reminders: Option<&Reminders>,
    bus: &Bus,
) -> Result<()> {
    serve_one(stream, REQUEST_DEADLINES, |req| {
        dispatch(req, store, wake, reminders, bus)
    })
    .await
}

/// One request line, one response line. A single-shot connection, so silence
/// is bounded too: the socket is mode 0666 and any local process can open it.
const REQUEST_DEADLINES: LineDeadlines = LineDeadlines {
    idle: Some(REQUEST_LINE_DEADLINE),
    line: Some(REQUEST_LINE_DEADLINE),
};

/// Read the request line under `deadlines`, answer it with `apply`, and write
/// the response. An over-long or stalled line gets a `Response::Err`.
async fn serve_one(
    stream: UnixStream,
    deadlines: LineDeadlines,
    apply: impl FnOnce(Request) -> Response,
) -> Result<()> {
    let (read, mut write) = stream.into_split();
    let mut reader = BufReader::new(read);
    let mut line = Vec::new();
    let outcome = read_request_line(&mut reader, &mut line, deadlines).await?;
    if outcome == RequestLine::Eof {
        return Ok(());
    }
    let resp = if let Some(message) = outcome.refusal() {
        Response::Err { message }
    } else {
        match serde_json::from_slice::<Request>(line.trim_ascii()) {
            Ok(req) => apply(req),
            Err(e) => Response::Err {
                message: format!("bad request: {e}"),
            },
        }
    };
    let mut out = serde_json::to_string(&resp)?;
    out.push('\n');
    write.write_all(out.as_bytes()).await?;
    write.flush().await?;
    Ok(())
}

/// Apply one request to the store, firing `wake` on a new/changed upsert so
/// the serve loop runs a turn. Each arm calls a small named handler function
/// directly — no sub-match/`unreachable!()` indirection per family (that
/// pattern got reviewed out of the todo family in the diagnostic-logging
/// follow-up PR; kept the reminder family consistent with it here rather
/// than reintroducing it). `reminders` is `None` when that store failed to
/// open at boot, in which case every reminder op returns an `Err` — each
/// handler checks for its own `None` case.
fn dispatch(
    req: Request,
    store: &Todos,
    wake: &Notify,
    reminders: Option<&Reminders>,
    bus: &Bus,
) -> Response {
    match req {
        Request::UpsertTodo {
            subsystem,
            key,
            summary,
            source,
            reopen_if_acked,
        } => upsert_todo(
            store,
            wake,
            &subsystem,
            key.as_deref(),
            &summary,
            source.as_deref(),
            reopen_if_acked,
        ),
        Request::ClearTodo {
            subsystem,
            key,
            all,
        } => clear_todo(store, &subsystem, key.as_deref(), all),
        Request::ListTodos { subsystem } => list_todos(store, subsystem.as_deref()),
        Request::MarkTodoDone { id } => mark_todo_done(store, id),
        Request::MarkTodosDone { ids } => mark_todos_done(store, &ids),
        Request::StoreReminder {
            message,
            timing,
            file_path,
        } => store_reminder(reminders, &message, &timing, file_path.as_deref()),
        Request::ListReminders => list_reminders(reminders),
        Request::CancelReminder { id } => cancel_reminder(reminders, id),
        Request::CountPendingReminders => count_pending_reminders(reminders),
        Request::ReminderRollup { since_secs } => reminder_rollup(reminders, since_secs),
        Request::Compact { wake_prompt } => compact(bus, wake_prompt),
    }
}

/// `StoreReminder` handler: persists a reminder due at `timing`.
fn store_reminder(
    reminders: Option<&Reminders>,
    message: &str,
    timing: &hive_sh4re::inbox::ReminderTiming,
    file_path: Option<&str>,
) -> Response {
    let Some(r) = reminders else {
        return no_reminders_store();
    };
    match crate::reminder_timer::store(r, message, timing, file_path) {
        Ok(_id) => Response::Ok,
        Err(message) => Response::Err { message },
    }
}

/// `ListReminders` handler: this agent's own pending reminders.
fn list_reminders(reminders: Option<&Reminders>) -> Response {
    let Some(r) = reminders else {
        return no_reminders_store();
    };
    match r.list_pending() {
        Ok(rows) => Response::LooseEnds {
            loose_ends: rows.into_iter().map(reminder_to_loose_end).collect(),
        },
        Err(e) => err(&e),
    }
}

/// `CancelReminder` handler: drops one pending reminder by id.
fn cancel_reminder(reminders: Option<&Reminders>, id: i64) -> Response {
    let Some(r) = reminders else {
        return no_reminders_store();
    };
    match r.cancel(id) {
        Ok(count) => Response::Acked {
            count: u64::try_from(count).unwrap_or(0),
        },
        Err(e) => err(&e),
    }
}

/// `CountPendingReminders` handler.
fn count_pending_reminders(reminders: Option<&Reminders>) -> Response {
    let Some(r) = reminders else {
        return no_reminders_store();
    };
    match r.count_pending() {
        Ok(count) => Response::PendingRemindersCount { count },
        Err(e) => err(&e),
    }
}

/// `ReminderRollup` handler: scheduled/delivered/pending counts over a
/// trailing `since_secs` window (`0` = all time).
fn reminder_rollup(reminders: Option<&Reminders>, since_secs: u64) -> Response {
    let Some(r) = reminders else {
        return no_reminders_store();
    };
    let since_secs = i64::try_from(since_secs).unwrap_or(i64::MAX);
    match r.rollup(since_secs) {
        Ok(stats) => Response::ReminderRollup { stats },
        Err(e) => err(&e),
    }
}

/// `UpsertTodo` handler: writes/refreshes a todo row, logs the outcome, and
/// fires `wake` on a new-or-changed upsert so the serve loop runs a turn.
fn upsert_todo(
    store: &Todos,
    wake: &Notify,
    subsystem: &str,
    key: Option<&str>,
    summary: &str,
    source: Option<&str>,
    reopen_if_acked: bool,
) -> Response {
    match store.upsert(subsystem, key, summary, source, reopen_if_acked) {
        Ok((id, changed)) => {
            tracing::debug!(%subsystem, ?key, id, changed, "todo upsert");
            if changed {
                wake.notify_one();
            }
            Response::Ok
        }
        Err(e) => err(&e),
    }
}

/// `ClearTodo` handler: drops one keyed todo, or every todo in `subsystem`
/// when `all` is set.
fn clear_todo(store: &Todos, subsystem: &str, key: Option<&str>, all: bool) -> Response {
    let result = if all {
        store.clear_subsystem(subsystem)
    } else {
        store.clear(subsystem, key)
    };
    match result {
        Ok(count) => {
            tracing::debug!(%subsystem, ?key, all, count, "todo clear");
            Response::Acked {
                count: u64::try_from(count).unwrap_or(0),
            }
        }
        Err(e) => err(&e),
    }
}

/// `ListTodos` handler: read-only, so no debug logging — not relevant to
/// diagnosing wake behaviour.
fn list_todos(store: &Todos, subsystem: Option<&str>) -> Response {
    match store.list(subsystem) {
        Ok(todos) => Response::LooseEnds {
            loose_ends: todos.into_iter().map(to_loose_end).collect(),
        },
        Err(e) => err(&e),
    }
}

/// `MarkTodoDone` handler: marks a single todo done by id.
fn mark_todo_done(store: &Todos, id: i64) -> Response {
    match store.mark_done(id) {
        Ok(count) => {
            tracing::debug!(id, count, "todo mark-done");
            Response::Acked {
                count: u64::try_from(count).unwrap_or(0),
            }
        }
        Err(e) => err(&e),
    }
}

/// `MarkTodosDone` handler: bulk-acks an explicit list of todo ids — the
/// escape hatch for a backlog too large to triage one-by-one (see
/// `Todos::mark_done_many`'s doc for why it's ids, not a threshold).
fn mark_todos_done(store: &Todos, ids: &[i64]) -> Response {
    match store.mark_done_many(ids) {
        Ok(count) => {
            tracing::debug!(n_ids = ids.len(), count, "todo bulk mark-done");
            Response::Acked {
                count: u64::try_from(count).unwrap_or(0),
            }
        }
        Err(e) => err(&e),
    }
}

/// `Request::Compact` handler: gate on context usage, then queue the same
/// deferred `compact_pending` flag the operator's `/compact` button sets.
/// Mirrors `hive-agent::web_ui::actions::post_compact` but reachable from
/// the agent's own MCP tool instead of the dashboard, and refuses below
/// [`COMPACT_MIN_USAGE_FRACTION`] instead of always honouring the request —
/// an agent can call this speculatively, a human clicking the dashboard
/// button already made the judgment call. `wake_prompt`, when set, is
/// forwarded to `Bus::request_compact` so the turn loop drives a synthetic
/// follow-up turn once the compaction actually finishes.
fn compact(bus: &Bus, wake_prompt: Option<String>) -> Response {
    let Some(usage) = bus.last_ctx_usage() else {
        return Response::Err {
            message: "compact refused: no completed turn yet — nothing to compact".to_owned(),
        };
    };
    let model = bus.model();
    let window = bus.effective_context_window(&model);
    if window == 0 {
        return Response::Err {
            message: "compact refused: effective context window is unknown (0)".to_owned(),
        };
    }
    // Token counts stay well under 2^53 in practice, so the f64 conversion
    // is exact; this is a threshold ratio, not a byte-count display, but the
    // same "cosmetic precision loss" reasoning as `hivectl::quota::human_bytes` applies.
    #[allow(clippy::cast_precision_loss)]
    let fraction = usage.context_tokens() as f64 / window as f64;
    if fraction < COMPACT_MIN_USAGE_FRACTION {
        return Response::Err {
            message: format!(
                "compact refused: context usage is {:.0}% of the {window}-token window, \
                 below the {:.0}% floor — not worth compacting yet",
                fraction * 100.0,
                COMPACT_MIN_USAGE_FRACTION * 100.0
            ),
        };
    }
    let will_wake = wake_prompt.is_some();
    bus.request_compact(wake_prompt);
    bus.emit(crate::events::LiveEvent::Note {
        text: if will_wake {
            "agent: self-requested /compact (with wake prompt) — running at the end of the \
             current turn"
                .into()
        } else {
            "agent: self-requested /compact — running at the end of the current turn".into()
        },
    });
    Response::Ok
}

/// Shared "reminders db unavailable" response for every reminder op when
/// the store failed to open at boot (see `main.rs`'s best-effort open).
fn no_reminders_store() -> Response {
    Response::Err {
        message: "reminders store unavailable on this agent".to_owned(),
    }
}

/// Map a stored [`Reminder`] to a [`LooseEnd::Reminder`], deriving
/// `age_seconds` from `created_at` (mirrors the old c0re rendering —
/// "age" is how long the reminder has been *scheduled*, not how soon
/// it's due).
fn reminder_to_loose_end(r: Reminder) -> LooseEnd {
    let now = chrono::Utc::now().timestamp();
    LooseEnd::Reminder {
        id: r.id,
        owner: crate::identity::label(),
        message: r.message,
        due_at: r.due_at,
        age_seconds: saturating_age(now, r.created_at.timestamp()),
    }
}

/// Format a store error into an `Err` response.
fn err(e: &anyhow::Error) -> Response {
    Response::Err {
        message: format!("{e:#}"),
    }
}

/// Map a stored [`Todo`] to a [`LooseEnd::Todo`], deriving `age_seconds`
/// from `updated_at` via [`saturating_age`].
fn to_loose_end(t: Todo) -> LooseEnd {
    let now = chrono::Utc::now().timestamp();
    LooseEnd::Todo {
        id: t.id,
        subsystem: t.subsystem,
        subsystem_key: t.subsystem_key,
        summary: t.summary,
        source: t.source,
        age_seconds: saturating_age(now, t.updated_at.timestamp()),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::{DateTime, Duration, Utc};
    use hive_sock_client::REQUEST_LINE_MAX_BYTES;
    use tokio::io::AsyncBufReadExt;

    /// At the limit the loop stops accepting; a freed slot lets the next
    /// connection in.
    #[tokio::test]
    async fn the_accept_loop_serves_no_more_than_its_limit_at_once() {
        use std::sync::atomic::{AtomicUsize, Ordering};

        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("s.sock");
        let listener = UnixListener::bind(&path).expect("bind");
        let started = Arc::new(AtomicUsize::new(0));
        let release = Arc::new(Notify::new());
        let (s, r) = (started.clone(), release.clone());
        let server = tokio::spawn(accept_loop(listener, 1, move |_stream| {
            let (s, r) = (s.clone(), r.clone());
            async move {
                s.fetch_add(1, Ordering::SeqCst);
                r.notified().await;
            }
        }));

        let _a = UnixStream::connect(&path).await.expect("first");
        let _b = UnixStream::connect(&path).await.expect("second");
        let reached = |n: usize| {
            let started = started.clone();
            async move {
                tokio::time::timeout(std::time::Duration::from_secs(5), async {
                    while started.load(Ordering::SeqCst) < n {
                        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
                    }
                })
                .await
                .is_ok()
            }
        };
        assert!(reached(1).await, "first was served");
        // The negative case only fails if the cap is broken, so a short wait
        // is enough.
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        assert_eq!(started.load(Ordering::SeqCst), 1, "second waited its turn");

        release.notify_one();
        assert!(reached(2).await, "freed slot admitted the second");
        server.abort();
    }

    /// `saturating_age`'s own arms are exact (`hive-sh4re::inbox`); what
    /// is only checkable here is *which* stored timestamp each projection
    /// feeds it, and both read the clock internally — so these assert a
    /// range wide enough that a slow runner can't fail them.
    const SLACK: u64 = 600;

    /// Not asserted anywhere below: `LooseEnd::Reminder.owner`. It comes
    /// from `identity::label()`, which reads a process-global env var
    /// that `identity`'s own tests set and restore — in the same test
    /// binary, in parallel. Any value assertion here would be a race, not
    /// a check.
    fn reminder(created_at: DateTime<Utc>, due_at: DateTime<Utc>) -> Reminder {
        Reminder {
            id: 7,
            message: "check the gate".to_owned(),
            file_path: None,
            due_at,
            created_at,
        }
    }

    fn todo(updated_at: DateTime<Utc>) -> Todo {
        Todo {
            id: 11,
            subsystem: "matrix".to_owned(),
            subsystem_key: Some("!room:x".to_owned()),
            summary: "3 unread".to_owned(),
            source: Some("#ops".to_owned()),
            updated_at,
        }
    }

    #[test]
    fn a_reminder_ages_from_created_at_not_from_when_it_is_due() {
        // Scheduled 600s ago, due an hour out. Reading the age off
        // `due_at` would clamp to 0, so the two fields are separable.
        let r = reminder(
            Utc::now() - Duration::seconds(600),
            Utc::now() + Duration::seconds(3600),
        );
        let due = r.due_at;
        match reminder_to_loose_end(r) {
            LooseEnd::Reminder {
                id,
                message,
                due_at,
                age_seconds,
                ..
            } => {
                assert_eq!(id, 7);
                assert_eq!(message, "check the gate");
                assert_eq!(due_at, due, "due_at passes through untouched");
                assert!(
                    (600..600 + SLACK).contains(&age_seconds),
                    "age {age_seconds} should track created_at, not due_at"
                );
            }
            other => panic!("expected a Reminder loose end, got {other:?}"),
        }
    }

    #[test]
    fn a_todo_ages_from_updated_at_and_carries_its_dedup_key() {
        let t = todo(Utc::now() - Duration::seconds(600));
        match to_loose_end(t) {
            LooseEnd::Todo {
                id,
                subsystem,
                subsystem_key,
                summary,
                source,
                age_seconds,
            } => {
                assert_eq!(id, 11);
                assert_eq!(subsystem, "matrix");
                // Distinct values: the two `Option<String>` fields are
                // adjacent and same-typed, so a swap has to be visible.
                assert_eq!(subsystem_key.as_deref(), Some("!room:x"));
                assert_eq!(source.as_deref(), Some("#ops"));
                assert_eq!(summary, "3 unread");
                assert!((600..600 + SLACK).contains(&age_seconds));
            }
            other => panic!("expected a Todo loose end, got {other:?}"),
        }
    }

    #[test]
    fn a_timestamp_in_the_future_reads_as_zero_through_both_projections() {
        // Not a hypothetical: these stores are written by in-container
        // producers, and `updated_at` crossing `now` is exactly what a
        // clock sync during a turn produces.
        let ahead = Utc::now() + Duration::seconds(3600);
        let behind = Utc::now() - Duration::seconds(600);

        let (future_r, past_r) = (age_of_reminder(ahead), age_of_reminder(behind));
        let (future_t, past_t) = (age_of_todo(ahead), age_of_todo(behind));

        assert_eq!(future_r, 0);
        assert_eq!(future_t, 0);
        // Controls: same call, past timestamp — so the zeros above are
        // the clamp firing, not the projections always emitting 0.
        assert!(past_r >= 600, "reminder control read {past_r}");
        assert!(past_t >= 600, "todo control read {past_t}");
    }

    fn age_of_reminder(created_at: DateTime<Utc>) -> u64 {
        match reminder_to_loose_end(reminder(created_at, Utc::now())) {
            LooseEnd::Reminder { age_seconds, .. } => age_seconds,
            other => panic!("expected a Reminder loose end, got {other:?}"),
        }
    }

    fn age_of_todo(updated_at: DateTime<Utc>) -> u64 {
        match to_loose_end(todo(updated_at)) {
            LooseEnd::Todo { age_seconds, .. } => age_seconds,
            other => panic!("expected a Todo loose end, got {other:?}"),
        }
    }

    /// Send `input` (then keep the write half open), return the one response line.
    async fn exchange(input: Vec<u8>, deadlines: LineDeadlines) -> (Response, Result<()>) {
        let (client, server) = UnixStream::pair().expect("socketpair");
        let handler = tokio::spawn(serve_one(server, deadlines, |_| Response::Ok));
        let (read, mut write) = client.into_split();
        let _writer = tokio::spawn(async move {
            let _ = write.write_all(&input).await;
            std::future::pending::<()>().await;
        });
        let mut reply = String::new();
        BufReader::new(read)
            .read_line(&mut reply)
            .await
            .expect("reply");
        let resp = serde_json::from_str(&reply).expect("response json");
        (resp, handler.await.expect("join"))
    }

    /// What keeps a test's stores and `Bus` alive: the scratch dir, and the
    /// harness dir a `Bus` opens its events db in (unset in the build sandbox).
    type Scratch = (tempfile::TempDir, crate::paths::HarnessDirGuard);

    /// Todo and reminder stores over a scratch dir, which must outlive them.
    fn stores() -> (Scratch, Todos, Reminders) {
        let harness = crate::paths::harness_dir_guard();
        let dir = tempfile::tempdir().expect("tempdir");
        let todos = Todos::open(&dir.path().join("todos.sqlite")).expect("todos");
        let reminders = Reminders::open(&dir.path().join("reminders.sqlite")).expect("reminders");
        ((dir, harness), todos, reminders)
    }

    fn upsert(subsystem: &str, key: Option<&str>, summary: &str) -> Request {
        Request::UpsertTodo {
            subsystem: subsystem.to_owned(),
            key: key.map(str::to_owned),
            summary: summary.to_owned(),
            source: None,
            reopen_if_acked: false,
        }
    }

    fn list(todos: &Todos, subsystem: Option<&str>) -> Vec<LooseEnd> {
        let req = Request::ListTodos {
            subsystem: subsystem.map(str::to_owned),
        };
        match dispatch(req, todos, &Notify::new(), None, &Bus::new()) {
            Response::LooseEnds { loose_ends } => loose_ends,
            other => panic!("expected loose ends, got {other:?}"),
        }
    }

    /// Whether a wake is pending, consuming it. `notify_one` leaves a permit
    /// behind, so one poll is enough: nothing here needs a clock.
    fn woke(wake: &Notify) -> bool {
        use futures_util::FutureExt as _;
        wake.notified().now_or_never().is_some()
    }

    fn acked(resp: &Response) -> u64 {
        match resp {
            Response::Acked { count } => *count,
            other => panic!("expected Acked, got {other:?}"),
        }
    }

    #[test]
    fn an_upsert_wakes_the_loop_only_when_it_is_new_or_changed() {
        let (_dir, todos, _) = stores();
        let (wake, bus) = (Notify::new(), Bus::new());
        let up = |summary| {
            dispatch(
                upsert("matrix", Some("!r"), summary),
                &todos,
                &wake,
                None,
                &bus,
            )
        };

        assert!(matches!(up("one unread"), Response::Ok));
        assert!(woke(&wake), "a new row wakes the loop");
        assert!(matches!(up("one unread"), Response::Ok));
        assert!(!woke(&wake), "an identical keyed re-push stays quiet");
        assert!(matches!(up("two unread"), Response::Ok));
        assert!(woke(&wake), "a changed summary wakes it again");
        assert_eq!(list(&todos, None).len(), 1, "one keyed row, not three");
    }

    #[test]
    fn listing_filters_by_subsystem_and_carries_the_row() {
        let (_dir, todos, _) = stores();
        let (wake, bus) = (Notify::new(), Bus::new());
        dispatch(upsert("matrix", Some("!r"), "m"), &todos, &wake, None, &bus);
        dispatch(upsert("bash", Some("t1"), "b"), &todos, &wake, None, &bus);

        assert_eq!(list(&todos, None).len(), 2);
        let bash = list(&todos, Some("bash"));
        let [
            LooseEnd::Todo {
                subsystem,
                subsystem_key,
                summary,
                ..
            },
        ] = bash.as_slice()
        else {
            panic!("one bash todo");
        };
        assert_eq!(
            (
                subsystem.as_str(),
                subsystem_key.as_deref(),
                summary.as_str()
            ),
            ("bash", Some("t1"), "b")
        );
    }

    #[test]
    fn clearing_drops_one_keyed_row_or_the_whole_subsystem() {
        let (_dir, todos, _) = stores();
        let (wake, bus) = (Notify::new(), Bus::new());
        for (sub, key) in [("matrix", "!a"), ("matrix", "!b"), ("bash", "t1")] {
            dispatch(upsert(sub, Some(key), "x"), &todos, &wake, None, &bus);
        }
        let clear = |key: Option<&str>, all| Request::ClearTodo {
            subsystem: "matrix".to_owned(),
            key: key.map(str::to_owned),
            all,
        };

        let one = dispatch(clear(Some("!a"), false), &todos, &wake, None, &bus);
        assert_eq!(acked(&one), 1);
        assert_eq!(list(&todos, Some("matrix")).len(), 1);
        let all = dispatch(clear(None, true), &todos, &wake, None, &bus);
        assert_eq!(acked(&all), 1);
        assert!(list(&todos, Some("matrix")).is_empty());
        assert_eq!(list(&todos, Some("bash")).len(), 1, "other subsystems stay");
    }

    #[test]
    fn marking_done_acks_the_given_ids_and_hides_them() {
        let (_dir, todos, _) = stores();
        let (wake, bus) = (Notify::new(), Bus::new());
        for key in ["a", "b", "c"] {
            dispatch(upsert("s", Some(key), key), &todos, &wake, None, &bus);
        }
        let ids: Vec<i64> = list(&todos, None)
            .into_iter()
            .map(|t| match t {
                LooseEnd::Todo { id, .. } => id,
                other => panic!("expected a Todo, got {other:?}"),
            })
            .collect();

        let one = dispatch(
            Request::MarkTodoDone { id: ids[0] },
            &todos,
            &wake,
            None,
            &bus,
        );
        assert_eq!(acked(&one), 1);
        let many = dispatch(
            Request::MarkTodosDone {
                ids: ids[1..].to_vec(),
            },
            &todos,
            &wake,
            None,
            &bus,
        );
        assert_eq!(acked(&many), 2);
        assert!(list(&todos, None).is_empty());
        let again = dispatch(
            Request::MarkTodoDone { id: ids[0] },
            &todos,
            &wake,
            None,
            &bus,
        );
        assert_eq!(acked(&again), 0, "an already-done id acks nothing");
    }

    #[test]
    fn every_reminder_op_errors_when_the_store_did_not_open() {
        let (_dir, todos, _) = stores();
        let (wake, bus) = (Notify::new(), Bus::new());
        let reqs = [
            Request::StoreReminder {
                message: "m".to_owned(),
                timing: hive_sh4re::inbox::ReminderTiming::InSeconds { seconds: 60 },
                file_path: None,
            },
            Request::ListReminders,
            Request::CancelReminder { id: 1 },
            Request::CountPendingReminders,
            Request::ReminderRollup { since_secs: 0 },
        ];
        for req in reqs {
            let resp = dispatch(req, &todos, &wake, None, &bus);
            assert!(
                matches!(&resp, Response::Err { message } if message.contains("unavailable")),
                "{resp:?}"
            );
        }
    }

    #[test]
    fn a_reminder_is_stored_listed_counted_rolled_up_and_cancelled() {
        let (_dir, todos, reminders) = stores();
        let (wake, bus) = (Notify::new(), Bus::new());
        let run = |req| dispatch(req, &todos, &wake, Some(&reminders), &bus);

        let stored = run(Request::StoreReminder {
            message: "check the deploy".to_owned(),
            timing: hive_sh4re::inbox::ReminderTiming::InSeconds { seconds: 3600 },
            file_path: None,
        });
        assert!(matches!(stored, Response::Ok), "{stored:?}");
        assert!(matches!(
            run(Request::CountPendingReminders),
            Response::PendingRemindersCount { count: 1 }
        ));
        let Response::LooseEnds { loose_ends } = run(Request::ListReminders) else {
            panic!("expected loose ends");
        };
        let [LooseEnd::Reminder { id, message, .. }] = loose_ends.as_slice() else {
            panic!("one reminder, got {loose_ends:?}");
        };
        assert_eq!(message, "check the deploy");
        assert!(matches!(
            run(Request::ReminderRollup { since_secs: 0 }),
            Response::ReminderRollup { .. }
        ));

        assert_eq!(acked(&run(Request::CancelReminder { id: *id })), 1);
        assert_eq!(acked(&run(Request::CancelReminder { id: *id })), 0);
        assert!(matches!(
            run(Request::CountPendingReminders),
            Response::PendingRemindersCount { count: 0 }
        ));
    }

    /// A bus whose context window is 100 tokens and whose last turn used
    /// `input` of them.
    fn bus_with_context(input: u64) -> Bus {
        let bus = Bus::new();
        bus.set_api_context_window(100);
        let usage: hive_claude::TokenUsage = serde_json::from_value(serde_json::json!({
            "input_tokens": input,
            "output_tokens": 0,
            "cache_read_input_tokens": 0,
            "cache_creation_input_tokens": 0,
        }))
        .expect("a usage block");
        bus.record_turn_usage(usage, usage);
        bus
    }

    fn compact_req(prompt: Option<&str>) -> Request {
        Request::Compact {
            wake_prompt: prompt.map(str::to_owned),
        }
    }

    #[test]
    fn compact_is_refused_before_any_turn_and_below_the_usage_floor() {
        let (_dir, todos, _) = stores();
        let wake = Notify::new();

        let fresh = Bus::new();
        let resp = dispatch(compact_req(None), &todos, &wake, None, &fresh);
        assert!(
            matches!(&resp, Response::Err { message } if message.contains("no completed turn")),
            "{resp:?}"
        );
        assert!(fresh.take_compact().is_none());

        let low = bus_with_context(10);
        let resp = dispatch(compact_req(None), &todos, &wake, None, &low);
        assert!(
            matches!(&resp, Response::Err { message } if message.contains("below")),
            "{resp:?}"
        );
        assert!(low.take_compact().is_none(), "a refusal queues nothing");
    }

    #[test]
    fn compact_above_the_floor_queues_the_request_with_its_wake_prompt() {
        let (_dir, todos, _) = stores();
        let bus = bus_with_context(90);
        let resp = dispatch(
            compact_req(Some("carry on")),
            &todos,
            &Notify::new(),
            None,
            &bus,
        );
        assert!(matches!(resp, Response::Ok), "{resp:?}");
        let queued = bus.take_compact().expect("a queued compact");
        assert_eq!(queued.wake_prompt.as_deref(), Some("carry on"));
        assert!(bus.take_compact().is_none(), "taken once");
    }

    #[tokio::test]
    async fn a_valid_request_is_applied() {
        let (resp, done) = exchange(b"{\"cmd\":\"compact\"}\n".to_vec(), REQUEST_DEADLINES).await;
        assert!(matches!(resp, Response::Ok), "{resp:?}");
        done.expect("served");
    }

    #[tokio::test]
    async fn an_over_long_line_gets_an_error_not_an_unbounded_buffer() {
        let (resp, done) =
            exchange(vec![b'x'; REQUEST_LINE_MAX_BYTES + 1], REQUEST_DEADLINES).await;
        assert!(
            matches!(&resp, Response::Err { message } if message.starts_with("request too long")),
            "{resp:?}"
        );
        done.expect("served");
    }

    #[tokio::test]
    async fn a_sender_that_stalls_mid_line_times_out() {
        let short = std::time::Duration::from_millis(50);
        let deadlines = LineDeadlines {
            idle: Some(short),
            line: Some(short),
        };
        let (resp, done) = exchange(b"{\"cmd\"".to_vec(), deadlines).await;
        assert!(
            matches!(&resp, Response::Err { message } if message.starts_with("request timed out")),
            "{resp:?}"
        );
        done.expect("served");
    }

    #[tokio::test]
    async fn a_sender_that_never_writes_times_out() {
        let short = std::time::Duration::from_millis(50);
        let deadlines = LineDeadlines {
            idle: Some(short),
            line: Some(short),
        };
        let (resp, done) = exchange(Vec::new(), deadlines).await;
        assert!(
            matches!(&resp, Response::Err { message } if message.starts_with("request timed out")),
            "{resp:?}"
        );
        done.expect("served");
    }
}