//! Manager socket listener. Privileged tool surface: agent-style send/recv
//! plus lifecycle verbs (Phase 4). Phase 5 will gate Spawn/Kill behind the
//! commit-approval flow; for now they hit the same code path the host admin
//! socket uses.

use std::sync::Arc;

use anyhow::{Context, Result};
use hive_sh4re::{MANAGER_AGENT, ManagerRequest, ManagerResponse, Message};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};

use crate::coordinator::Coordinator;
use crate::lifecycle;

pub fn start(coord: Arc<Coordinator>) -> Result<()> {
    let dir = Coordinator::manager_dir();
    std::fs::create_dir_all(&dir)
        .with_context(|| format!("create manager dir {}", dir.display()))?;
    let socket = Coordinator::manager_socket_path();
    if socket.exists() {
        std::fs::remove_file(&socket).context("remove stale manager socket")?;
    }
    let listener = UnixListener::bind(&socket)
        .with_context(|| format!("bind manager socket {}", socket.display()))?;
    // 0666 so the in-container hm1nd user (non-root) can connect;
    // the bind source dir is manager-only on host. See agent_server.rs.
    use std::os::unix::fs::PermissionsExt as _;
    std::fs::set_permissions(&socket, std::fs::Permissions::from_mode(0o666))
        .with_context(|| format!("chmod manager socket {}", socket.display()))?;
    tracing::info!(socket = %socket.display(), "manager socket listening");

    tokio::spawn(async move {
        loop {
            match listener.accept().await {
                Ok((stream, _)) => {
                    let coord = coord.clone();
                    tokio::spawn(async move {
                        if let Err(e) = serve(stream, coord).await {
                            tracing::warn!(error = ?e, "manager connection failed");
                        }
                    });
                }
                Err(e) => {
                    tracing::warn!(error = ?e, "manager listener accept failed");
                    return;
                }
            }
        }
    });
    Ok(())
}

async fn serve(stream: UnixStream, coord: Arc<Coordinator>) -> Result<()> {
    let (read, mut write) = stream.into_split();
    let mut reader = BufReader::new(read);
    let mut line = String::new();
    loop {
        line.clear();
        let n = reader.read_line(&mut line).await?;
        if n == 0 {
            return Ok(());
        }
        let resp = match serde_json::from_str::<ManagerRequest>(line.trim()) {
            Ok(req) => dispatch(&req, &coord).await,
            Err(e) => ManagerResponse::Err {
                message: format!("parse error: {e}"),
            },
        };
        let mut payload = serde_json::to_string(&resp)?;
        payload.push('\n');
        write.write_all(payload.as_bytes()).await?;
        write.flush().await?;
    }
}

/// Max long-poll window for manager `Recv`. Same semantics as the
/// sub-agent socket: omitted `wait_seconds` (or `0`) = peek and
/// return immediately, positive value = park up to that many
/// seconds (clamped at MAX).
const MANAGER_RECV_LONG_POLL_MAX: std::time::Duration = std::time::Duration::from_secs(180);

/// Same shape + rationale as `agent_server::RECV_BATCH_MAX`. Kept
/// numerically aligned across surfaces so a tool description that
/// quotes the cap stays accurate either way.
const MANAGER_RECV_BATCH_MAX: u32 = 32;

fn manager_recv_timeout(wait_seconds: Option<u64>) -> std::time::Duration {
    match wait_seconds {
        Some(s) => std::time::Duration::from_secs(s).min(MANAGER_RECV_LONG_POLL_MAX),
        None => std::time::Duration::ZERO,
    }
}

