basemind 0.29.1

//! `basemind agents` — the CLI half of the `agents` domain.
//!
//! Real clap subcommands rather than a `--mode` flag, so each operation keeps its own `--help` and
//! its own argument validation; they map one-to-one onto the MCP `agents` tool's
//! [`AgentsMode`](crate::mcp::mode::AgentsMode) values, which is what `tests/cli_parity.rs` asserts.
//!
//! Unlike the code-map / memory CLI groups — which build a one-shot
//! [`BasemindServer`](crate::mcp::BasemindServer) and call the identical `#[tool]` an MCP client
//! would — these subcommands connect to the user-global broker daemon DIRECTLY via
//! [`CommsClient::ensure_and_connect`]. Building a full server here would take the repo index lock
//! and clash with a running `basemind serve`; the daemon is a separate process, so a thin client is
//! both correct and lock-free.
//!
//! This is also the human-admin path: a person can inspect (`thread-list`, `members`, `history`)
//! and ARCHIVE any thread they created. `--json` emits the structured response for every mode.
//!
//! Multi-identity: every mode accepts `--as-agent <AGENT_ID>` to connect to the broker AS a named
//! sub-identity instead of the CLI's default (`cli_agent_id`).

#![cfg(all(feature = "comms", any(unix, windows)))]

use std::io::Write;
use std::path::Path;

use anyhow::{Context, Result, bail};
use clap::Subcommand;

use crate::cli::render::render_json;
use crate::comms::client::{CommsClient, scope_context_for};
use crate::comms::cursor::Cursor;
use crate::comms::ids::{AgentId, ThreadId};
use crate::comms::model::{AgentCard, Thread};
use crate::comms::protocol::SeqMeta;
use crate::mcp::types_comms::{
    AgentListResponse, AgentRegisterResponse, AgentSummary, CursorAdvance, InboxAckResponse, InboxReadResponse,
    InboxWaitResponse, MessageFrontMatter, MessageGetResponse, ThreadArchiveResponse, ThreadHistoryResponse,
    ThreadListResponse, ThreadMemberChangeResponse, ThreadMembersResponse, ThreadMembershipResponse,
    ThreadPostResponse, ThreadStartResponse, ThreadSummary,
};

/// Default page size for `history` / `inbox` when `--limit` is omitted.
const DEFAULT_LIMIT: u32 = 100;
/// Hard page cap, mirroring the broker's `MAX_LIMIT`.
const MAX_LIMIT: u32 = 1000;
/// Default recency window for `history` / `inbox` when `--since-hours` is omitted.
const DEFAULT_SINCE_HOURS: u32 = 24;
/// Microseconds in one hour — scale factor for `--since-hours` → absolute `since_micros` cutoff.
const MICROS_PER_HOUR: i64 = 3_600_000_000;
/// Thread-freshness window in hours: 168h = 7 days, matching the MCP `ThreadSummary` rule.
const STALE_AFTER_HOURS: i64 = 168;
/// Default long-poll timeout for `wait` when `--timeout-secs` is omitted.
const DEFAULT_WAIT_SECS: u32 = 30;
/// Hard cap on `wait`'s `--timeout-secs`, mirroring the MCP `agents` tool's `wait` cap.
const MAX_WAIT_SECS: u32 = 300;

