hyperhive/hive-sh4re/src/inbox.rs

//! Per-agent socket — `/run/hyperhive/agents/<name>/mcp.sock` on the
//! host, bind-mounted into the container at `/run/hive/mcp.sock`. The
//! inbox/messaging wire shapes: message envelopes, the loose-ends
//! response types, and the shared wake-prompt/`recv`-result hint
//! constants + builder.

use chrono::{DateTime, Utc};
use hive_types::Ident;
use serde::{Deserialize, Serialize};

/// Server-side hard cap on `Recv.max` (see the `Recv` request). Bounds
/// the size of a single round-trip so a confused caller can't drain the
/// entire inbox in one go and blow past wire-buffer sizes; everything
/// above the cap silently clamps. 5 keeps individual turns small — a big
/// backlog is drained over several recv calls instead of one giant pop.
/// Lives here so both the enforcing side (hive-c0re's `socket_server`) and
/// the hinting side (hive-agent's wake prompt + tool docs) reference one
/// constant instead of a scattered magic value.
pub const RECV_BATCH_MAX: u32 = 5;

/// Banner prepended to a wake prompt / `recv` result when the message was
/// redelivered after a harness restart (the turn that first drove it never
/// acked). Shared between the harness serve loop (wake prompt) and the MCP
/// server (`recv` tool result) so both surfaces phrase it identically.
pub const REDELIVERY_HINT: &str = "[redelivered after harness restart — may already be handled]\n";

/// Banner prepended to a wake prompt when the previous turn was cut off by
/// an explicit operator `/cancel` (SIGINT) rather than ending normally. Set
/// once, read-and-cleared by the next turn's wake-prompt build — see
/// `hive-agent`'s `post_cancel_turn` (sets it) and `handle_turn` (clears
/// it). Lives here for the same reason as `REDELIVERY_HINT`: a single
/// phrasing, not duplicated between call sites.
pub const INTERRUPTED_HINT: &str = "[your previous turn was interrupted by the operator (/cancel) \
     before it finished — check for new messages before resuming prior work]\n";

/// Shared "(N more message(s) pending …)" advisory appended after both the
/// wake prompt body and the `recv` tool result whenever the inbox still has
/// queued messages once the current message/batch is popped. Returns an empty
/// string when `remaining == 0`. The leading `\n\n` separates it from the
/// preceding body/message block, and the suggested `max` is clamped to the
/// server-side recv cap so the hint never asks for more than one round-trip
/// can deliver. One builder so the wake prompt (harness serve loop) and the
/// in-turn recv result (MCP server) stay identical.
#[must_use]
pub fn pending_hint(remaining: u64) -> String {
    if remaining == 0 {
        return String::new();
    }
    let batch = remaining.min(u64::from(RECV_BATCH_MAX));
    format!(
        "\n\n({remaining} more message(s) pending in your inbox — call `mcp__hyperhive__recv` \
         with `max: {batch}` to drain the next batch before acting. If the \
         backlog is stale/already handled, `ack_until(up_to: <highest [msg #N] seen>)` \
         clears everything up to that id in one call instead.)"
    )
}

/// A logical message between agents.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Message {
    pub from: Ident,
    pub to: String,
    pub body: String,
    /// Optional broker row-id of the message this is a reply to.
    /// Stored in the DB and echoed back on `Recv` so the dashboard can
    /// render conversation threads. `None` for messages that start a
    /// new thread. Ignored if the referenced id is unknown or out of
    /// retention — purely advisory.
    pub in_reply_to: Option<i64>,
}

/// One row of a broker inbox query — what the dashboard renders in
/// its operator-inbox section and what a per-agent web UI returns
/// from a `Recent` request. Lives in `hive_sh4re` so it can travel
/// over both the dashboard's `/api/state` and the agent socket
/// without an internal-to-wire conversion.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct InboxRow {
    pub id: i64,
    pub from: String,
    pub body: String,
    pub at: i64,
    /// Row-id of the message this is a reply to, if any.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub in_reply_to: Option<i64>,
}

/// One delivered message in a `Recv` response.
/// See `docs/process/conventions.md::Broker delivery + ack cycle` for the
/// full delivery/ack/requeue story.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DeliveredMessage {
    pub from: String,
    pub body: String,
    /// Broker row id, tracked by the harness for `AckTurn`. Opaque to
    /// claude. `default` for wire backwards-compat.
    #[serde(default)]
    pub id: i64,
    /// `true` if this row was resurfaced by `RequeueInflight` (previously
    /// popped, never acked). Formatter prepends a "may already be handled"
    /// hint when set.
    #[serde(default)]
    pub redelivered: bool,
    /// Row-id of the message this is a reply to, if any.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub in_reply_to: Option<i64>,
}

/// Reminder timing: either relative (wait N seconds) or absolute (at unix
/// timestamp).
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "timing_type", rename_all = "snake_case")]
pub enum ReminderTiming {
    /// Remind after this many seconds from now.
    InSeconds { seconds: u64 },
    /// Remind at this unix timestamp (seconds since epoch).
    At { unix_timestamp: i64 },
}