#[allow(clippy::too_many_lines)]
async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResponse {
    match req {
        ManagerRequest::Send {
            to,
            body,
            in_reply_to,
        } => {
            if let Err(message) = crate::limits::check_size("send", body) {
                return ManagerResponse::Err { message };
            }
            if to == "*" {
                let errors = coord.broadcast_send(MANAGER_AGENT, body);
                if errors.is_empty() {
                    ManagerResponse::Ok
                } else {
                    ManagerResponse::Err {
                        message: format!("broadcast failed for agents: {}", errors.join(", ")),
                    }
                }
            } else {
                // Resolve magic-recipient sentinels (currently `<parent>`)
                // against topology.json; no-op for ordinary names. The
                // manager has no parent in topology, so `<parent>`
                // resolves to OPERATOR_RECIPIENT — the "no parent → tell
                // the operator" fallback. See `docs/conventions.md::
                // Recipient sentinels`.
                let resolved = crate::topology::resolve_recipient(MANAGER_AGENT, to);
                match coord.broker.send(&Message {
                    from: MANAGER_AGENT.to_owned(),
                    to: resolved,
                    body: body.clone(),
                    in_reply_to: *in_reply_to,
                }) {
                    Ok(()) => ManagerResponse::Ok,
                    Err(e) => ManagerResponse::Err {
                        message: format!("{e:#}"),
                    },
                }
            }
        }
        ManagerRequest::Wake { from, body } => match coord.broker.send(&Message {
            from: from.clone(),
            to: MANAGER_AGENT.to_owned(),
            body: body.clone(),
            in_reply_to: None,
        }) {
            Ok(()) => ManagerResponse::Ok,
            Err(e) => ManagerResponse::Err {
                message: format!("{e:#}"),
            },
        },
        ManagerRequest::OperatorMsg { body } => match coord.broker.send(&Message {
            from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
            to: MANAGER_AGENT.to_owned(),
            body: body.clone(),
            in_reply_to: None,
        }) {
            Ok(()) => ManagerResponse::Ok,
            Err(e) => ManagerResponse::Err {
                message: format!("{e:#}"),
            },
        },
        ManagerRequest::Status => match coord.broker.count_pending(MANAGER_AGENT) {
            Ok(unread) => ManagerResponse::Status { unread },
            Err(e) => ManagerResponse::Err {
                message: format!("{e:#}"),
            },
        },
        ManagerRequest::Recent { limit } => match coord.broker.recent_for(MANAGER_AGENT, *limit) {
            Ok(rows) => ManagerResponse::Recent { rows },
            Err(e) => ManagerResponse::Err {
                message: format!("{e:#}"),
            },
        },
        ManagerRequest::Recv { wait_seconds, max } => {
            let cap = max.unwrap_or(1).min(MANAGER_RECV_BATCH_MAX) as usize;
            match coord
                .broker
                .recv_blocking_batch(MANAGER_AGENT, manager_recv_timeout(*wait_seconds), cap)
                .await
            {
                Ok(deliveries) => ManagerResponse::Messages {
                    messages: deliveries
                        .into_iter()
                        .map(|d| hive_sh4re::DeliveredMessage {
                            from: d.message.from,
                            body: d.message.body,
                            id: d.id,
                            redelivered: d.redelivered,
                            in_reply_to: d.message.in_reply_to,
                        })
                        .collect(),
                },
                Err(e) => ManagerResponse::Err {
                    message: format!("{e:#}"),
                },
            }
        }
        ManagerRequest::RequestInitConfig { name, description } => {
            tracing::info!(%name, "manager: request_init_config");
            let proposed_dir = crate::coordinator::Coordinator::agent_proposed_dir(name);
            if proposed_dir.join(".git").exists() {
                return ManagerResponse::Err {
                    message: format!(
                        "proposed config repo for '{name}' already exists at {} - \
                         use request_apply_commit to update an existing agent's config",
                        proposed_dir.display()
                    ),
                };
            }
            match coord.approvals.submit_kind(
                name,
                hive_sh4re::ApprovalKind::InitConfig,
                "",
                description.as_deref(),
            ) {
                Ok(id) => {
                    tracing::info!(%id, %name, "init_config approval queued");
                    coord.emit_approval_added(
                        id,
                        name,
                        "init_config",
                        None,
                        None,
                        description.clone(),
                    );
                    ManagerResponse::Ok
                }
                Err(e) => ManagerResponse::Err {
                    message: format!("{e:#}"),
                },
            }
        }
        ManagerRequest::Kill { name } => {
            tracing::info!(%name, "manager: kill");
            if name == crate::lifecycle::MANAGER_NAME {
                return ManagerResponse::Err {
                    message: "refusing to kill the manager".into(),
                };
            }
            let result: Result<()> = async {
                lifecycle::kill(name).await?;
                coord.unregister_agent(name);
                Ok(())
            }
            .await;
            match result {
                Ok(()) => {
                    coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
                        agent: name.clone(),
                    });
                    ManagerResponse::Ok
                }
                Err(e) => ManagerResponse::Err {
                    message: format!("{e:#}"),
                },
            }
        }
        ManagerRequest::Start { name } => {
            tracing::info!(%name, "manager: start");
            if name == crate::lifecycle::MANAGER_NAME {
                return ManagerResponse::Err {
                    message: "refusing to start the manager from itself".into(),
                };
            }
            match lifecycle::start(name).await {
                Ok(()) => {
                    coord.kick_agent(name, "container started");
                    ManagerResponse::Ok
                }
                Err(e) => ManagerResponse::Err {
                    message: format!("{e:#}"),
                },
            }
        }
        ManagerRequest::Restart { name } => {
            tracing::info!(%name, "manager: restart");
            if name == crate::lifecycle::MANAGER_NAME {
                return ManagerResponse::Err {
                    message: "refusing to restart the manager from itself".into(),
                };
            }
            match lifecycle::restart(name).await {
                Ok(()) => {
                    coord.kick_agent(name, "container restarted");
                    ManagerResponse::Ok
                }
                Err(e) => ManagerResponse::Err {
                    message: format!("{e:#}"),
                },
            }
        }
        ManagerRequest::Update { name } => {
            tracing::info!(%name, "manager: enqueue update");
            coord.rebuild_queue.enqueue(
                crate::rebuild_queue::QueueKind::Rebuild,
                name.to_owned(),
                crate::rebuild_queue::QueueSource::Manual,
                "manager `update` tool".to_owned(),
                None,
            );
            coord.emit_rebuild_queue_snapshot();
            ManagerResponse::Ok
        }
        ManagerRequest::RequestUpdateMetaInputs {
            inputs,
            description,
        } => {
            let label = if inputs.is_empty() {
                "all inputs".to_string()
            } else {
                inputs.join(", ")
            };
            tracing::info!(%label, "manager: request_update_meta_inputs");
            // Encode the inputs list as JSON and store it in commit_ref
            // (there's no git commit involved; the field carries the
            // payload for the approval handler to decode at run time).
            let commit_ref = serde_json::to_string(inputs).unwrap_or_default();
            let id = match coord
                .approvals
                .submit_kind(
                    hive_sh4re::MANAGER_AGENT,
                    hive_sh4re::ApprovalKind::UpdateMetaInputs,
                    &commit_ref,
                    description.as_deref(),
                )
                .map_err(|e| anyhow::anyhow!("{e:#}"))
            {
                Ok(id) => id,
                Err(e) => {
                    return ManagerResponse::Err {
                        message: format!("queue update_meta_inputs approval: {e:#}"),
                    };
                }
            };
            tracing::info!(%id, %label, "update_meta_inputs approval queued");
            coord.emit_approval_added(
                id,
                hive_sh4re::MANAGER_AGENT,
                "update_meta_inputs",
                None,
                None,
                description.clone(),
            );
            ManagerResponse::Ok
        }
        ManagerRequest::RequestSchedulePrompt(payload) => {
            handle_request_schedule_prompt(coord, hive_sh4re::MANAGER_AGENT, payload)
        }
        ManagerRequest::CancelSchedule { id, targets } => {
            handle_cancel_schedule(coord, hive_sh4re::MANAGER_AGENT, *id, targets.as_deref())
        }
        ManagerRequest::EditSchedule {
            id,
            body,
            description,
            interval_seconds,
            next_fire_at_unix,
            targets_add,
            targets_remove,
        } => handle_edit_schedule(
            coord,
            hive_sh4re::MANAGER_AGENT,
            *id,
            body.clone(),
            description.clone(),
            *interval_seconds,
            *next_fire_at_unix,
            targets_add.clone(),
            targets_remove.clone(),
        ),
        ManagerRequest::ListSchedules => match coord.scheduled_prompts.list() {
            Ok(schedules) => ManagerResponse::Schedules {
                schedules: schedules.into_iter().map(schedule_to_wire).collect(),
            },
            Err(e) => ManagerResponse::Err {
                message: format!("list scheduled prompts: {e:#}"),
            },
        },
        ManagerRequest::FireScheduleNow { id } => {
            handle_fire_schedule_now(coord, hive_sh4re::MANAGER_AGENT, *id).await
        }
        ManagerRequest::Ask {
            question,
            options,
            multi,
            ttl_seconds,
            to,
        } => crate::questions::handle_ask(
            coord,
            MANAGER_AGENT,
            question,
            options,
            *multi,
            *ttl_seconds,
            to.as_deref(),
        )
        .map_or_else(
            |message| ManagerResponse::Err { message },
            |id| ManagerResponse::QuestionQueued { id },
        ),
        ManagerRequest::Answer { id, answer } => {
            crate::questions::handle_answer(coord, MANAGER_AGENT, *id, answer).map_or_else(
                |message| ManagerResponse::Err { message },
                |()| ManagerResponse::Ok,
            )
        }
        ManagerRequest::GetLogs { agent, lines } => {
            let n = lines.unwrap_or(50);
            // `journalctl -M` wants the *machine* name, not the
            // logical agent name: `gui` → `h-gui`. `container_name`
            // does that and passes `hm1nd` through unprefixed — but
            // it doesn't know the broker-logical manager name
            // `"manager"` (it'd wrongly produce `h-manager`), so
            // handle that alias explicitly. Either manager spelling
            // resolves to the unprefixed `hm1nd` machine.
            let machine = if agent == MANAGER_AGENT {
                crate::lifecycle::MANAGER_NAME.to_owned()
            } else {
                crate::lifecycle::container_name(agent)
            };
            tracing::info!(%agent, %machine, %n, "manager: get_logs");
            match tokio::process::Command::new("journalctl")
                .args([
                    "-M",
                    &machine,
                    "-n",
                    &n.to_string(),
                    "--no-pager",
                    "--output=short",
                ])
                .output()
                .await
            {
                Ok(out) => {
                    let content = if out.status.success() || !out.stdout.is_empty() {
                        String::from_utf8_lossy(&out.stdout).into_owned()
                    } else {
                        let stderr = String::from_utf8_lossy(&out.stderr);
                        format!("journalctl exited {}: {stderr}", out.status)
                    };
                    ManagerResponse::Logs { content }
                }
                Err(e) => ManagerResponse::Err {
                    message: format!("journalctl spawn failed: {e:#}"),
                },
            }
        }
        ManagerRequest::Remind {
            message,
            timing,
            file_path,
        } => match crate::agent_server::store_remind(
            coord,
            MANAGER_AGENT,
            message,
            timing,
            file_path.as_deref(),
        ) {
            Ok(()) => ManagerResponse::Ok,
            Err(message) => ManagerResponse::Err { message },
        },
        ManagerRequest::RequestApplyCommit {
            agent,
            commit_ref,
            description,
        } => {
            tracing::info!(%agent, %commit_ref, "manager: request_apply_commit");
            match submit_apply_commit(coord, agent, commit_ref, description.as_deref()).await {
                Ok((id, sha)) => {
                    tracing::info!(%id, %agent, manager_ref = %commit_ref, %sha, "approval queued + proposal tag planted");
                    ManagerResponse::Ok
                }
                Err(e) => ManagerResponse::Err {
                    message: format!("{e:#}"),
                },
            }
        }
        ManagerRequest::GetLooseEnds { agent } => {
            let result = match agent.as_deref() {
                Some("*") => crate::loose_ends::hive_wide(coord),
                Some(name) => crate::loose_ends::for_agent(coord, name),
                None => crate::loose_ends::for_agent(coord, MANAGER_AGENT),
            };
            match result {
                Ok(loose_ends) => ManagerResponse::LooseEnds { loose_ends },
                Err(e) => ManagerResponse::Err {
                    message: format!("{e:#}"),
                },
            }
        }
        ManagerRequest::CountPendingReminders { agent } => {
            let target = agent.as_deref().unwrap_or(MANAGER_AGENT);
            match coord.broker.count_pending_reminders_for(target) {
                Ok(count) => ManagerResponse::PendingRemindersCount { count },
                Err(e) => ManagerResponse::Err {
                    message: format!("{e:#}"),
                },
            }
        }
        ManagerRequest::ReminderRollup { since_secs, agent } => {
            let target = agent.as_deref().unwrap_or(MANAGER_AGENT);
            match coord.broker.reminder_rollup_for(target, *since_secs) {
                Ok(stats) => ManagerResponse::ReminderRollup(stats),
                Err(e) => ManagerResponse::Err {
                    message: format!("{e:#}"),
                },
            }
        }
        ManagerRequest::SetStatus { text } => {
            // Cap length + reject multi-line so a confused caller
            // can't dump a multi-paragraph session report into the
            // dashboard chip.
            if let Err(message) = crate::limits::check_status_text(text) {
                return ManagerResponse::Err { message };
            }
            let path = Coordinator::agent_notes_dir(MANAGER_AGENT).join("hyperhive-status");
            let result = if text.trim().is_empty() {
                std::fs::remove_file(&path).or_else(|e| {
                    if e.kind() == std::io::ErrorKind::NotFound {
                        Ok(())
                    } else {
                        Err(e)
                    }
                })
            } else {
                std::fs::write(&path, format!("{}\n", text.trim()))
            };
            match result {
                Ok(()) => {
                    let coord2 = Arc::clone(coord);
                    tokio::spawn(async move { coord2.rescan_containers_and_emit().await });
                    ManagerResponse::Ok
                }
                Err(e) => ManagerResponse::Err {
                    message: format!("set_status write failed: {e}"),
                },
            }
        }
        ManagerRequest::GetAgentMeta { name } => {
            let target = name.as_deref().unwrap_or(MANAGER_AGENT);
            // Gate status on the target's running state so a stopped
            // container's stale on-disk status doesn't leak through.
            // Also surface `running` itself so callers can tell
            // (e.g. "iris is down" vs "iris has no status set").
            let (status_text, status_set_at, running) =
                crate::container_view::read_agent_status_live(target).await;
            let role = if target == MANAGER_AGENT {
                "manager"
            } else {
                "agent"
            }
            .to_owned();
            let (hive_name, swarm_name) = crate::container_view::hive_swarm_names();
            ManagerResponse::AgentMeta {
                name: target.to_owned(),
                role,
                running,
                hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake),
                status_text,
                status_set_at,
                hive_name,
                swarm_name,
            }
        }
        ManagerRequest::CancelLooseEnd { kind, id } => {
            crate::questions::handle_cancel_loose_end(coord, MANAGER_AGENT, *kind, *id).map_or_else(
                |message| ManagerResponse::Err { message },
                |()| ManagerResponse::Ok,
            )
        }
        ManagerRequest::AckTurn => match coord.broker.ack_turn(MANAGER_AGENT) {
            Ok(_n) => ManagerResponse::Ok,
            Err(e) => ManagerResponse::Err {
                message: format!("{e:#}"),
            },
        },
        ManagerRequest::RequeueInflight => match coord.broker.requeue_inflight(MANAGER_AGENT) {
            Ok(n) => {
                if n > 0 {
                    tracing::info!(agent = %MANAGER_AGENT, requeued = %n, "requeued in-flight messages");
                }
                ManagerResponse::Ok
            }
            Err(e) => ManagerResponse::Err {
                message: format!("{e:#}"),
            },
        },
    }
}

