//! Harness serve-loop binary. Long-polls the broker inbox and drives one
//! claude turn per message. There is one role: agent. The `Surface`
//! trait + `AgentSurface` zero-sized type tag keeps the turn loop
//! generic and testable. Sibling: `hive-agent-mcp` (the MCP server this
//! loop points claude at).
//! Architecture lives in
//! [`docs/turn-loop/::Harness binary shape`](../../../docs/turn-loop/README.md).
//!
//! Single bin crate: the module tree below (formerly this crate's `lib.rs`,
//! before lib + bin were collapsed into one) plus the serve loop.

mod db_migrate;
mod disk_watch;
mod events;
mod harness_state;
mod identity;
mod login;
mod login_session;
mod mcp_config;
mod paths;
mod plugins;
mod prompt;
mod questions;
mod reminder_timer;
mod reminders;
mod serve_common;
mod state_entry_watch;
mod stats;
mod stream_enrich;
mod todo_server;
mod todos;
mod turn;
mod turn_stats;
mod vacuum;
mod web_ui;

/// Default socket path inside the container — bind-mounted by `hive-c0re`.
const DEFAULT_SOCKET: &str = "/run/hive/mcp.sock";

/// Retry policy for every request to the host-served control socket.
/// Nothing on this side of the socket has a natural retry — the serve loop
/// and the web UI both hand a failure straight to a human or to claude —
/// so a hive-c0re restart is worth waiting out rather than surfacing.
const CONTROL_SOCKET_RETRY: Retry = Retry::RideOutRestart;

/// Default web UI port — used when `HIVE_PORT` env is unset.
const DEFAULT_WEB_PORT: u16 = 8042;

/// How often the serve loop re-stats the pause marker while parked.
/// Only paid while an agent is actually paused, and only against the
/// local harness dir, so a tight-ish interval is cheap and keeps
/// `hivectl resume` feeling immediate.
const PAUSE_POLL: Duration = Duration::from_secs(5);

/// Consecutive todo-wake turns that skip `get_loose_ends` before the serve
/// loop pauses itself (writes the same `paused_marker()` file `hivectl
/// agent <name> pause`/`resume` already toggle — no new plumbing). A
/// deliberately small number: a genuine miss should be rare, and pausing
/// quickly beats letting an unacked todo backlog balloon to a size that
/// makes `get_loose_ends` itself expensive/unwieldy to read (the failure
/// mode that motivated this in the first place — a reviewer's call on the
/// exact threshold, adjustable if it proves too twitchy in practice).
const TODO_MISS_PAUSE_THRESHOLD: u32 = 3;

use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use crate::events::{Bus, LiveEvent, TurnState};
use crate::login::LoginState;
use crate::turn_stats::TurnStats;
use anyhow::Result;
use clap::Parser;
use hive_core_agent_sock::{Request, Response};
use hive_sh4re::manager::{HelperEvent, SYSTEM_SENDER};
use hive_sock_client::Retry;

#[derive(Parser)]
#[command(name = "hive-agent", about = "hyperhive harness serve loop")]
struct Cli {
    /// Path to the per-agent MCP socket (bind-mounted from the host).
    #[arg(long, default_value = DEFAULT_SOCKET)]
    socket: PathBuf,

    /// Inbox poll interval in milliseconds.
    #[arg(long, default_value_t = 1000)]
    poll_ms: u64,
}

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
        )
        .init();

    let cli = Cli::parse();
    serve_main::<AgentSurface>(&cli.socket, cli.poll_ms).await
}

// ---------- shared turn helpers ----------

/// Surface a `SYSTEM_SENDER` message in the live event bus + tracing
/// log. Both agents and the manager receive `QuestionAnswered`,
/// `ContainerCrash`, reparent notifications, and friends; the parse
/// and log path is identical. Quiet no-op when `from` isn't
/// `SYSTEM_SENDER`.
///
/// Also keeps the harness-local questions mirror in sync on the two
/// question-shaped events: `QuestionAsked` (a peer/manager asked *this*
/// agent — mirror an `Answering` row) and
/// `QuestionAnswered` (a question *this* agent asked got a reply —
/// clear the mirrored `Asked` row). Best-effort loopback dial of the
/// in-agent socket (see `todo_server::dial`'s docs on why a dial beats
/// threading an `Arc<Questions>` through this whole call chain) — a
/// dial failure is a missed mirror update, not a turn failure, so it's
/// logged and swallowed rather than propagated.
async fn log_system_event(bus: &Bus, from: &str, body: &str) {
    if from != SYSTEM_SENDER {
        return;
    }
    let parsed = serde_json::from_str::<HelperEvent>(body).ok();
    if let Some(event) = &parsed {
        tracing::info!(?event, "helper event");
    } else {
        tracing::info!(%from, %body, "system message");
    }
    match parsed {
        Some(HelperEvent::QuestionAsked {
            id,
            asker,
            question,
            ..
        }) => {
            mirror_question(hive_agent_sock::Request::RecordAnsweringQuestion {
                id,
                asker,
                question,
            })
            .await;
        }
        Some(HelperEvent::QuestionAnswered { id, .. }) => {
            mirror_question(hive_agent_sock::Request::ClearQuestion { id }).await;
        }
        _ => {}
    }
    bus.emit(LiveEvent::Note {
        text: format!("[system] {body}"),
    });
}