/// The `agents` domain's CLI subcommands, one per MCP mode; they talk to the broker daemon
/// directly.
#[derive(Subcommand, Debug)]
pub enum AgentsCmd {
    /// Preview retention cleanup by default, or apply it explicitly.
    Cleanup {
        /// Preview only (the default); accepted for explicit operator intent.
        #[arg(long, conflicts_with = "apply")]
        dry_run: bool,
        /// Apply the reported cleanup.
        #[arg(long)]
        apply: bool,
        #[arg(long)]
        message_ttl_hours: Option<u32>,
        #[arg(long)]
        thread_idle_hours: Option<u32>,
        #[arg(long)]
        thread_retention_hours: Option<u32>,
        #[arg(long)]
        agent_ttl_hours: Option<u32>,
        #[arg(long)]
        claim_ttl_hours: Option<u32>,
        #[arg(long)]
        as_agent: Option<String>,
    },
    /// Report active/stale agents and the last successful maintenance time.
    Status {
        /// Override the generated-agent stale window in hours.
        #[arg(long)]
        agent_ttl_hours: Option<u32>,
        #[arg(long)]
        as_agent: Option<String>,
    },
    /// Register or update this agent's A2A card with the broker.
    Register {
        /// Human-readable agent name.
        #[arg(long)]
        name: Option<String>,
        /// One-line description of the agent's purpose.
        #[arg(long)]
        description: Option<String>,
        /// Agent version string.
        #[arg(long)]
        version: Option<String>,
        /// Advertised skill label (repeatable).
        #[arg(long = "skill")]
        skills: Vec<String>,
        /// Act as this sub-identity instead of the CLI's default agent id.
        #[arg(long)]
        as_agent: Option<String>,
    },
    /// List agents known to the broker, optionally restricted to one thread's members.
    List {
        /// Restrict to members of this thread.
        #[arg(long)]
        thread: Option<String>,
        /// Act as this sub-identity instead of the CLI's default agent id.
        #[arg(long)]
        as_agent: Option<String>,
    },
    /// Start a thread addressed by at least two of subject / path / members.
    ThreadStart {
        /// Topic string.
        #[arg(long)]
        subject: Option<String>,
        /// Path or GLOB pattern (globset syntax) for path-based discovery.
        #[arg(long)]
        path: Option<String>,
        /// Explicit additional member agent id (repeatable; you are added automatically).
        #[arg(long = "member")]
        members: Vec<String>,
        /// Act as this sub-identity instead of the CLI's default agent id.
        #[arg(long)]
        as_agent: Option<String>,
    },
    /// List threads discoverable to this agent (member / path-match / subject filter).
    ThreadList {
        /// Case-sensitive substring filter over thread subjects.
        #[arg(long)]
        subject_contains: Option<String>,
        /// Also list archived threads.
        #[arg(long)]
        include_archived: bool,
        /// Act as this sub-identity instead of the CLI's default agent id.
        #[arg(long)]
        as_agent: Option<String>,
    },
    /// Join a thread.
    Join {
        /// Thread to join.
        thread: String,
        /// Act as this sub-identity instead of the CLI's default agent id.
        #[arg(long)]
        as_agent: Option<String>,
    },
    /// Leave a thread.
    Leave {
        /// Thread to leave.
        thread: String,
        /// Act as this sub-identity instead of the CLI's default agent id.
        #[arg(long)]
        as_agent: Option<String>,
    },
    /// List the members of a thread.
    Members {
        /// Thread whose members to list.
        thread: String,
        /// Act as this sub-identity instead of the CLI's default agent id.
        #[arg(long)]
        as_agent: Option<String>,
    },
    /// Add a member to a thread (creator only).
    AddMember {
        /// Thread to modify.
        thread: String,
        /// Member agent id to add.
        member: String,
        /// Act as this sub-identity instead of the CLI's default agent id.
        #[arg(long)]
        as_agent: Option<String>,
    },
    /// Remove a member from a thread (creator only).
    RemoveMember {
        /// Thread to modify.
        thread: String,
        /// Member agent id to remove.
        member: String,
        /// Act as this sub-identity instead of the CLI's default agent id.
        #[arg(long)]
        as_agent: Option<String>,
    },
    /// Archive a thread (creator / human-admin only).
    Archive {
        /// Thread to archive.
        thread: String,
        /// Act as this sub-identity instead of the CLI's default agent id.
        #[arg(long)]
        as_agent: Option<String>,
    },
    /// Post a message to a thread.
    Post {
        /// Target thread.
        thread: String,
        /// Subject line.
        subject: String,
        /// Message body (markdown). Empty when omitted.
        #[arg(long)]
        body: Option<String>,
        /// Free-form tag (repeatable).
        #[arg(long = "tag")]
        tags: Vec<String>,
        /// Id of the message being replied to.
        #[arg(long)]
        reply_to: Option<String>,
        /// Act as this sub-identity instead of the CLI's default agent id.
        #[arg(long)]
        as_agent: Option<String>,
        /// Caller-chosen key: re-sending it for the same agent and thread returns the original
        /// message id instead of storing a duplicate. Generated when omitted.
        #[arg(long)]
        idempotency_key: Option<String>,
    },
    /// Read a thread's history (front-matter only; bodies via `message`).
    History {
        /// Thread to read.
        thread: String,
        /// Resume token from a previous page's `next_cursor`.
        #[arg(long)]
        cursor: Option<String>,
        /// Maximum messages to return (default 100, max 1000).
        #[arg(long)]
        limit: Option<u32>,
        /// Only return messages from the last N hours (default 24). Pass 0 for ALL history.
        #[arg(long)]
        since_hours: Option<u32>,
        /// Act as this sub-identity instead of the CLI's default agent id.
        #[arg(long)]
        as_agent: Option<String>,
    },
    /// Print a single message BODY by compact reference or legacy id (the only body path).
    Message {
        /// Compact `message_ref` (preferred) or legacy `id` from a front-matter row.
        message_id: String,
        /// Act as this sub-identity instead of the CLI's default agent id.
        #[arg(long)]
        as_agent: Option<String>,
    },
    /// Read this agent's inbox across joined threads (front-matter only).
    Inbox {
        /// Resume token from a previous page's `next_cursor`.
        #[arg(long)]
        cursor: Option<String>,
        /// Maximum messages to return (default 100, max 1000).
        #[arg(long)]
        limit: Option<u32>,
        /// Advance read cursors past the returned messages.
        #[arg(long)]
        mark_read: bool,
        /// Only return messages from the last N hours (default 24). Pass 0 for ALL history.
        #[arg(long)]
        since_hours: Option<u32>,
        /// Act as this sub-identity instead of the CLI's default agent id.
        #[arg(long)]
        as_agent: Option<String>,
    },
    /// Clear read messages from your inbox by advancing per-thread read cursors.
    Ack {
        /// Message id to acknowledge (repeatable).
        #[arg(long = "message-id")]
        message_ids: Vec<String>,
        /// Thread whose read cursor to advance in bulk; pair with `--to-seq`.
        #[arg(long)]
        thread: Option<String>,
        /// Advance `--thread`'s read cursor straight to this per-thread seq.
        #[arg(long, requires = "thread")]
        to_seq: Option<u64>,
        /// Act as this sub-identity instead of the CLI's default agent id.
        #[arg(long)]
        as_agent: Option<String>,
    },
    /// Block until a peer posts to a joined thread (or one thread), or until the timeout elapses.
    Wait {
        /// Restrict the wake to this thread only. Omit to wake on any joined thread.
        #[arg(long)]
        thread: Option<String>,
        /// Maximum seconds to block before returning `timed_out: true` (default 30, max 300).
        #[arg(long)]
        timeout_secs: Option<u32>,
        /// Only consider messages from the last N hours (default 24). Pass 0 for ALL history.
        #[arg(long)]
        since_hours: Option<u32>,
        /// Resume token from a previous page's `next_cursor`.
        #[arg(long)]
        cursor: Option<String>,
        /// Act as this sub-identity instead of the CLI's default agent id.
        #[arg(long)]
        as_agent: Option<String>,
    },
}