/// `request_apply_commit` takes a commit SHA only — not a branch or
/// tag name. A branch is mutable; pinning the proposal to a concrete
/// sha keeps "what the manager asked to deploy" unambiguous and means
/// the `proposal/<id>` tag is a faithful record of the request.
/// Accepts a 7..=40 char hex string (short or full sha); the exact
/// commit is resolved + existence-checked against the proposed repo
/// later in `lifecycle::git_fetch_to_tag`.
fn validate_commit_ref(commit_ref: &str) -> Result<()> {
    let n = commit_ref.len();
    let hex = commit_ref.chars().all(|c| c.is_ascii_hexdigit());
    if !(7..=40).contains(&n) || !hex {
        anyhow::bail!(
            "commit_ref '{commit_ref}' is not a commit sha — request_apply_commit \
             takes a 7-40 char hex sha, not a branch or tag name"
        );
    }
    Ok(())
}

/// Submit-time half of the apply flow: queue the approval row, then
/// fetch the manager's commit from the proposed repo into applied and
/// pin it as `refs/tags/proposal/<id>`. From this point on the manager
/// repo is irrelevant for this approval — even if the manager amends
/// or force-pushes, the canonical sha hive-c0re will eventually
/// approve/deny lives in applied's object DB.
///
/// If anything fails after the row is inserted (sha missing in
/// proposed, fs error, git plumbing crash) we mark the row failed and
/// surface the error to the manager. We don't try to roll the row
/// back — the failure is part of the audit trail.
async fn submit_apply_commit(
    coord: &Arc<Coordinator>,
    agent: &str,
    commit_ref: &str,
    description: Option<&str>,
) -> anyhow::Result<(i64, String)> {
    validate_commit_ref(commit_ref)?;
    let proposed_dir = crate::coordinator::Coordinator::agent_proposed_dir(agent);
    let applied_dir = crate::coordinator::Coordinator::agent_applied_dir(agent);
    if !proposed_dir.exists() {
        anyhow::bail!(
            "proposed repo missing for agent '{agent}' (expected at {})",
            proposed_dir.display()
        );
    }
    if !applied_dir.join(".git").exists() {
        // First deploy: seed the applied repo from proposed so we can plant
        // the proposal/<id> tag below. setup_applied seeds at the root
        // (template) commit of proposed, not at main, so deployed/0 is the
        // template baseline. This makes the diff mara sees on approval
        // show the manager's actual changes rather than an empty diff.
        lifecycle::setup_applied(&applied_dir, Some(&proposed_dir), agent)
            .await
            .context("seed applied repo for first spawn")?;
    }
    let id = coord
        .approvals
        .submit_kind(
            agent,
            hive_sh4re::ApprovalKind::ApplyCommit,
            commit_ref,
            description,
        )
        .map_err(|e| anyhow::anyhow!("queue approval row: {e:#}"))?;
    let tag = format!("proposal/{id}");
    let sha =
        match crate::lifecycle::git_fetch_to_tag(&applied_dir, &proposed_dir, commit_ref, &tag)
            .await
        {
            Ok(s) => s,
            Err(e) => {
                // Surface the failure on the approval row so the
                // dashboard reflects it instead of leaving a phantom
                // pending entry. The note doubles as the operator-visible
                // explanation of why the approval can't be approved.
                let note = format!("{e:#}");
                let _ = coord.approvals.mark_failed(id, &note);
                coord.emit_approval_resolved(
                    id,
                    agent,
                    "apply_commit",
                    None,
                    "failed",
                    Some(note),
                    description.map(str::to_owned),
                );
                return Err(anyhow::anyhow!("git_fetch_to_tag: {e:#}"));
            }
        };
    coord
        .approvals
        .set_fetched_sha(id, &sha)
        .map_err(|e| anyhow::anyhow!("persist fetched_sha: {e:#}"))?;
    // Pre-flight gates: both reject the apply before approval if
    // the agent's flake state would inflate meta's lock with duplicates
    // or lie about what nix will fetch. Both checks independently read
    // `<tag>:flake.lock` via git — they don't share state. Order matters
    // only for early-exit + messaging: sync first means a stale lock
    // bails with the actionable "run `nix flake lock`" hint rather than
    // a dedup pass on a lock nix would never produce.
    //
    // Runs after `set_fetched_sha` so the failed row carries the sha
    // that broke. Both failure paths mark + emit, then bail.
    let sha_short = sha[..sha.len().min(12)].to_owned();
    if let Err(e) = crate::flake_check::check_lock_in_sync(&applied_dir, &tag, id).await {
        let note = format!("{e:#}");
        let _ = coord.approvals.mark_failed(id, &note);
        coord.emit_approval_resolved(
            id,
            agent,
            "apply_commit",
            Some(sha_short.clone()),
            "failed",
            Some(note),
            description.map(str::to_owned),
        );
        return Err(anyhow::anyhow!("flake lock-sync check: {e:#}"));
    }
    if let Err(e) = crate::flake_check::check_no_duplicate_inputs(&applied_dir, &tag).await {
        let note = format!("{e:#}");
        let _ = coord.approvals.mark_failed(id, &note);
        coord.emit_approval_resolved(
            id,
            agent,
            "apply_commit",
            Some(sha_short.clone()),
            "failed",
            Some(note),
            description.map(str::to_owned),
        );
        return Err(anyhow::anyhow!("flake dedup check: {e:#}"));
    }
    // Mirror the freshly-planted proposal/<id> tag to the forge.
    if let Err(e) = crate::forge::push_config(agent).await {
        tracing::warn!(%agent, %id, error = ?e, "forge: push_config after submit failed");
    }
    // Phase 5b: surface the new pending approval on the dashboard
    // event channel. Compute the diff once here so live subscribers
    // get a fully-formed row without a snapshot refetch. `sha_short`
    // is reused from the dedup gate above.
    let diff = crate::dashboard::approval_diff(agent, id).await;
    coord.emit_approval_added(
        id,
        agent,
        "apply_commit",
        Some(sha_short),
        Some(diff),
        description.map(str::to_owned),
    );
    Ok((id, sha))
}