/// One row in the response to `GetLooseEnds`. Tagged enum so new
/// thread kinds can land without breaking existing handlers.
/// Per-flavour scoping + per-variant fields + clock-anomaly
/// saturation behaviour live in
/// `docs/process/conventions.md::Loose-ends wire shape`.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum LooseEnd {
    /// A pending approval row.
    Approval {
        id: i64,
        agent: String,
        commit_ref: String,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        description: Option<String>,
        age_seconds: u64,
    },
    /// A scheduled but un-delivered reminder row.
    Reminder {
        id: i64,
        owner: String,
        message: String,
        due_at: DateTime<Utc>,
        age_seconds: u64,
    },
    /// Undelivered inbox messages waiting to be `recv`'d by this agent.
    /// Not cancellable — drain them with `recv`. Surfaced so an agent
    /// doing a between-turns `get_loose_ends` sweep sees it still owes
    /// itself a `recv` without having to poll the inbox separately. Only
    /// emitted when `count > 0`.
    PendingMessages {
        /// Number of undelivered messages queued for this agent.
        count: u64,
    },
    /// Unread matrix notifications in one or more rooms. Not cancellable —
    /// use `mark_read` via the matrix MCP to clear. Injected by the
    /// in-container harness (not hive-c0re) because the matrix daemon
    /// runs inside the agent container.
    UnreadMatrix {
        /// Number of rooms with at least one unread notification.
        rooms: u32,
        /// Per-room summary: one line per room with truncated last-message
        /// body when count is 1, or just the unread count otherwise. Empty
        /// when the daemon returned no per-room detail.
        #[serde(default)]
        summary: String,
    },
    /// A dynamic, subsystem-pushed todo (loose-ends v2). Produced by an
    /// in-container subsystem via `UpsertTodo` — matrix/bash/forge are the
    /// *built-in* producers that ship today, but `subsystem` is a plain
    /// string, not a closed set: any user-configured MCP server declared
    /// in an agent's `agent.nix` can dial the in-agent socket and push its
    /// own todos the same way. Cleared by that subsystem (`ClearTodo`) or
    /// by the agent itself (`MarkTodoDone`, by `id`).
    Todo {
        id: i64,
        /// Producing subsystem marker (`"matrix"`, `"forge"`, `"bash"`, …
        /// — built-in producers; a user-configured MCP server can push
        /// its own arbitrary marker here too, nothing enforces the set).
        subsystem: String,
        /// Optional subsystem-specific key (matrix room id, bash task id).
        #[serde(default, skip_serializing_if = "Option::is_none")]
        subsystem_key: Option<String>,
        summary: String,
        /// Optional free-text provenance (room name / task label).
        #[serde(default, skip_serializing_if = "Option::is_none")]
        source: Option<String>,
        age_seconds: u64,
    },
}

/// Age in seconds between two unix timestamps, clamped to 0 when `then`
/// is in the future. This is what backs the `age_seconds` promise on
/// every [`LooseEnd`] variant above, so it lives beside the enum rather
/// than in either producer: hive-c0re derives approval ages here, and
/// hive-agent's in-agent socket server derives reminder + todo ages.
///
/// Both clamps are load-bearing and neither substitutes for the other:
/// `saturating_sub` keeps the subtraction itself from overflowing on
/// absurd inputs, and the `try_from` is what turns a negative delta into
/// 0 — on its own `saturating_sub` saturates towards `i64::MIN`, which is
/// still negative and would wrap to ~`u64::MAX` in an `as` cast.
#[must_use]
pub fn saturating_age(now: i64, then: i64) -> u64 {
    let delta = now.saturating_sub(then);
    u64::try_from(delta).unwrap_or(0)
}

/// Kind discriminator for `CancelLooseEnd`. Per-kind store +
/// authorisation rules live in
/// `docs/process/conventions.md::Loose-ends wire shape`.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum CancelLooseEndKind {
    Reminder,
    /// Withdraw a pending approval (manager surface only).
    Approval,
}

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

    #[test]
    fn saturating_age_normal_case() {
        assert_eq!(saturating_age(1_000_000, 999_990), 10);
    }

    #[test]
    fn saturating_age_zero_when_equal() {
        assert_eq!(saturating_age(42, 42), 0);
    }

    #[test]
    fn saturating_age_handles_clock_back_step() {
        // `now` < `then`: the clock went backwards between rows. 0 beats
        // both a negative and the ~u64::MAX an `as` cast would produce,
        // which renders as "27 billion years ago" in the wake prompt.
        assert_eq!(saturating_age(100, 200), 0);
    }

    #[test]
    fn a_far_future_timestamp_stays_zero_rather_than_wrapping() {
        // The back-step case above is one second of skew; this is the
        // whole i64 range, where the subtraction itself saturates. Both
        // must land on 0, and the pair is what pins the two clamps
        // together — either one alone passes one of these and not both.
        assert_eq!(saturating_age(i64::MIN, i64::MAX), 0);
        assert_eq!(saturating_age(0, i64::MAX), 0);
    }

    #[test]
    fn a_far_past_timestamp_is_a_real_age_not_a_clamp() {
        // Control for the two arms above: the maximum representable
        // delta still comes back as itself, so their 0 is the clamp
        // firing and not this function bottoming out on large inputs.
        let age = saturating_age(i64::MAX, i64::MIN);
        assert_eq!(age, u64::try_from(i64::MAX).unwrap());
        assert_eq!(
            saturating_age(i64::MAX, 0),
            u64::try_from(i64::MAX).unwrap()
        );
    }
}