/// Open the harness-local questions mirror against the consolidated state
/// db — same open-alongside shape as the reminders store
/// (`reminders::Reminders::open`) opened just above this call site in
/// `serve_main`, sharing the same file (distinct table, see
/// `questions::SCHEMA`). `None` on open failure disables question ops the
/// same way a failed reminders open disables reminder ops
/// (`no_questions_store` in `todo_server`). Pulled out to its own fn to
/// keep `serve_main` under the pedantic line-count lint.
fn open_question_store() -> Option<Arc<questions::Questions>> {
    match questions::Questions::open(&paths::state_db()) {
        Ok(store) => Some(Arc::new(store)),
        Err(e) => {
            tracing::error!(error = ?e, "open questions db failed — question mirror disabled");
            None
        }
    }
}

/// Fire-and-forget helper for `log_system_event`'s questions-mirror dial:
/// logs a warning on dial failure / an `Err` response, otherwise silent.
async fn mirror_question(req: hive_agent_sock::Request) {
    match todo_server::dial(&req).await {
        Some(hive_agent_sock::Response::Err { message }) => {
            tracing::warn!(%message, ?req, "questions mirror dial returned an error");
        }
        None => {
            tracing::warn!(?req, "questions mirror dial failed (socket unavailable?)");
        }
        Some(_) => {}
    }
}

/// Body string for the turn-failure notification we route to
/// `<parent>` on `TurnError::Failed`. Reads the hive-qualified
/// identity so the receiver sees `agent@hive` rather than relying on
/// the caller threading a `label` through every turn-handling layer.
/// Falls back to `<unknown>` when `HIVE_LABEL` is missing so a
/// misconfigured harness still produces a parseable line.
fn format_turn_failure(err: &anyhow::Error) -> String {
    let who = crate::identity::qualified_label();
    let who = if who.is_empty() {
        "<unknown>".to_owned()
    } else {
        who
    };
    format!("[system] `{who}` claude turn failed:\n{err:#}")
}

/// What a finished turn tells the serve loop to do next.
struct TurnControl {
    /// The turn ended in `AuthFailed` — caller parks on login.
    auth_failed: bool,
    /// `Some(called)` when this turn was driven by a todo wake (`from ==
    /// "todo"`) — `called` is whether the turn actually invoked
    /// `mcp__hyperhive__get_loose_ends` at some point. `None` for every
    /// other wake source, so the serve loop's miss-streak only reacts to
    /// todo-driven turns. See `TODO_MISS_PAUSE_THRESHOLD`.
    todo_wake_checked: Option<bool>,
}

/// Synthesize the message that drives a turn when an in-container producer
/// upserted a new/changed *todo* over the in-agent socket (loose-ends v2).
/// The harness owns the todo store locally and signals the serve loop
/// directly — so this wake never touches the broker (no long-poll, no
/// marker file). `id = 0` is a non-broker sentinel: the synthetic message
/// has no DB row, and `AckTurn` keys off the recipient's in-flight list
/// (which is empty here) rather than this id.
///
/// `stern` renders a harsher final-warning body instead of the usual one —
/// set when this is the last todo wake before the miss-streak would hit
/// `TODO_MISS_PAUSE_THRESHOLD`, so the agent gets one unambiguous chance to
/// avoid being auto-paused.
fn synthetic_todo_message(stern: bool) -> hive_sh4re::inbox::DeliveredMessage {
    let body = if stern {
        "you have todos — call get_loose_ends NOW. you've skipped it on recent todo \
         wakes in a row; if this turn doesn't call it, the harness will pause your \
         own turn loop until an operator resumes you."
            .to_owned()
    } else {
        "you have todos — call get_loose_ends to see them".to_owned()
    };
    hive_sh4re::inbox::DeliveredMessage {
        from: "todo".into(),
        body,
        id: 0,
        redelivered: false,
        in_reply_to: None,
    }
}

/// Synthetic message that drives the follow-up turn when a `/compact` call
/// (self-requested via the `compact` MCP tool's `wake_prompt` arg) finishes
/// and asked to be woken. Mirrors `synthetic_todo_message`'s "no broker row"
/// sentinel shape (`id = 0`) — the compact tool call itself is the durable
/// record that a wake was requested, not a broker message.
fn post_compact_wake_message(prompt: String) -> hive_sh4re::inbox::DeliveredMessage {
    hive_sh4re::inbox::DeliveredMessage {
        from: "compact".into(),
        body: prompt,
        id: 0,
        redelivered: false,
        in_reply_to: None,
    }
}

/// Synthetic message that drives the single stop-checkpoint turn when c0re
/// signals a graceful stop. The agent gets one final turn to flush durable
/// `/state` before the container is stopped; new inbound is already fenced.
fn graceful_stop_message() -> hive_sh4re::inbox::DeliveredMessage {
    hive_sh4re::inbox::DeliveredMessage {
        from: "graceful-stop".into(),
        body: "You are being gracefully stopped — the container will shut down after this turn, \
               and new inbound messages are already fenced. This is your one checkpoint turn: \
               flush anything worth keeping into your durable /state files now — update your \
               notes / CLAUDE.md / TODO.md with in-flight task state, decisions made, important \
               file paths, and whatever you'd need to resume cleanly later with only a summary \
               of this conversation to go on. Do not start new work or reply to anyone; just \
               write your notes and end your turn. The session may be compacted after this turn \
               so a later cold start resumes cheaply."
            .into(),
        id: 0,
        redelivered: false,
        in_reply_to: None,
    }
}

// ---------- surface trait ----------