/// Submit a `RequestSchedulePrompt` payload as an `ApprovalKind::SchedulePrompt`
/// row. Encodes the payload into the approval's `commit_ref` so the
/// approve handler can re-parse it without a side table. Validates
/// inputs (non-empty targets, non-empty body, sane interval) at
/// submit time — the operator should never see a malformed schedule
/// pending approval.
fn handle_request_schedule_prompt(
    coord: &Arc<Coordinator>,
    requester: &str,
    payload: &hive_sh4re::SchedulePromptPayload,
) -> ManagerResponse {
    if payload.targets.is_empty() {
        return ManagerResponse::Err {
            message: "schedule must have at least one target".into(),
        };
    }
    if payload.body.trim().is_empty() {
        return ManagerResponse::Err {
            message: "schedule body must be non-empty".into(),
        };
    }
    if let Some(0) = payload.interval_seconds {
        return ManagerResponse::Err {
            message: "interval_seconds must be > 0 (use None for one-shot)".into(),
        };
    }
    let commit_ref = match serde_json::to_string(payload) {
        Ok(s) => s,
        Err(e) => {
            return ManagerResponse::Err {
                message: format!("encode SchedulePromptPayload: {e:#}"),
            };
        }
    };
    let id = match coord.approvals.submit_kind(
        requester,
        hive_sh4re::ApprovalKind::SchedulePrompt,
        &commit_ref,
        payload.description.as_deref(),
    ) {
        Ok(id) => id,
        Err(e) => {
            return ManagerResponse::Err {
                message: format!("queue schedule_prompt approval: {e:#}"),
            };
        }
    };
    tracing::info!(
        %id,
        requester,
        targets = ?payload.targets,
        first_fire_at = payload.first_fire_at_unix,
        interval = ?payload.interval_seconds,
        "schedule_prompt approval queued"
    );
    coord.emit_approval_added(
        id,
        requester,
        "schedule_prompt",
        None,
        None,
        payload.description.clone(),
    );
    ManagerResponse::Ok
}