/// Clamp a caller limit to `[1, MAX_LIMIT]`, defaulting when absent.
fn clamp_limit(limit: Option<u32>) -> u32 {
    limit.unwrap_or(DEFAULT_LIMIT).clamp(1, MAX_LIMIT)
}

/// Translate `--since-hours` into the absolute `since_micros` cutoff the broker filters on.
fn since_cutoff(since_hours: Option<u32>) -> Option<i64> {
    let hours = since_hours.unwrap_or(DEFAULT_SINCE_HOURS);
    if hours == 0 {
        None
    } else {
        Some(crate::comms::model::now_micros() - i64::from(hours) * MICROS_PER_HOUR)
    }
}

/// Resolve the CLI agent identity through the ONE shared resolver
/// ([`crate::comms::identity::cli_agent_id`]) so a CLI mode and the `serve` session in the SAME
/// workspace share an identity — and two different workspaces never do.
fn cli_agent_id(root: &Path) -> AgentId {
    crate::comms::identity::cli_agent_id(root)
}

/// Dispatch one `agents` subcommand. Builds a small current-thread runtime, then runs it.
pub fn run(root: &Path, json: bool, cmd: AgentsCmd) -> Result<()> {
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .context("build tokio runtime")?;
    runtime.block_on(async move {
        let mut out = std::io::stdout().lock();
        dispatch(root, json, cmd, &mut out).await
    })
}