/// What a `Recv` long-poll returned. Decoupled from the `Response`
/// enum so `serve_loop` can pattern-match without seeing it directly.
enum RecvOutcome {
    /// Long-poll returned at least one message; first one is detached.
    /// Also reused for a fired reminder: `reminder_timer` pushes a *real*
    /// `DeliveredMessage` (unlike `LocalTodo`'s synthetic hint) since the
    /// body/id is per-row data the producer already resolved, so the
    /// select arm wraps it straight into this variant — no dedicated
    /// `LocalReminder` variant needed.
    Message(hive_sh4re::inbox::DeliveredMessage),
    /// Long-poll timed out cleanly (empty `Messages` response). Caller
    /// sleeps then retries.
    Empty,
    /// Wire returned an error / unexpected variant. Caller logs +
    /// retries; the surface impl is responsible for tracing the
    /// detail before returning this.
    TransportError,
    /// c0re signalled a graceful stop for this agent. The serve loop runs
    /// one stop-checkpoint turn (flush durable `/state`), reports
    /// `GracefulStopComplete`, and exits so the container can be stopped.
    GracefulStop,
    /// An in-container producer upserted a new/changed todo over the
    /// in-agent socket; the serve loop drives a `synthetic_todo_message`
    /// turn. Not a broker message — the harness signalled itself directly.
    LocalTodo,
}

/// Wire surface abstraction. `AgentSurface` is the only impl — the trait
/// exists to keep the turn loop generic and testable. Every function that
/// talks to the broker goes through this so there are zero hard-coded
/// `Request` / `Response` references in the turn loop itself.
trait Surface {
    /// Ack the in-flight turn. Logs warnings on transport/broker
    /// errors but never propagates — turn loop continues either way.
    fn ack_turn(socket: &Path) -> impl Future<Output = ()>;

    /// Requeue any messages that were "in-flight" (delivered but not
    /// ack'd) — fires on harness boot to recover from a crash mid-turn.
    fn requeue_inflight(socket: &Path) -> impl Future<Output = ()>;

    /// Current inbox unread count via `Status`. Returns 0 on any
    /// transport/wire error so the caller falls through cleanly.
    fn inbox_unread(socket: &Path) -> impl Future<Output = u64>;

    /// `(open_threads, open_reminders)` for the post-turn stats row.
    /// Either field is `None` when the underlying request errors.
    fn post_turn_counts(socket: &Path) -> impl Future<Output = (Option<u64>, Option<u64>)>;

    /// Tell c0re the graceful-stop checkpoint is done and the harness is
    /// exiting its serve loop (fire-and-forget; logs on error). Lets the
    /// `GracefulStop` orchestration stop the container without waiting out
    /// its timeout fallback.
    fn graceful_stop_complete(socket: &Path) -> impl Future<Output = ()>;

    /// Tell c0re the harness's own pause-marker check (between turns) just
    /// saw the marker appear (fire-and-forget; logs on error). Lets the
    /// pause DAG's drain node resolve without waiting out its timeout
    /// fallback. Same shape as `graceful_stop_complete`.
    fn pause_acknowledged(socket: &Path) -> impl Future<Output = ()>;

    /// Send a message addressed to `<parent>` (broker resolves the
    /// sentinel via `topology::parent_of` at delivery time; root
    /// agents/manager fall through to operator).
    fn send_to_parent(socket: &Path, body: String) -> impl Future<Output = ()>;

    /// Long-poll the broker for the next message. Wraps the
    /// `Messages`/empty/error trichotomy in `RecvOutcome` so the
    /// generic `serve_loop` doesn't need the per-role Response enum
    /// at all.
    fn recv_next(socket: &Path) -> impl Future<Output = RecvOutcome>;
}

// ---------- AgentSurface ----------

/// Zero-sized type tag for the agent wire surface.
/// Talks `Request` / `Response`.
struct AgentSurface;

/// Issue an `Ok`-expecting fire-and-forget broker request, logging any
/// rejection / unexpected response / transport error under `label`. Shared by
/// the `Surface` methods that don't need the reply (`ack_turn`,
/// `requeue_inflight`, `graceful_stop_complete`).
async fn fire_and_forget(socket: &Path, req: Request, label: &str) {
    match hive_sock_client::request::<_, Response>(socket, &req, CONTROL_SOCKET_RETRY).await {
        Ok(Response::Ok) => {}
        Ok(Response::Err { message }) => {
            tracing::warn!(%message, "{label} rejected by broker");
        }
        Ok(other) => tracing::warn!(?other, "{label} unexpected response"),
        Err(e) => tracing::warn!(error = ?e, "{label} transport error"),
    }
}

impl Surface for AgentSurface {
    async fn ack_turn(socket: &Path) {
        fire_and_forget(socket, Request::AckTurn, "ack_turn").await;
    }

    async fn requeue_inflight(socket: &Path) {
        fire_and_forget(socket, Request::RequeueInflight, "requeue_inflight").await;
    }

    async fn graceful_stop_complete(socket: &Path) {
        fire_and_forget(
            socket,
            Request::GracefulStopComplete,
            "graceful_stop_complete",
        )
        .await;
    }

    async fn pause_acknowledged(socket: &Path) {
        fire_and_forget(socket, Request::PauseAcknowledged, "pause_acknowledged").await;
    }

    async fn inbox_unread(socket: &Path) -> u64 {
        match hive_sock_client::request::<_, Response>(
            socket,
            &Request::Status,
            CONTROL_SOCKET_RETRY,
        )
        .await
        {
            Ok(Response::Status { unread }) => unread,
            _ => 0,
        }
    }