/// Cancel a schedule (whole or per-target). Manager-surface
/// authorization: a manager can cancel its own schedules + any
/// schedule whose owner is one of its sub-agents (topology-walked).
/// The operator surface bypasses this and can cancel anything;
/// agents reaching this path through the manager get the
/// topology-scoped check.
fn handle_cancel_schedule(
    coord: &Arc<Coordinator>,
    requester: &str,
    schedule_id: i64,
    targets: Option<&[String]>,
) -> ManagerResponse {
    let schedule = match coord.scheduled_prompts.get(schedule_id) {
        Ok(Some(s)) => s,
        Ok(None) => {
            return ManagerResponse::Err {
                message: format!("schedule {schedule_id} not found"),
            };
        }
        Err(e) => {
            return ManagerResponse::Err {
                message: format!("read schedule {schedule_id}: {e:#}"),
            };
        }
    };
    if !cancel_authorized(requester, &schedule.owner) {
        return ManagerResponse::Err {
            message: format!(
                "not authorized: {requester} cannot cancel schedule owned by {owner}",
                owner = schedule.owner
            ),
        };
    }
    let result = match targets {
        Some(list) if !list.is_empty() => coord
            .scheduled_prompts
            .cancel_targets(schedule_id, list)
            .map_err(|e| format!("cancel targets: {e:#}")),
        _ => coord
            .scheduled_prompts
            .cancel_all(schedule_id)
            .map_err(|e| format!("cancel all: {e:#}")),
    };
    match result {
        Ok(()) => ManagerResponse::Ok,
        Err(message) => ManagerResponse::Err { message },
    }
}