/// Connect a [`CommsClient`] to the broker as a resolved identity.
async fn connect_as(root: &Path, as_agent: Option<String>) -> Result<CommsClient> {
    let agent = match as_agent {
        Some(raw) => AgentId::parse(raw.clone()).with_context(|| format!("invalid --as-agent {raw:?}"))?,
        None => cli_agent_id(root),
    };
    let (remote, cwd) = scope_context_for(root);
    CommsClient::ensure_and_connect(agent, remote, cwd)
        .await
        .map_err(|e| anyhow::anyhow!("connect to comms daemon: {e}"))
}

/// Parse the repeatable `--member` args into validated [`AgentId`]s.
fn parse_members(raw: Vec<String>) -> Result<Vec<AgentId>> {
    raw.into_iter()
        .map(|m| AgentId::parse(m.clone()).with_context(|| format!("invalid --member {m:?}")))
        .collect()
}

/// Run the mode: resolve the identity, connect, call the client method, and render to `out`.
async fn dispatch(root: &Path, json: bool, cmd: AgentsCmd, out: &mut impl Write) -> Result<()> {
    match cmd {
        AgentsCmd::Cleanup {
            dry_run: _,
            apply,
            message_ttl_hours,
            thread_idle_hours,
            thread_retention_hours,
            agent_ttl_hours,
            claim_ttl_hours,
            as_agent,
        } => {
            let mut client = connect_as(root, as_agent).await?;
            let report = client
                .cleanup_agents(
                    apply,
                    hours_or_default(message_ttl_hours, crate::comms::store::MESSAGE_TTL),
                    hours_or_default(thread_idle_hours, crate::comms::store::THREAD_IDLE_TTL),
                    hours_or_default(thread_retention_hours, crate::comms::store::THREAD_RETENTION_TTL),
                    hours_or_default(agent_ttl_hours, crate::comms::store::EPHEMERAL_AGENT_TTL),
                    hours_or_default(claim_ttl_hours, crate::comms::identity::CLAIM_TTL),
                )
                .await
                .map_err(|error| anyhow::anyhow!("cleanup agents: {error}"))?;
            render_serializable(&report, json, out)?;
        }
        AgentsCmd::Status {
            agent_ttl_hours,
            as_agent,
        } => {
            let mut client = connect_as(root, as_agent).await?;
            let report = client
                .agents_status(hours_or_default(
                    agent_ttl_hours,
                    crate::comms::store::EPHEMERAL_AGENT_TTL,
                ))
                .await
                .map_err(|error| anyhow::anyhow!("agents status: {error}"))?;
            render_serializable(&report, json, out)?;
        }
        AgentsCmd::Register {
            name,
            description,
            version,
            skills,
            as_agent,
        } => {
            let mut client = connect_as(root, as_agent).await?;
            let card = AgentCard {
                name: name.unwrap_or_default(),
                description: description.unwrap_or_default(),
                version: version.unwrap_or_default(),
                skills,
            };
            let agent_id = client.agent().as_str().to_string();
            client
                .register_agent(card)
                .await
                .map_err(|e| anyhow::anyhow!("register: {e}"))?;
            if json {
                render_response(
                    &AgentRegisterResponse {
                        agent_id,
                        registered: true,
                    },
                    out,
                )?;
            } else {
                writeln!(out, "registered as {agent_id}")?;
            }
        }
        AgentsCmd::List { thread, as_agent } => {
            let mut client = connect_as(root, as_agent).await?;
            let thread = thread.map(ThreadId::parse).transpose().context("thread id")?;
            let agents = client
                .list_agents(thread)
                .await
                .map_err(|e| anyhow::anyhow!("list agents: {e}"))?;
            if json {
                let agents: Vec<AgentSummary> = agents.iter().map(AgentSummary::from_record).collect();
                render_response(
                    &AgentListResponse {
                        total: agents.len(),
                        agents,
                    },
                    out,
                )?;
            } else if agents.is_empty() {
                writeln!(out, "no agents")?;
            } else {
                for a in &agents {
                    writeln!(out, "{}\t{}\t{}", a.agent_id.as_str(), a.card.name, a.card.version)?;
                }
            }
        }
        AgentsCmd::ThreadStart {
            subject,
            path,
            members,
            as_agent,
        } => {
            let mut client = connect_as(root, as_agent).await?;
            let members = parse_members(members)?;
            let thread = client
                .start_thread(subject, path, members)
                .await
                .map_err(|e| anyhow::anyhow!("thread start: {e}"))?;
            render_thread(&thread, json, out)?;
        }
        AgentsCmd::ThreadList {
            subject_contains,
            include_archived,
            as_agent,
        } => {
            let mut client = connect_as(root, as_agent).await?;
            let (remote, cwd) = scope_context_for(root);
            let threads = client
                .list_threads(remote, cwd, subject_contains, include_archived)
                .await
                .map_err(|e| anyhow::anyhow!("list threads: {e}"))?;
            let now = crate::comms::model::now_micros();
            if json {
                let threads: Vec<ThreadSummary> = threads.iter().map(|t| ThreadSummary::from_thread(t, now)).collect();
                render_response(
                    &ThreadListResponse {
                        total: threads.len(),
                        threads,
                    },
                    out,
                )?;
            } else if threads.is_empty() {
                writeln!(out, "no threads")?;
            } else {
                for t in &threads {
                    let marker = if is_stale(t, now) { "STALE" } else { "ACTIVE" };
                    let state = if t.active { marker } else { "ARCHIVED" };
                    writeln!(
                        out,
                        "{}\t{}\t{}\t{}",
                        t.id.as_str(),
                        t.subject.as_deref().unwrap_or("-"),
                        t.last_activity,
                        state
                    )?;
                }
            }
        }
        AgentsCmd::Join { thread, as_agent } => {
            let mut client = connect_as(root, as_agent).await?;
            let thread_id = ThreadId::parse(thread).context("thread id")?;
            let label = thread_id.as_str().to_string();
            client
                .join_thread(thread_id)
                .await
                .map_err(|e| anyhow::anyhow!("join: {e}"))?;
            render_flag(json, out, &label, "joined")?;
        }
        AgentsCmd::Leave { thread, as_agent } => {
            let mut client = connect_as(root, as_agent).await?;
            let thread_id = ThreadId::parse(thread).context("thread id")?;
            let label = thread_id.as_str().to_string();
            client
                .leave_thread(thread_id)
                .await
                .map_err(|e| anyhow::anyhow!("leave: {e}"))?;
            render_flag(json, out, &label, "left")?;
        }
        AgentsCmd::Members { thread, as_agent } => {
            let mut client = connect_as(root, as_agent).await?;
            let thread_id = ThreadId::parse(thread).context("thread id")?;
            let label = thread_id.as_str().to_string();
            let members = client
                .thread_members(thread_id)
                .await
                .map_err(|e| anyhow::anyhow!("members: {e}"))?;
            let ids: Vec<String> = members.iter().map(|m| m.as_str().to_string()).collect();
            if json {
                render_response(
                    &ThreadMembersResponse {
                        thread: label,
                        members: ids,
                    },
                    out,
                )?;
            } else if ids.is_empty() {
                writeln!(out, "no members")?;
            } else {
                for m in &ids {
                    writeln!(out, "{m}")?;
                }
            }
        }
        AgentsCmd::AddMember {
            thread,
            member,
            as_agent,
        } => {
            let mut client = connect_as(root, as_agent).await?;
            let thread_id = ThreadId::parse(thread).context("thread id")?;
            let member_id = AgentId::parse(member).context("member id")?;
            let label = thread_id.as_str().to_string();
            let member_label = member_id.as_str().to_string();
            client
                .add_member(thread_id, member_id)
                .await
                .map_err(|e| anyhow::anyhow!("add member: {e}"))?;
            if json {
                render_response(
                    &ThreadMemberChangeResponse {
                        thread: label,
                        member: member_label,
                        added: true,
                        removed: false,
                    },
                    out,
                )?;
            } else {
                writeln!(out, "added {member_label} to {label}")?;
            }
        }
        AgentsCmd::RemoveMember {
            thread,
            member,
            as_agent,
        } => {
            let mut client = connect_as(root, as_agent).await?;
            let thread_id = ThreadId::parse(thread).context("thread id")?;
            let member_id = AgentId::parse(member).context("member id")?;
            let label = thread_id.as_str().to_string();
            let member_label = member_id.as_str().to_string();
            client
                .remove_member(thread_id, member_id)
                .await
                .map_err(|e| anyhow::anyhow!("remove member: {e}"))?;
            if json {
                render_response(
                    &ThreadMemberChangeResponse {
                        thread: label,
                        member: member_label,
                        added: false,
                        removed: true,
                    },
                    out,
                )?;
            } else {
                writeln!(out, "removed {member_label} from {label}")?;
            }
        }
        AgentsCmd::Archive { thread, as_agent } => {
            let mut client = connect_as(root, as_agent).await?;
            let thread_id = ThreadId::parse(thread).context("thread id")?;
            let label = thread_id.as_str().to_string();
            client
                .archive_thread(thread_id)
                .await
                .map_err(|e| anyhow::anyhow!("archive: {e}"))?;
            if json {
                render_response(
                    &ThreadArchiveResponse {
                        thread: label,
                        archived: true,
                    },
                    out,
                )?;
            } else {
                writeln!(out, "archived {label}")?;
            }
        }
        AgentsCmd::Post {
            thread,
            subject,
            body,
            tags,
            reply_to,
            as_agent,
            idempotency_key,
        } => {
            let mut client = connect_as(root, as_agent).await?;
            let thread_id = ThreadId::parse(thread).context("thread id")?;
            let body = body.unwrap_or_default().into_bytes();
            let message_id = match idempotency_key {
                Some(key) => {
                    client
                        .post_message_keyed(thread_id, subject, body, tags, reply_to, key)
                        .await
                }
                None => client.post_message(thread_id, subject, body, tags, reply_to).await,
            }
            .map_err(|e| anyhow::anyhow!("post: {e}"))?;
            if json {
                render_response(&ThreadPostResponse { message_id }, out)?;
            } else {
                writeln!(out, "{message_id}")?;
            }
        }
        AgentsCmd::History {
            thread,
            cursor,
            limit,
            since_hours,
            as_agent,
        } => {
            let mut client = connect_as(root, as_agent).await?;
            let thread_id = ThreadId::parse(thread).context("thread id")?;
            let (messages, next_cursor) = client
                .read_history(
                    thread_id,
                    cursor.map(Cursor),
                    clamp_limit(limit),
                    since_cutoff(since_hours),
                )
                .await
                .map_err(|e| anyhow::anyhow!("history: {e}"))?;
            render_front_matter(&messages, next_cursor.as_ref(), None, json, out)?;
        }
        AgentsCmd::Message { message_id, as_agent } => {
            let mut client = connect_as(root, as_agent).await?;
            let id = message_id.clone();
            let body = client
                .get_body(message_id)
                .await
                .map_err(|e| anyhow::anyhow!("message: {e}"))?;
            let text = body.map(|b| String::from_utf8_lossy(&b).into_owned());
            if json {
                render_response(
                    &MessageGetResponse {
                        message_id: id,
                        found: text.is_some(),
                        body: text,
                    },
                    out,
                )?;
            } else {
                match text {
                    Some(b) => writeln!(out, "{b}")?,
                    None => writeln!(out, "(no such message)")?,
                }
            }
        }
        AgentsCmd::Inbox {
            cursor,
            limit,
            mark_read,
            since_hours,
            as_agent,
        } => {
            let mut client = connect_as(root, as_agent).await?;
            let (remote, cwd) = scope_context_for(root);
            let (messages, unread, next_cursor) = client
                .read_inbox(
                    remote,
                    cwd,
                    cursor.map(Cursor),
                    clamp_limit(limit),
                    mark_read,
                    since_cutoff(since_hours),
                )
                .await
                .map_err(|e| anyhow::anyhow!("inbox: {e}"))?;
            render_front_matter(&messages, next_cursor.as_ref(), Some(unread), json, out)?;
        }
        AgentsCmd::Ack {
            message_ids,
            thread,
            to_seq,
            as_agent,
        } => {
            if message_ids.is_empty() && to_seq.is_none() {
                bail!("`agents ack` requires --message-id, or a (--thread, --to-seq) pair");
            }
            let mut client = connect_as(root, as_agent).await?;
            let thread_id = thread.map(ThreadId::parse).transpose().context("thread id")?;
            let (acked, cursors) = client
                .ack_inbox(message_ids, thread_id, to_seq)
                .await
                .map_err(|e| anyhow::anyhow!("ack: {e}"))?;
            if json {
                let cursors_advanced: Vec<CursorAdvance> = cursors
                    .iter()
                    .map(|(thread, seq)| CursorAdvance {
                        thread: thread.clone(),
                        seq: *seq,
                    })
                    .collect();
                render_response(
                    &InboxAckResponse {
                        acked: acked as usize,
                        cursors_advanced,
                    },
                    out,
                )?;
            } else {
                writeln!(out, "acked: {acked}")?;
                for (thread, seq) in &cursors {
                    writeln!(out, "{thread}\t{seq}")?;
                }
            }
        }
        AgentsCmd::Wait {
            thread,
            timeout_secs,
            since_hours,
            cursor,
            as_agent,
        } => {
            let mut client = connect_as(root, as_agent).await?;
            let (remote, cwd) = scope_context_for(root);
            let thread_id = thread.map(ThreadId::parse).transpose().context("thread id")?;
            let timeout = std::time::Duration::from_secs(u64::from(
                timeout_secs.unwrap_or(DEFAULT_WAIT_SECS).clamp(1, MAX_WAIT_SECS),
            ));
            let (timed_out, messages, unread, next_cursor) = client
                .wait_inbox(
                    remote,
                    cwd,
                    thread_id,
                    since_cutoff(since_hours),
                    cursor.map(Cursor),
                    DEFAULT_LIMIT,
                    timeout,
                )
                .await
                .map_err(|e| anyhow::anyhow!("wait: {e}"))?;
            if json {
                let now = crate::comms::model::now_micros();
                let messages: Vec<MessageFrontMatter> = messages
                    .iter()
                    .map(|sm| MessageFrontMatter::from_seq_meta(sm, now))
                    .collect();
                render_response(
                    &InboxWaitResponse {
                        timed_out,
                        total: messages.len(),
                        unread,
                        messages,
                        next_cursor,
                    },
                    out,
                )?;
            } else {
                writeln!(out, "timed_out: {timed_out}")?;
                render_front_matter(&messages, next_cursor.as_ref(), Some(unread), json, out)?;
            }
        }
    }
    Ok(())
}