    async fn post_turn_counts(socket: &Path) -> (Option<u64>, Option<u64>) {
        let threads = match hive_sock_client::request::<_, Response>(
            socket,
            &Request::GetLooseEnds { agent: None },
            CONTROL_SOCKET_RETRY,
        )
        .await
        {
            Ok(Response::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(),
            _ => None,
        };
        // Reminders are harness-local — dial the in-agent socket directly
        // instead of the broker.
        let reminders =
            match todo_server::dial(&hive_agent_sock::Request::CountPendingReminders).await {
                Some(hive_agent_sock::Response::PendingRemindersCount { count }) => Some(count),
                _ => None,
            };
        (threads, reminders)
    }

    async fn send_to_parent(socket: &Path, body: String) {
        let res = hive_sock_client::request::<_, Response>(
            socket,
            &Request::Send {
                to: hive_sh4re::manager::PARENT_RECIPIENT.into(),
                body,
                in_reply_to: None,
            },
            CONTROL_SOCKET_RETRY,
        )
        .await;
        if let Err(e) = res {
            tracing::warn!(error = ?e, "failed to notify parent of turn failure");
        }
    }

    async fn recv_next(socket: &Path) -> RecvOutcome {
        let recv: Result<Response> = hive_sock_client::request(
            socket,
            &Request::Recv {
                wait_seconds: Some(180),
                max: None,
            },
            CONTROL_SOCKET_RETRY,
        )
        .await;
        match recv {
            Ok(Response::Messages { messages, .. }) if !messages.is_empty() => {
                let first = messages.into_iter().next().expect("checked non-empty");
                RecvOutcome::Message(first)
            }
            Ok(Response::Messages { .. }) => RecvOutcome::Empty,
            Ok(Response::GracefulStop) => RecvOutcome::GracefulStop,
            Ok(Response::Err { message }) => {
                tracing::warn!(%message, "recv error");
                RecvOutcome::TransportError
            }
            Ok(other) => {
                tracing::warn!(?other, "recv produced unexpected response kind");
                RecvOutcome::TransportError
            }
            Err(e) => {
                tracing::warn!(error = ?e, "recv failed; retrying");
                RecvOutcome::TransportError
            }
        }
    }
}

// ---------- generic turn loop ----------

/// Opens the todos store and spawns the in-agent todo socket (loose-ends
/// v2 + harness-local reminders + questions): the harness owns the todo +
/// reminder + question stores locally and serves the in-container
/// producers on `HIVE_AGENT_SOCKET`. A new/changed todo upsert fires the
/// returned `Notify` so the serve loop drives a turn directly — no broker
/// round-trip, no marker files. Best-effort: if the todos store can't
/// open, the whole socket isn't served (reminder + question ops ride
/// along on the same listener, so they're gated on the same store —
/// acceptable since a from-scratch harness boot either has a writable
/// harness dir or doesn't). Split out of `serve_main` to keep it under
/// clippy's `too_many_lines` limit; kept alongside the returned `Notify`
/// so the serve loop's `LocalTodo` arm can gate a wake on `has_any()`
/// before spawning a turn — see its doc comment (the phantom-todo-wake
/// issue: a burst of same-turn upserts can arm a second `Notify` permit
/// that outlives the turn that already drained its payload).
fn spawn_todo_socket(
    reminder_store: Option<Arc<reminders::Reminders>>,
    question_store: Option<Arc<questions::Questions>>,
    bus: &Bus,
) -> (Arc<tokio::sync::Notify>, Option<Arc<todos::Todos>>) {
    let todo_wake = Arc::new(tokio::sync::Notify::new());
    let todos_store: Option<Arc<todos::Todos>> = match todos::Todos::open(&paths::state_db()) {
        Ok(store) => {
            let store = Arc::new(store);
            let wake = todo_wake.clone();
            let bus_for_socket = bus.clone();
            let store_for_socket = store.clone();
            tokio::spawn(async move {
                if let Err(e) = todo_server::run(
                    store_for_socket,
                    wake,
                    reminder_store,
                    question_store,
                    bus_for_socket,
                )
                .await
                {
                    tracing::error!(error = %e, "in-agent todo socket exited with error");
                }
            });
            // Disk-pressure watch: an in-process todo producer, so it
            // shares this store + wake directly instead of dialling the
            // socket the out-of-process producers use.
            tokio::spawn(disk_watch::run(store.clone(), todo_wake.clone()));
            // Same shape, different signal: nudge on a crowded state-dir
            // top level instead of disk pressure.
            tokio::spawn(state_entry_watch::run(store.clone(), todo_wake.clone()));
            Some(store)
        }
        Err(e) => {
            tracing::error!(error = ?e, "open todos db failed — in-agent todo socket disabled");
            None
        }
    };
    (todo_wake, todos_store)
}

/// Boot — wires up the web UI, login state, stats, plugins, forge
/// notifier, and either drops into `serve_loop` directly (`Online`) or
/// parks on the login flow first (`NeedsLogin`). See
/// `docs/turn-loop/README.md::Boot wiring`.
async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
    let port = std::env::var("HIVE_PORT")
        .ok()
        .and_then(|s| s.parse::<u16>().ok())
        .unwrap_or(DEFAULT_WEB_PORT);
    // `HIVE_LABEL` is set unconditionally by the meta-flake envelope
    // for any container-deployed agent; the `"hive"` fallback here
    // covers standalone `nix run .#hive` invocations and pre-meta
    // dev shells. Role-independent: no semantic reason for the
    // fallback to differ when the env var is missing.
    let label = std::env::var("HIVE_LABEL").unwrap_or_else(|_| "hive".into());
    let claude_dir = login::default_dir();
    let initial = LoginState::from_dir(&claude_dir);
    tracing::info!(state = ?initial, claude_dir = %claude_dir.display(), "harness boot");
    let login_state = Arc::new(Mutex::new(initial));
    let bus = Bus::new();
    // Set by the web UI's `/api/cancel` on a successful SIGINT, read-and-
    // cleared by `handle_turn` before building the next wake prompt — see
    // `hive_sh4re::inbox::INTERRUPTED_HINT`. Shared between the web server task and
    // the serve loop the same way `bus`/`todo_wake` are.
    let interrupted = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let stats = TurnStats::open_default();
    if let Some(s) = &stats {
        let (ctx, cost) = s.last_usage();
        if ctx.is_some() || cost.is_some() {
            bus.seed_usage(ctx, cost);
        }
    }
    let files = turn::TurnFiles::prepare(socket, &label).await?;
    // Plugin install failures come back as a Vec<String> — route each
    // through `<parent>` via the `send_to_parent` failure-notify path.
    // The broker resolves `<parent>` per `topology::parent_of`;
    // root agents fall through to operator.
    for failure in plugins::install_configured().await {
        S::send_to_parent(socket, failure).await;
    }
    // The forge notification poller used to be spawned here. It is its own
    // process now (`hive-forge-notify`, its own systemd unit) so a harness
    // restart doesn't take forge notifications down with it; it reaches the
    // harness the same way the bash and matrix daemons do, by upserting
    // todos on the in-agent socket.
    //
    // Agent-side cleanup of this agent's own harness artifacts (completed
    // bash-task files + verbose event rows). Runs here, not host-side in
    // hive-c0re, because the files are agent-owned — see `vacuum` module docs.
    tokio::spawn(crate::vacuum::run());
    // Log web_ui::serve's error instead of dropping it. A bare
    // `tokio::spawn(web_ui::serve(...))` discards the JoinHandle, so
    // any Err (e.g. EACCES from `bind_unix` when HIVE_WEB_SOCKET points
    // at a dir the agent user can't write) vanishes — leaving an
    // operator with no log line and no socket, debuggable only by
    // staring at lifecycle.rs.
    let web_ui_args = (
        label.clone(),
        port,
        login_state.clone(),
        bus.clone(),
        socket.to_path_buf(),
        interrupted.clone(),
    );
    tokio::spawn(async move {
        let (label, port, login_state, bus, socket, interrupted) = web_ui_args;
        if let Err(e) = web_ui::serve(label, port, login_state, bus, socket, interrupted).await {
            tracing::error!(error = %e, "web_ui::serve exited with error");
        }
    });
    // Harness-local reminders: a due reminder fires
    // straight into an mpsc channel the serve loop races against the
    // broker long-poll (unlike todos, a fire carries real per-row data,
    // so a bare `Notify` doesn't fit — see `reminder_timer` docs).
    // Best-effort: `reminder_timer::run` parks forever (never sends)
    // instead of exiting when the store can't open, so the channel
    // never observes a "closed" state. Opened before the todo socket
    // below so the same `Arc` can be handed to its request dispatch
    // (reminder ops share the todo socket/listener).
    // Fold the pre-consolidation per-concern db files into the single
    // `hyperhive-state.sqlite`, if this is the first boot since the
    // upgrade — see `db_migrate` module docs. Best-effort: a failure here
    // just leaves the legacy file(s) in place to retry next boot, so the
    // subsystems below still start (`Todos`/`Reminders::open` create the
    // schema fresh if nothing migrated across).
    if let Err(e) = db_migrate::run(
        &paths::state_db(),
        &paths::legacy_todos_db(),
        &paths::legacy_reminders_db(),
    ) {
        tracing::warn!(error = ?e, "legacy db_migrate failed — continuing with fresh/partial state db");
    }
    let (reminder_tx, reminder_rx) = tokio::sync::mpsc::unbounded_channel();
    let reminder_store = match reminders::Reminders::open(&paths::state_db()) {
        Ok(store) => Some(Arc::new(store)),
        Err(e) => {
            tracing::error!(error = ?e, "open reminders db failed — reminder delivery disabled");
            None
        }
    };
    tokio::spawn(reminder_timer::run(reminder_store.clone(), reminder_tx));
    let question_store = open_question_store();
    let (todo_wake, todos_store) =
        spawn_todo_socket(reminder_store.clone(), question_store.clone(), &bus);
    if matches!(initial, LoginState::NeedsLogin) {
        login::wait_for_login(
            &claude_dir,
            login_state.clone(),
            &bus,
            poll_ms,
            login::NO_PRIOR_FAILURE,
        )
        .await;
    } else {
        // Clear any stale `hyperhive-needs-login` sentinel left over
        // from a prior boot — `online` status writes the sentinel
        // cleanup in `Bus::emit_status`.
        bus.emit_status("online");
    }
    serve_loop::<S>(
        socket,
        Duration::from_millis(poll_ms),
        login_state,
        claude_dir,
        bus,
        stats,
        &files,
        todo_wake,
        todos_store,
        reminder_rx,
        interrupted,
    )
    .await
}