/// Authorize + dispatch a `FireScheduleNow` request from the
/// manager surface. Same ownership rules as `CancelSchedule`:
/// requester can fire its own schedules + any owned by an agent
/// in its subtree. The actual fan-out lives in
/// `scheduled_prompts_worker::fire_now`.
async fn handle_fire_schedule_now(
    coord: &Arc<Coordinator>,
    requester: &str,
    schedule_id: i64,
) -> ManagerResponse {
    let schedule = match coord.scheduled_prompts.get(schedule_id) {
        Ok(Some(s)) => s,
        Ok(None) => {
            return ManagerResponse::Err {
                message: format!("schedule {schedule_id} not found"),
            };
        }
        Err(e) => {
            return ManagerResponse::Err {
                message: format!("read schedule {schedule_id}: {e:#}"),
            };
        }
    };
    if !cancel_authorized(requester, &schedule.owner) {
        return ManagerResponse::Err {
            message: format!(
                "not authorized: {requester} cannot fire schedule owned by {owner}",
                owner = schedule.owner
            ),
        };
    }
    match crate::scheduled_prompts_worker::fire_now(coord, schedule_id).await {
        Ok(_report) => ManagerResponse::Ok,
        Err(e) => ManagerResponse::Err {
            message: format!("fire schedule {schedule_id} now: {e:#}"),
        },
    }
}