fn hours_or_default(hours: Option<u32>, default: std::time::Duration) -> u64 {
    hours.map_or(default.as_secs(), |value| u64::from(value).saturating_mul(60 * 60))
}

/// Print a response type as pretty JSON, the same rendering the tool-backed subcommands use.
fn render_response(value: &impl serde::Serialize, out: &mut impl Write) -> Result<()> {
    render_json(&serde_json::to_value(value)?, out)
}

fn render_serializable(value: &impl serde::Serialize, json: bool, out: &mut impl Write) -> Result<()> {
    let value = serde_json::to_value(value)?;
    if json {
        render_json(&value, out)?;
    } else if let Some(fields) = value.as_object() {
        for (name, value) in fields {
            writeln!(out, "{name}: {value}")?;
        }
    }
    Ok(())
}

/// Render a `{key: value, flag: true}` JSON object (or a plain line) for a membership toggle.
fn render_flag(json: bool, out: &mut impl Write, value: &str, flag: &str) -> Result<()> {
    if json {
        let mut membership = ThreadMembershipResponse {
            thread: value.to_string(),
            joined: false,
            left: false,
        };
        match flag {
            "joined" => membership.joined = true,
            _ => membership.left = true,
        }
        render_response(&membership, out)?;
    } else {
        writeln!(out, "{flag} {value}")?;
    }
    Ok(())
}