/// The long-running message loop. Long-polls the broker via
/// `S::recv_next`, drives a turn per message, parks on auth-failed,
/// otherwise retries.
#[allow(
    clippy::too_many_arguments,
    reason = "the harness's long-lived deps threaded into one serve loop, \
              wired once from main; bundling into a struct would just move the \
              same fields one level out (cf. Coordinator::open)"
)]
async fn serve_loop<S: Surface>(
    socket: &Path,
    interval: Duration,
    login_state: Arc<Mutex<LoginState>>,
    claude_dir: std::path::PathBuf,
    bus: Bus,
    stats: Option<TurnStats>,
    files: &turn::TurnFiles,
    todo_wake: Arc<tokio::sync::Notify>,
    todos_store: Option<Arc<todos::Todos>>,
    mut reminder_rx: tokio::sync::mpsc::UnboundedReceiver<hive_sh4re::inbox::DeliveredMessage>,
    interrupted: Arc<std::sync::atomic::AtomicBool>,
) -> Result<()> {
    tracing::info!(socket = %socket.display(), "harness serve");
    S::requeue_inflight(socket).await;
    // The durable claude session, built once and reused for every turn +
    // idle compaction below (it's effectively stateless).
    let session = turn::make_session(&bus);
    // Tracks the last observed pause state so the transitions get logged
    // once each instead of twelve lines a minute while parked.
    let mut was_paused = false;
    // Consecutive todo-wake turns in a row that skipped `get_loose_ends`.
    // Incremented/reset by the `todo_wake_checked` signal off each turn's
    // `TurnControl`; see `TODO_MISS_PAUSE_THRESHOLD`. Reset to 0 on resume
    // too, so a just-unparked agent gets a clean slate rather than being
    // one miss away from an instant re-pause.
    let mut todo_miss_streak: u32 = 0;
    loop {
        // Pause gate. While the marker is present this loop drives no
        // turns at all.
        //
        // Nothing here touches the broker: not calling `S::recv_next` is
        // exactly the "messages queue unacked, resume drains the
        // backlog" semantic, with no fencing and nothing to requeue.
        // Reminders (unbounded channel) and todo wakes (a `Notify`
        // permit) buffer on their own. The web UI and MCP daemons run as
        // separate tasks, so the agent stays inspectable while parked.
        //
        // A `GracefulStop` can't be observed while parked, and doesn't
        // need to be: hive-c0re skips the stop-checkpoint handshake for
        // a paused agent, because this check sits at the top of the loop
        // and so a paused agent provably has no turn in flight.
        if paths::paused_marker().exists() {
            if !was_paused {
                tracing::info!("pause marker present — parking the turn loop");
                bus.emit(LiveEvent::Note {
                    text: "paused: turn loop parked, messages will queue".into(),
                });
                was_paused = true;
                // Fire-and-forget: this is the same "no turn in flight"
                // moment `GracefulStopComplete` reports at, and needs no
                // extra tracking for the same reason — the check above is
                // already between-turns only. Lets the pause DAG's drain
                // node resolve immediately instead of timing out.
                S::pause_acknowledged(socket).await;
            }
            tokio::time::sleep(PAUSE_POLL).await;
            continue;
        }
        if was_paused {
            tracing::info!("pause marker cleared — resuming the turn loop");
            bus.emit(LiveEvent::Note {
                text: "resumed: draining whatever queued while paused".into(),
            });
            was_paused = false;
            todo_miss_streak = 0;
        }
        let next = match {
            // Idle wait: race the broker long-poll against a local
            // todo signal so an in-container producer's upsert drives a
            // turn without any broker round-trip. `biased` polls the
            // broker recv first, so a genuinely-ready inbox message is
            // never dropped in favour of the todo wake.
            tokio::select! {
                biased;
                o = S::recv_next(socket) => o,
                () = todo_wake.notified() => RecvOutcome::LocalTodo,
                Some(dm) = reminder_rx.recv() => RecvOutcome::Message(dm),
            }
        } {
            RecvOutcome::Message(first) => first,
            RecvOutcome::LocalTodo => {
                // Gate on `has_any()` before spawning a turn: a burst of
                // same-turn upserts can arm a second `Notify` permit that
                // outlives the turn which already drained its payload
                // (the phantom-todo-wake issue) — `notify_one` doesn't
                // coalesce once the first permit's been consumed, so the surplus wake
                // fires the instant the loop is back here even though
                // there's nothing left to show. Fail open (drive a turn
                // anyway) on a `has_any` error so a flaky sqlite read
                // never silently swallows a real wake.
                let has_any = todos_store
                    .as_ref()
                    .is_none_or(|store| store.has_any().unwrap_or(true));
                if !has_any {
                    tracing::debug!("todo wake fired against an empty store — stale, skipping");
                    continue;
                }
                tracing::debug!("todo wake consumed, sending synthetic todo message");
                // If this wake goes missed too, the streak hits the pause
                // threshold — say so up front instead of pausing silently.
                let stern = todo_miss_streak + 1 >= TODO_MISS_PAUSE_THRESHOLD;
                synthetic_todo_message(stern)
            }
            RecvOutcome::Empty => {
                // Idle: no message this poll. Service a queued operator
                // `/compact` here so it runs even when no turn is driving
                // (the in-flight case is handled at the end of drive_turn).
                let compacted = turn::run_pending_compact(files, &bus, &session).await;
                if !compacted {
                    tokio::time::sleep(interval).await;
                    continue;
                }
                // The compact that just ran may have carried a wake prompt
                // (the agent's own `compact` tool, not the operator button —
                // see `Bus::request_compact`). If so, drive it as a synthetic
                // turn right now instead of looping back to `recv_next` and
                // waiting for the next external event.
                let Some(prompt) = bus.take_post_compact_wake() else {
                    continue;
                };
                tracing::debug!("post-compact wake queued, driving synthetic follow-up turn");
                post_compact_wake_message(prompt)
            }
            RecvOutcome::TransportError => {
                // `recv_next` already logged the detail; just retry.
                // No backoff: the long-poll wait is itself the throttle.
                continue;
            }
            RecvOutcome::GracefulStop => {
                // c0re fenced our inbox and wants a clean stop. Run one
                // checkpoint turn so the agent flushes durable /state,
                // report completion, then exit the loop → the harness
                // process ends and the container can be stopped.
                tracing::info!(
                    "graceful stop signalled — running stop-checkpoint turn, then exiting"
                );
                let _ = handle_turn::<S>(
                    socket,
                    &bus,
                    stats.as_ref(),
                    files,
                    &session,
                    graceful_stop_message(),
                    &interrupted,
                )
                .await;
                S::graceful_stop_complete(socket).await;
                return Ok(());
            }
        };
        let turn_ctx = TurnCtx {
            socket,
            bus: &bus,
            stats: stats.as_ref(),
            files,
            session: &session,
            interrupted: &interrupted,
            login_state: &login_state,
            claude_dir: &claude_dir,
            interval,
        };
        drive_turn_and_wake_chain::<S>(&turn_ctx, next, &mut todo_miss_streak).await;
    }
}