/// Authorize + dispatch a `EditSchedule` patch. Same ownership
/// rules as `CancelSchedule` — the manager can edit
/// schedules it owns + any owned by an agent in its subtree.
/// Forwards the partial payload to
/// `ScheduledPrompts::update` which enforces the cancelled-row /
/// zero-interval validation. Returns `Ok` on a clean update;
/// `Err` with the underlying message on any auth / validation
/// failure so the dashboard can surface it verbatim.
#[allow(clippy::too_many_arguments)]
#[allow(
    clippy::option_option,
    reason = "double-Option carries three-state PATCH semantics: outer None = \
              leave alone, Some(None) = clear, Some(Some(v)) = set"
)]
fn handle_edit_schedule(
    coord: &Arc<Coordinator>,
    requester: &str,
    schedule_id: i64,
    body: Option<String>,
    description: Option<Option<String>>,
    interval_seconds: Option<Option<u64>>,
    next_fire_at_unix: Option<i64>,
    targets_add: Option<Vec<String>>,
    targets_remove: Option<Vec<String>>,
) -> ManagerResponse {
    let schedule = match coord.scheduled_prompts.get(schedule_id) {
        Ok(Some(s)) => s,
        Ok(None) => {
            return ManagerResponse::Err {
                message: format!("schedule {schedule_id} not found"),
            };
        }
        Err(e) => {
            return ManagerResponse::Err {
                message: format!("read schedule {schedule_id}: {e:#}"),
            };
        }
    };
    if !cancel_authorized(requester, &schedule.owner) {
        return ManagerResponse::Err {
            message: format!(
                "not authorized: {requester} cannot edit schedule owned by {owner}",
                owner = schedule.owner
            ),
        };
    }
    let patch = crate::scheduled_prompts::UpdateSchedule {
        body,
        description,
        interval_seconds,
        next_fire_at_unix,
        targets_add,
        targets_remove,
    };
    match coord.scheduled_prompts.update(schedule_id, patch) {
        Ok(()) => ManagerResponse::Ok,
        Err(e) => ManagerResponse::Err {
            message: format!("edit schedule {schedule_id}: {e:#}"),
        },
    }
}