/// Whether a thread reads as STALE at `now_micros`: no posts yet or last post older than 7 days.
fn is_stale(thread: &Thread, now_micros: i64) -> bool {
    thread.last_activity == 0 || (now_micros - thread.last_activity) > STALE_AFTER_HOURS * MICROS_PER_HOUR
}

/// Render a single thread (started / fetched).
fn render_thread(thread: &Thread, json: bool, out: &mut impl Write) -> Result<()> {
    if json {
        render_response(
            &ThreadStartResponse {
                thread: ThreadSummary::from_thread(thread, crate::comms::model::now_micros()),
            },
            out,
        )?;
    } else {
        writeln!(
            out,
            "{}\t{}",
            thread.id.as_str(),
            thread.subject.as_deref().unwrap_or("-")
        )?;
    }
    Ok(())
}

/// Render a page of message FRONT-MATTER (never bodies). `unread` is `Some` for inbox output.
fn render_front_matter(
    messages: &[SeqMeta],
    next_cursor: Option<&Cursor>,
    unread: Option<u32>,
    json: bool,
    out: &mut impl Write,
) -> Result<()> {
    if json {
        let now = crate::comms::model::now_micros();
        let messages: Vec<MessageFrontMatter> = messages
            .iter()
            .map(|sm| MessageFrontMatter::from_seq_meta(sm, now))
            .collect();
        let total = messages.len();
        let next_cursor = next_cursor.cloned();
        return match unread {
            Some(unread) => render_response(
                &InboxReadResponse {
                    total,
                    unread,
                    messages,
                    next_cursor,
                },
                out,
            ),
            None => render_response(
                &ThreadHistoryResponse {
                    total,
                    messages,
                    next_cursor,
                },
                out,
            ),
        };
    }
    if let Some(u) = unread {
        writeln!(out, "unread: {u}")?;
    }
    if messages.is_empty() {
        writeln!(out, "(no messages)")?;
    } else {
        for sm in messages {
            let m = &sm.meta;
            writeln!(
                out,
                "{}\t{}\t{}\t{}",
                crate::comms::model::message_reference(&m.id),
                m.subject,
                m.from.as_str(),
                m.ts_micros
            )?;
        }
    }
    if let Some(c) = next_cursor {
        writeln!(out, "next_cursor: {}", c.0)?;
    }
    Ok(())
}

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

    fn json_of(render: impl FnOnce(&mut Vec<u8>) -> Result<()>) -> serde_json::Value {
        let mut buf = Vec::new();
        render(&mut buf).expect("render");
        serde_json::from_slice(&buf).expect("pretty JSON")
    }

    /// `--json` must carry the MCP `ThreadSummary` shape, not a CLI-private one: the field is
    /// `last_activity_micros` (the CLI used to say `last_activity`).
    #[test]
    fn thread_json_matches_the_mcp_thread_summary() {
        let thread = Thread {
            id: ThreadId::parse("thread-1".to_string()).expect("thread id"),
            subject: Some("topic".into()),
            path: None,
            members: vec![],
            creator: AgentId::parse("alice".to_string()).expect("agent id"),
            active: true,
            created_at: 5,
            last_activity: 0,
        };
        let value = json_of(|out| render_thread(&thread, true, out));
        let row = &value["thread"];
        assert_eq!(row["last_activity_micros"], 0);
        assert_eq!(row["stale"], true);
        assert!(row.get("last_activity").is_none(), "{row}");
        assert!(
            row.get("path").is_none(),
            "None fields are omitted like the MCP type: {row}"
        );
    }

    #[test]
    fn membership_flag_json_matches_the_mcp_response() {
        let joined = json_of(|out| render_flag(true, out, "t1", "joined"));
        assert_eq!(joined, serde_json::json!({"thread": "t1", "joined": true}));
        let left = json_of(|out| render_flag(true, out, "t1", "left"));
        assert_eq!(left, serde_json::json!({"thread": "t1", "left": true}));
    }

    #[test]
    fn empty_inbox_json_carries_unread_and_cursor() {
        let cursor = Cursor("abc".into());
        let value = json_of(|out| render_front_matter(&[], Some(&cursor), Some(3), true, out));
        assert_eq!(
            value,
            serde_json::json!({"total": 0, "unread": 3, "messages": [], "next_cursor": "abc"})
        );
        let history = json_of(|out| render_front_matter(&[], None, None, true, out));
        assert_eq!(history, serde_json::json!({"total": 0, "messages": []}));
    }
}