/// Loop-invariant turn-driving context for `drive_turn_and_wake_chain`,
/// threaded as one bundle instead of double-digit positional args
/// (clippy's `too_many_arguments`). Everything here is constant for the
/// lifetime of one `serve_loop` call; only the message to drive and the
/// todo-miss streak vary per turn and stay as separate params.
struct TurnCtx<'a> {
    socket: &'a Path,
    bus: &'a Bus,
    stats: Option<&'a TurnStats>,
    files: &'a turn::TurnFiles,
    session: &'a turn::AgentSession,
    interrupted: &'a Arc<std::sync::atomic::AtomicBool>,
    login_state: &'a Arc<Mutex<LoginState>>,
    claude_dir: &'a Path,
    interval: Duration,
}

/// Drive `next`, then keep driving synthetic follow-up turns for as long as
/// a compact that just ran carries a wake prompt
/// (`Bus::take_post_compact_wake`) — see `serve_loop`'s comment at the call
/// site for why this loops in place instead of returning to the outer
/// `select!`/`recv_next`. Ordinarily runs exactly one iteration; only chains
/// further if the follow-up turn itself requests another woken compact.
/// Split out of `serve_loop` purely to keep that function under clippy's
/// line limit.
async fn drive_turn_and_wake_chain<S: Surface>(
    ctx: &TurnCtx<'_>,
    mut next: hive_sh4re::inbox::DeliveredMessage,
    todo_miss_streak: &mut u32,
) {
    loop {
        let ctrl = handle_turn::<S>(
            ctx.socket,
            ctx.bus,
            ctx.stats,
            ctx.files,
            ctx.session,
            next,
            ctx.interrupted,
        )
        .await;
        apply_todo_wake_checked(ctrl.todo_wake_checked, todo_miss_streak, ctx.bus);
        if ctrl.auth_failed {
            *ctx.login_state.lock().unwrap() = LoginState::NeedsLogin;
            // Baseline the resume check on *this instant*, not on a
            // directory snapshot taken after `wait_for_login` starts
            // polling — closes the race where a login lands between the
            // 401 and the first poll. See `wait_for_login`'s doc comment.
            login::wait_for_login(
                ctx.claude_dir,
                ctx.login_state.clone(),
                ctx.bus,
                u64::try_from(ctx.interval.as_millis()).unwrap_or(2000),
                std::time::SystemTime::now(),
            )
            .await;
        }
        let Some(prompt) = ctx.bus.take_post_compact_wake() else {
            break;
        };
        tracing::debug!("post-compact wake queued mid-turn-flow, driving synthetic follow-up turn");
        next = post_compact_wake_message(prompt);
    }
}