/// Permission check for `CancelSchedule` on the manager surface.
/// `requester` (always `hm1nd` here) can cancel its own schedules.
/// Sub-agent ownership is delegated to topology — see
/// `crate::topology::is_descendant_of`. Also reused by
/// `handle_fire_schedule_now` — fire-auth follows the same shape.
fn cancel_authorized(requester: &str, owner: &str) -> bool {
    if requester == owner {
        return true;
    }
    if requester == hive_sh4re::OPERATOR_RECIPIENT {
        return true;
    }
    // Manager can cancel anything owned by an agent in its subtree.
    // For the current single-manager topology that covers everything,
    // but the check stays correct as the tree grows.
    crate::topology::is_descendant_of(owner, requester)
}

/// Map a `scheduled_prompts::Schedule` to its public wire shape.
/// Field-by-field copy — the two types are intentionally identical;
/// the separation keeps hive-sh4re free of hive-c0re-internal types.
/// Public alias `schedule_to_wire_public` re-exports for
/// `dashboard.rs::api_schedules` without crossing the module
/// boundary into the manager-server file.
pub fn schedule_to_wire_public(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSchedule {
    schedule_to_wire(s)
}

fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSchedule {
    hive_sh4re::WireSchedule {
        id: s.id,
        owner: s.owner,
        body: s.body,
        interval_seconds: s.interval_seconds,
        next_fire_at_unix: s.next_fire_at_unix,
        created_at_unix: s.created_at_unix,
        source: match s.source {
            crate::scheduled_prompts::ScheduleSource::Operator => {
                hive_sh4re::WireScheduleSource::Operator
            }
            crate::scheduled_prompts::ScheduleSource::Approval { id } => {
                hive_sh4re::WireScheduleSource::Approval { id }
            }
        },
        cancelled_at_unix: s.cancelled_at_unix,
        description: s.description,
        targets: s
            .targets
            .into_iter()
            .map(|t| hive_sh4re::WireScheduleTarget {
                target: t.target,
                cancelled_at_unix: t.cancelled_at_unix,
                last_fired_at_unix: t.last_fired_at_unix,
                last_result: t.last_result,
            })
            .collect(),
    }
}

/// On `Ask { ttl_seconds: Some(n) }`, sleep n seconds and then try to
/// resolve the question with `[expired]`. If the operator (or any
/// other path) already answered it, `answer()` returns Err and we
/// no-op silently. Otherwise fire a `QuestionAnswered` helper event
/// with `answerer = "ttl-watchdog"` so the asker can distinguish a
/// real answer from a deadline trip without parsing the answer text.
const TTL_SENTINEL: &str = "[expired]";
/// Synthetic `answerer` label used when the ttl watchdog resolves a
/// question instead of a real human / agent. Lives in a distinct
/// namespace from agent names + the operator so the asker can pattern
/// match `event.answerer == "ttl-watchdog"`.
const TTL_ANSWERER: &str = "ttl-watchdog";

pub fn spawn_question_watchdog(coord: &Arc<Coordinator>, id: i64, ttl_secs: u64) {
    let coord = coord.clone();
    tokio::spawn(async move {
        tokio::time::sleep(std::time::Duration::from_secs(ttl_secs)).await;
        // Watchdog has its own answerer label so the authorisation
        // check in `answer()` permits it for any target. We bypass
        // the public `answer()` path by calling it with the operator
        // identity, since the operator is always permitted; the
        // event we fire carries the real watchdog label for observers.
        if let Ok((question, asker, target)) =
            coord
                .questions
                .answer(id, TTL_SENTINEL, hive_sh4re::OPERATOR_RECIPIENT)
        {
            tracing::info!(%id, %asker, "question expired (ttl)");
            coord.notify_agent(
                &asker,
                &hive_sh4re::HelperEvent::QuestionAnswered {
                    id,
                    question,
                    answer: TTL_SENTINEL.to_owned(),
                    answerer: TTL_ANSWERER.to_owned(),
                },
            );
            coord.emit_question_resolved(id, TTL_SENTINEL, TTL_ANSWERER, false, target.as_deref());
        }
    });
}

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

    #[test]
    fn accepts_short_and_full_sha() {
        assert!(validate_commit_ref("e194f78").is_ok());
        assert!(validate_commit_ref("e194f7812ab").is_ok());
        assert!(validate_commit_ref(&"a".repeat(40)).is_ok());
        // Uppercase hex resolves fine through `git rev-parse`.
        assert!(validate_commit_ref("E194F78").is_ok());
    }

    #[test]
    fn rejects_branch_and_tag_names() {
        // The exact bug class this guard exists for.
        assert!(validate_commit_ref("main").is_err());
        assert!(validate_commit_ref("HEAD").is_err());
        assert!(validate_commit_ref("deployed/0").is_err());
        assert!(validate_commit_ref("feature-branch").is_err());
    }

    #[test]
    fn rejects_too_short_too_long_and_empty() {
        assert!(validate_commit_ref("").is_err());
        assert!(validate_commit_ref("abc123").is_err()); // 6 chars
        assert!(validate_commit_ref(&"a".repeat(41)).is_err());
    }
}