/// Apply a finished turn's `todo_wake_checked` signal to the miss-streak:
/// resets on a checked wake, increments (and auto-pauses at the threshold)
/// on a missed one, no-ops for a non-todo-wake turn. Split out of
/// `serve_loop` purely to keep that function under clippy's line limit —
/// see `TODO_MISS_PAUSE_THRESHOLD`'s doc comment for the design rationale.
fn apply_todo_wake_checked(checked: Option<bool>, todo_miss_streak: &mut u32, bus: &Bus) {
    match checked {
        Some(true) => *todo_miss_streak = 0,
        Some(false) => {
            *todo_miss_streak += 1;
            let streak = *todo_miss_streak;
            tracing::warn!(
                streak,
                threshold = TODO_MISS_PAUSE_THRESHOLD,
                "todo wake turn ended without calling get_loose_ends"
            );
            if streak >= TODO_MISS_PAUSE_THRESHOLD {
                tracing::warn!("todo-miss streak hit the threshold — pausing the turn loop");
                bus.emit(LiveEvent::Note {
                    text: format!(
                        "auto-paused: skipped get_loose_ends on {streak} \
                         consecutive todo wakes — an operator needs to resume this agent"
                    ),
                });
                if let Err(e) = std::fs::write(paths::paused_marker(), "") {
                    tracing::warn!(error = ?e, "failed to write pause marker");
                }
                // Fresh slate for whenever this agent gets resumed — see
                // `serve_loop`'s `was_paused` reset for the mirror side.
                *todo_miss_streak = 0;
            }
        }
        None => {}
    }
}

/// Drive a single turn: emit boot-of-turn events, run claude, ack on
/// success / requeue on rate-limit-or-401 / notify parent on failure,
/// record stats. Returns a `TurnControl` carrying the auth-failed flag and
/// the todo-wake-checked signal — the serve loop decides what to do next.
async fn handle_turn<S: Surface>(
    socket: &Path,
    bus: &Bus,
    stats: Option<&TurnStats>,
    files: &turn::TurnFiles,
    session: &turn::AgentSession,
    first: hive_sh4re::inbox::DeliveredMessage,
    interrupted: &std::sync::atomic::AtomicBool,
) -> TurnControl {
    let from = first.from;
    let body = first.body;
    let redelivered = first.redelivered;
    let msg_id = first.id;
    log_system_event(bus, &from, &body).await;
    tracing::info!(%from, %body, %redelivered, "inbox");
    let unread = S::inbox_unread(socket).await;
    bus.emit(LiveEvent::TurnStart {
        from: from.clone(),
        body: body.clone(),
        unread,
    });
    bus.set_state(TurnState::Thinking);
    let started_at = chrono::Utc::now().timestamp();
    let started_instant = std::time::Instant::now();
    let model_at_start = bus.model();
    // Read-and-clear: this wake prompt is the one turn that gets to carry
    // the "you were interrupted" banner, then the flag resets so a later
    // ordinary turn doesn't repeat a stale notice.
    let was_interrupted = interrupted.swap(false, std::sync::atomic::Ordering::Relaxed);
    let prompt = serve_common::format_wake_prompt(
        msg_id,
        &from,
        &body,
        unread,
        redelivered,
        was_interrupted,
    );
    let outcome = turn::drive_turn(&prompt, files, bus, session).await;
    turn::emit_turn_end(bus, &outcome);
    bus.set_state(TurnState::Idle);
    if outcome.is_ok() {
        S::ack_turn(socket).await;
    }
    handle_turn_error_recovery::<S>(&outcome, bus, socket).await;
    // Single read of the per-turn tool-call counter: `take_tool_calls`
    // resets it, so this is the only chance to see which tools this turn
    // invoked. Used both for the stats row below (if stats are configured)
    // and for the todo-miss-streak signal returned to the serve loop —
    // deliberately unconditional (not gated on `stats.is_some()`) so the
    // counter is reset every turn regardless, and so a stats-less harness
    // (tests, or a future config without a stats sink) still gets the
    // miss-streak signal.
    let tool_calls = bus.take_tool_calls();
    let todo_wake_checked =
        (from == "todo").then(|| tool_calls.contains_key("mcp__hyperhive__get_loose_ends"));
    if let Some(stats) = stats {
        // Fresh session this turn → mint a `sessions` row and set its id on
        // the bus so this turn (and subsequent ones until the next fresh
        // start) stamp `turn_stats.session_id`. Takes the one-shot flag
        // `run_claude` set when it suppressed `--continue`.
        if bus.take_fresh_session() {
            let sid = stats.start_session(started_at, &model_at_start);
            bus.set_session_id(sid);
        }
        let ended_at = chrono::Utc::now().timestamp();
        let duration_ms = i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX);
        let (open_threads, open_reminders) = S::post_turn_counts(socket).await;
        let row = serve_common::build_row(serve_common::TurnRowArgs {
            started_at,
            ended_at,
            duration_ms,
            model: model_at_start,
            wake_from: from.clone(),
            outcome: &outcome,
            bus,
            tool_calls,
            open_threads_count: open_threads,
            open_reminders_count: open_reminders,
        });
        stats.record(&row);
    }
    let pending = S::inbox_unread(socket).await;
    if pending > 0 {
        tracing::info!(%pending, "pending messages after turn; fetching next");
    }
    TurnControl {
        auth_failed: matches!(outcome, Err(turn::TurnError::AuthFailed)),
        todo_wake_checked,
    }
}

/// The non-happy-path half of `handle_turn`: react to each `TurnError`
/// variant the turn could have failed with (park-and-retry on rate-limit/
/// stall/auth, requeue-for-a-fresh-turn on prompt-too-long/session-not-
/// found, notify the parent on a hard failure). Split out purely to keep
/// `handle_turn` itself under clippy's line-count lint — no behavior
/// change from when this lived inline.
async fn handle_turn_error_recovery<S: Surface>(
    outcome: &turn::TurnOutcome,
    bus: &Bus,
    socket: &Path,
) {
    if matches!(outcome, Err(turn::TurnError::RateLimited)) {
        let secs = turn::rate_limit_sleep_secs();
        bus.emit_status("rate_limited");
        bus.emit(LiveEvent::Note {
            text: format!("API rate-limited — sleeping {secs}s before retry"),
        });
        tracing::warn!(sleep_secs = secs, "rate-limited; parking");
        tokio::time::sleep(Duration::from_secs(secs)).await;
        S::requeue_inflight(socket).await;
        bus.emit_status("online");
    }
    if matches!(outcome, Err(turn::TurnError::ApiStall)) {
        // Idle watchdog killed claude on a suspected API stall. Park briefly to
        // let the API recover, then requeue — same shape as the rate-limit path.
        let secs = turn::stall_sleep_secs();
        bus.emit_status("api_stall");
        bus.emit(LiveEvent::Note {
            text: format!(
                "API stall timeout — sleeping {secs}s before retry \
                 (tune HIVE_STALL_SLEEP_SECS; disable the watchdog with HIVE_TURN_IDLE_SECS=0)"
            ),
        });
        tracing::warn!(sleep_secs = secs, "API stall; parking before retry");
        tokio::time::sleep(Duration::from_secs(secs)).await;
        S::requeue_inflight(socket).await;
        bus.emit_status("online");
    }
    if matches!(outcome, Err(turn::TurnError::AuthFailed)) {
        bus.emit_status("needs_login_idle");
        bus.emit(LiveEvent::Note {
            text: "API 401 — waiting for re-login via web UI".into(),
        });
        tracing::warn!("auth-failed; parking until re-login");
        S::requeue_inflight(socket).await;
    }
    if matches!(outcome, Err(turn::TurnError::PromptTooLong)) {
        // `drive_turn` already archived the session; requeue the message so it
        // redelivers into the fresh session (which fits — the wake prompt is
        // tiny, the overflow was the now-cleared context). No status park: the
        // agent is healthy, it just needs one more delivery.
        tracing::warn!("prompt-too-long; session archived, requeueing message for a fresh turn");
        S::requeue_inflight(socket).await;
    }
    if matches!(outcome, Err(turn::TurnError::SessionNotFound)) {
        // "Shouldn't happen": resume missed and the lib's create self-heal
        // didn't resolve it. Requeue rather than ack-and-drop so the wake
        // message isn't silently lost; the next turn creates the session fresh.
        tracing::warn!("session-not-found; requeueing message for a fresh turn");
        S::requeue_inflight(socket).await;
    }
    if let Err(turn::TurnError::Failed(e)) = outcome {
        S::send_to_parent(socket, format_turn_failure(e)).await;
    }
}