refactor(hive-c0re): split socket_server into submodules

mod.rs keeps dispatch + messaging/guards; schedules, reminders,
config approvals, and lifecycle handlers move to their own files

hive-c0re/src/socket_server/config_approvals.rs

@@ -0,0 +1,343 @@
+//! Config-approval request handlers: `RequestInitConfig` /
+//! `RequestApplyCommit` / `RequestUpdateMetaInputs`, plus the shared
+//! submit helpers (`submit_init_config` / `submit_apply_commit`) and the
+//! commit-sha shape check (`validate_commit_ref`).
+
+use std::sync::Arc;
+
+use anyhow::{Context, Result};
+use hive_sh4re::AgentResponse;
+
+use super::require_new_child;
+use crate::coordinator::Coordinator;
+
+/// `RequestInitConfig` — queue an `InitConfig` approval for an agent. The
+/// `name` must be brand-new (absent from the topology) or already in the
+/// caller's subtree; the requester is recorded as the new agent's parent (the
+/// root requesting a new agent → a top-level agent, matching reconcile's
+/// default).
+pub(super) fn handle_request_init_config(
+    coord: &Arc<Coordinator>,
+    agent: &str,
+    name: &str,
+    description: Option<String>,
+) -> AgentResponse {
+    if let Some(err) = require_new_child(agent, name, "request_init_config for") {
+        return err;
+    }
+    tracing::info!(%agent, %name, "request_init_config");
+    match submit_init_config(coord, name, Some(agent), description) {
+        Ok(_id) => AgentResponse::Ok,
+        Err(e) => AgentResponse::Err {
+            message: format!("{e:#}"),
+        },
+    }
+}
+
+/// `RequestApplyCommit` — queue an apply-commit approval for an agent. The
+/// target must be in the caller's subtree (the root covers every agent).
+pub(super) async fn handle_request_apply_commit(
+    coord: &Arc<Coordinator>,
+    agent: &str,
+    target_agent: &str,
+    commit_ref: &str,
+    description: Option<&str>,
+) -> AgentResponse {
+    if let Some(err) = require_new_child(agent, target_agent, "request_apply_commit for") {
+        return err;
+    }
+    tracing::info!(%agent, %target_agent, %commit_ref, "request_apply_commit");
+    match submit_apply_commit(coord, target_agent, commit_ref, description, agent).await {
+        Ok((id, sha)) => {
+            tracing::info!(%id, %target_agent, %sha, "apply_commit approval queued");
+            AgentResponse::Ok
+        }
+        Err(e) => AgentResponse::Err {
+            message: format!("{e:#}"),
+        },
+    }
+}
+
+/// `RequestUpdateMetaInputs` — queue an `UpdateMetaInputs` approval
+/// carrying the JSON-encoded input list in `commit_ref` (no git commit
+/// is involved; the field is the payload the approval handler decodes).
+pub(super) fn handle_request_update_meta_inputs(
+    coord: &Arc<Coordinator>,
+    requester: &str,
+    inputs: &[String],
+    description: Option<&str>,
+) -> AgentResponse {
+    let label = if inputs.is_empty() {
+        "all inputs".to_string()
+    } else {
+        inputs.join(", ")
+    };
+    tracing::info!(%requester, %label, "request_update_meta_inputs");
+    let commit_ref = serde_json::to_string(inputs).unwrap_or_default();
+    let id = match coord
+        .approvals
+        .submit_kind(
+            requester,
+            hive_sh4re::ApprovalKind::UpdateMetaInputs,
+            &commit_ref,
+            description,
+            requester,
+        )
+        .map_err(|e| anyhow::anyhow!("{e:#}"))
+    {
+        Ok(id) => id,
+        Err(e) => {
+            return AgentResponse::Err {
+                message: format!("queue update_meta_inputs approval: {e:#}"),
+            };
+        }
+    };
+    tracing::info!(%id, %label, "update_meta_inputs approval queued");
+    coord.emit_approval_added(crate::coordinator::ApprovalAdded {
+        id,
+        agent: requester,
+        approval_kind: "update_meta_inputs",
+        sha_short: None,
+        diff: None,
+        description: description.map(str::to_owned),
+        pr_number: None,
+    });
+    AgentResponse::Ok
+}
+
+/// `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`.
+pub(crate) 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(())
+}
+
+/// Queue an `InitConfig` approval for a brand-new agent whose config repo
+/// does not yet exist. Shared between the manager and agent sockets.
+///
+/// `parent`, when `Some`, is the agent that will own the new child once
+/// the operator approves: it is stashed in the approval's `commit_ref`
+/// field (unused for `InitConfig` otherwise — same pattern
+/// `UpdateMetaInputs` uses to carry its inputs JSON) and consumed in
+/// `run_approval_init_config` to write the `child -> parent` topology
+/// edge. Callers pass the requesting agent, so the requester becomes the
+/// new agent's parent (the root requesting a new agent → a top-level agent,
+/// matching `topology::reconcile`'s default). `None` writes no explicit
+/// edge (reconcile-default placement) — retained for that fallback.
+pub(crate) fn submit_init_config(
+    coord: &Arc<Coordinator>,
+    name: &str,
+    parent: Option<&str>,
+    description: Option<String>,
+) -> anyhow::Result<i64> {
+    let proposed_dir = crate::coordinator::Coordinator::agent_proposed_dir(name);
+    if proposed_dir.join(".git").exists() {
+        anyhow::bail!(
+            "proposed config repo for '{name}' already exists at {} - \
+             use request_apply_commit to update an existing agent's config",
+            proposed_dir.display()
+        );
+    }
+    let id = coord
+        .approvals
+        .submit_kind(
+            name,
+            hive_sh4re::ApprovalKind::InitConfig,
+            parent.unwrap_or(""),
+            description.as_deref(),
+            // `parent` is the requesting agent (becomes the new child's
+            // parent); it's also the submitter the approval events route
+            // back to. No declared parent = operator-initiated path.
+            parent.unwrap_or("operator"),
+        )
+        .map_err(|e| anyhow::anyhow!("queue approval row: {e:#}"))?;
+    tracing::info!(%id, %name, "init_config approval queued");
+    coord.emit_approval_added(crate::coordinator::ApprovalAdded {
+        id,
+        agent: name,
+        approval_kind: "init_config",
+        sha_short: None,
+        diff: None,
+        description,
+        pr_number: None,
+    });
+    Ok(id)
+}
+
+/// 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.
+pub(crate) async fn submit_apply_commit(
+    coord: &Arc<Coordinator>,
+    agent: &str,
+    commit_ref: &str,
+    description: Option<&str>,
+    submitter: &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.
+        crate::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,
+            submitter,
+        )
+        .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(crate::coordinator::ApprovalResolved {
+                    id,
+                    agent,
+                    approval_kind: "apply_commit",
+                    sha_short: None,
+                    status: "failed",
+                    note: Some(note),
+                    description: 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(crate::coordinator::ApprovalResolved {
+            id,
+            agent,
+            approval_kind: "apply_commit",
+            sha_short: Some(sha_short.clone()),
+            status: "failed",
+            note: Some(note),
+            description: 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(crate::coordinator::ApprovalResolved {
+            id,
+            agent,
+            approval_kind: "apply_commit",
+            sha_short: Some(sha_short.clone()),
+            status: "failed",
+            note: Some(note),
+            description: 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(crate::coordinator::ApprovalAdded {
+        id,
+        agent,
+        approval_kind: "apply_commit",
+        sha_short: Some(sha_short),
+        diff: Some(diff),
+        description: description.map(str::to_owned),
+        pr_number: None,
+    });
+    Ok((id, sha))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[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());
+    }
+}

hive-c0re/src/socket_server/lifecycle_handlers.rs

@@ -0,0 +1,201 @@
+//! Container-lifecycle request handlers (`Start` / `Restart` / `Kill` /
+//! `Update` / `ListDescendants`), including the capability-gated
+//! infra-container restart path. All are topology-guarded via
+//! `super::require_descendant`.
+
+use std::sync::Arc;
+
+use hive_sh4re::AgentResponse;
+
+use super::require_descendant;
+use crate::coordinator::Coordinator;
+
+/// `Start` — start a container, kicking its next turn. The caller must be an
+/// ancestor of `name` in the topology (the root covers every agent).
+pub(super) fn handle_start(coord: &Arc<Coordinator>, agent: &str, name: &str) -> AgentResponse {
+    if let Some(err) = require_descendant(agent, name, "start") {
+        return err;
+    }
+    tracing::info!(%agent, %name, "start container");
+    // Persist `wanted = Up` and submit the Start DAG; the submit layer
+    // upgrades a stale-rev start to a full rebuild so the container
+    // runs current nix derivations before it starts.
+    crate::job_queue::submit::start(
+        coord,
+        name,
+        crate::job_queue::Source::Manual,
+        format!("agent `{agent}` start tool"),
+    );
+    AgentResponse::Ok
+}
+
+/// `Restart` — enqueue a restart for a container. The caller must be an
+/// ancestor of `name` in the topology. The infra-container branch is
+/// orthogonal: it is gated on the `infra_admin` capability and audited, so it
+/// stays ahead of the topology guard.
+pub(super) async fn handle_restart(
+    coord: &Arc<Coordinator>,
+    agent: &str,
+    name: &str,
+) -> AgentResponse {
+    // Infra-container restart: an agent holding the `infra_admin`
+    // capability can restart a hive infrastructure container (hive-ci /
+    // hive-gateway / hive-forge / hive-matrix) by passing its name to the
+    // same restart tool. The `InfraContainer` enum parse both recognises
+    // these (never agent children, so disjoint from the child path below)
+    // and yields the typed value the restart path needs.
+    if let Ok(container) = name.parse::<hive_sh4re::priv_proto::InfraContainer>() {
+        return handle_restart_infra(coord, agent, container).await;
+    }
+    if let Some(err) = require_descendant(agent, name, "restart") {
+        return err;
+    }
+    tracing::info!(%agent, %name, "submit restart");
+    crate::job_queue::submit::restart(
+        coord,
+        name,
+        crate::job_queue::Source::Manual,
+        format!("agent `{agent}` restart tool"),
+    );
+    AgentResponse::Ok
+}
+
+/// Restart a hive infrastructure container on behalf of an agent that
+/// holds the `infra_admin` capability. The `container` is already a valid
+/// [`InfraContainer`] (the caller parsed it); this gates on the capability
+/// and routes the systemctl restart through hive-priv. Direct, not
+/// approval-gated.
+async fn handle_restart_infra(
+    coord: &Arc<Coordinator>,
+    agent: &str,
+    container: hive_sh4re::priv_proto::InfraContainer,
+) -> AgentResponse {
+    let name = container.unit_name();
+    // Record the attempt in the operator-visible privileged-action audit
+    // trail, then emit a live `AuditEntryAdded` so the dashboard audit view
+    // appends it off `/dashboard/stream`. Best-effort: `record` returns the
+    // canonical row (or `None` on a sqlite blip), and we stream exactly that
+    // row so the stored + streamed views can't drift. `action` is stable so
+    // the dashboard can group/filter.
+    let audit = |outcome: crate::audit_log::AuditOutcome, detail: Option<&str>| {
+        if let Some(entry) = coord
+            .audit_log
+            .record(agent, "restart_infra", name, outcome, detail)
+        {
+            coord.emit_audit_entry(entry);
+        }
+    };
+    if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::InfraAdmin) {
+        tracing::warn!(%agent, %name, "agent: infra restart denied (no infra_admin capability)");
+        audit(
+            crate::audit_log::AuditOutcome::Err,
+            Some("denied: missing infra_admin capability"),
+        );
+        return AgentResponse::Err {
+            message: format!(
+                "restarting infra container `{name}` requires the `infra_admin` capability"
+            ),
+        };
+    }
+    tracing::info!(%agent, %name, "agent: restart infra container");
+    match crate::priv_client::restart_infra_container(container).await {
+        Ok(()) => {
+            audit(crate::audit_log::AuditOutcome::Ok, None);
+            AgentResponse::Ok
+        }
+        Err(e) => {
+            let msg = format!("{e:#}");
+            audit(crate::audit_log::AuditOutcome::Err, Some(&msg));
+            AgentResponse::Err { message: msg }
+        }
+    }
+}
+
+/// `Kill` — kill a container, unregister it, notify the manager. The caller
+/// must be an ancestor of `name` in the topology.
+pub(super) async fn handle_kill(
+    coord: &Arc<Coordinator>,
+    agent: &str,
+    name: &str,
+) -> AgentResponse {
+    if let Some(err) = require_descendant(agent, name, "kill") {
+        return err;
+    }
+    tracing::info!(%agent, %name, "kill container");
+    // Persist the intent even if the kill fails — otherwise the next
+    // reconcile would restart the container.
+    if let Err(e) = coord.power.set(name, crate::power::Wanted::Offline) {
+        tracing::warn!(%name, error = ?e, "agent_power: set wanted=offline failed");
+    }
+    let result: anyhow::Result<()> = async {
+        crate::lifecycle::kill(name).await?;
+        coord.unregister_agent(name);
+        Ok(())
+    }
+    .await;
+    match result {
+        Ok(()) => {
+            coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
+                agent: name.to_owned(),
+            });
+            AgentResponse::Ok
+        }
+        Err(e) => AgentResponse::Err {
+            message: format!("{e:#}"),
+        },
+    }
+}
+
+/// `Update` — enqueue a rebuild for a container. The caller must be an
+/// ancestor of `name` in the topology.
+pub(super) fn handle_update(coord: &Arc<Coordinator>, agent: &str, name: &str) -> AgentResponse {
+    if let Some(err) = require_descendant(agent, name, "rebuild") {
+        return err;
+    }
+    tracing::info!(%agent, %name, "submit rebuild");
+    crate::job_queue::submit::rebuild(
+        coord,
+        name,
+        crate::job_queue::Source::Manual,
+        format!("agent `{agent}` update tool"),
+    );
+    AgentResponse::Ok
+}
+
+/// `ListDescendants` — every topological descendant of `agent` with
+/// its running/stopped state, parents before children.
+pub(super) async fn handle_list_descendants(agent: &str) -> AgentResponse {
+    tracing::debug!(%agent, "agent: list descendants");
+    // All containers known to nixos-container (running only).
+    let running_set: std::collections::HashSet<String> = match crate::lifecycle::list().await {
+        Ok(names) => names
+            .into_iter()
+            .filter_map(|c| {
+                c.strip_prefix(crate::lifecycle::AGENT_PREFIX)
+                    .map(str::to_owned)
+            })
+            .collect(),
+        Err(e) => {
+            return AgentResponse::Err {
+                message: format!("list containers failed: {e:#}"),
+            };
+        }
+    };
+    // Walk the full topology and collect every descendant.
+    let topo = crate::topology::read();
+    let mut names: Vec<String> = topo
+        .keys()
+        .filter(|name| crate::topology::is_descendant_of(name, agent))
+        .cloned()
+        .collect();
+    // Parents before children, then alpha within each tier.
+    crate::auto_update::topology_sort(&mut names, &topo);
+    let containers = names
+        .into_iter()
+        .map(|name| {
+            let running = running_set.contains(&name);
+            hive_sh4re::ContainerInfo { name, running }
+        })
+        .collect();
+    AgentResponse::Containers { containers }
+}

hive-c0re/src/socket_server/reminders.rs

@@ -0,0 +1,229 @@
+//! Reminder request handling: the `Remind` handler, the shared
+//! `store_remind` storage path with its pending-cap and large-body
+//! auto-save dance, timing resolution, and the agent-state target
+//! resolution shared by the loose-ends / reminder query handlers.
+
+use std::sync::Arc;
+
+use hive_sh4re::AgentResponse;
+
+use crate::coordinator::Coordinator;
+
+pub(super) fn handle_remind(
+    coord: &Arc<Coordinator>,
+    agent: &str,
+    message: &str,
+    timing: &hive_sh4re::ReminderTiming,
+    file_path: Option<&str>,
+) -> AgentResponse {
+    match store_remind(coord, agent, message, timing, file_path) {
+        Ok(()) => AgentResponse::Ok,
+        Err(message) => AgentResponse::Err { message },
+    }
+}
+
+/// Shared remind-storage path used by both the agent and the manager
+/// dispatchers. Validates timing, applies the auto-file overflow
+/// dance (see [`prepare_remind_storage`]), and writes the reminder
+/// row. Returns `Ok(())` on success, or a caller-ready error string
+/// the dispatcher wraps in `*Response::Err`.
+/// Maximum pending (un-delivered) reminders per agent. Exceeding this
+/// causes `store_remind` to return an error so the agent knows to back
+/// off instead of silently dropping. Override via
+/// `HIVE_REMIND_MAX_PENDING_PER_AGENT`; set to `0` to disable the cap
+/// (not recommended — a runaway agent can still flood the scheduler).
+const DEFAULT_REMIND_MAX_PENDING: u64 = 50;
+
+fn remind_max_pending() -> u64 {
+    std::env::var("HIVE_REMIND_MAX_PENDING_PER_AGENT")
+        .ok()
+        .and_then(|s| s.trim().parse::<u64>().ok())
+        .unwrap_or(DEFAULT_REMIND_MAX_PENDING)
+}
+
+pub(crate) fn store_remind(
+    coord: &Arc<Coordinator>,
+    agent: &str,
+    message: &str,
+    timing: &hive_sh4re::ReminderTiming,
+    file_path: Option<&str>,
+) -> Result<(), String> {
+    let max = remind_max_pending();
+    if max > 0 {
+        let pending = coord.broker.count_pending_reminders_for(agent).unwrap_or(0);
+        if pending >= max {
+            return Err(format!(
+                "reminder rejected: agent `{agent}` already has {pending} pending \
+                 reminders (cap {max}). Cancel some via `cancel_loose_end` or wait \
+                 for them to fire before scheduling more. Override the cap with \
+                 `HIVE_REMIND_MAX_PENDING_PER_AGENT`."
+            ));
+        }
+    }
+    let due_at = resolve_due_at(timing).map_err(|e| format!("invalid reminder timing: {e:#}"))?;
+    let (stored_message, stored_path) = prepare_remind_storage(agent, message, file_path)?;
+    let id = coord
+        .broker
+        .store_reminder(agent, &stored_message, stored_path.as_deref(), due_at)
+        .map_err(|e| format!("failed to store reminder: {e:#}"))?;
+    tracing::info!(%id, %agent, %due_at, "reminder scheduled");
+    coord.emit_reminders_snapshot();
+    Ok(())
+}
+
+/// Decide what we actually store in the reminders row, applying the
+/// same byte cap as the rest of the wire protocol
+/// ([`crate::limits::MESSAGE_MAX_BYTES`]). Three outcomes:
+///
+/// 1. Body within the cap → stored verbatim, with whatever `file_path`
+///    the caller passed (None or Some). The scheduler honours
+///    `file_path` at delivery time as before.
+/// 2. Body over the cap, no caller `file_path` → auto-generate a path
+///    under `/agents/<agent>/state/reminders/auto-<ts>.md`, write the
+///    body to disk now, store a short pointer hint as the message and
+///    clear `file_path` (so the scheduler doesn't re-write at
+///    delivery and overwrite the body with the hint).
+/// 3. Body over the cap, caller provided `file_path` → honour the
+///    caller's path: write the body to it now, store the same hint
+///    and clear `file_path` for the same reason as (2).
+///
+/// Returns `(stored_message, stored_file_path)` on success, or a
+/// caller-ready error string on auto-save failure (which is the only
+/// way a Remind request can be refused for size — the agent never has
+/// to think about the cap).
+fn prepare_remind_storage(
+    agent: &str,
+    message: &str,
+    file_path: Option<&str>,
+) -> Result<(String, Option<String>), String> {
+    if message.len() <= crate::limits::MESSAGE_MAX_BYTES {
+        return Ok((message.to_owned(), file_path.map(str::to_owned)));
+    }
+    let req_path = match file_path {
+        Some(p) => p.to_owned(),
+        None => auto_reminder_path(agent),
+    };
+    let host_path = crate::reminder_scheduler::resolve_host_path(agent, &req_path)
+        .map_err(|reason| format!("auto-save path `{req_path}` rejected: {reason}"))?;
+    crate::reminder_scheduler::write_payload(agent, &host_path, message).map_err(|reason| {
+        format!("auto-save of large reminder body to `{req_path}` failed: {reason}")
+    })?;
+    let hint = format!(
+        "[reminder body of {} bytes auto-saved to `{req_path}`; read with your filesystem tools]",
+        message.len()
+    );
+    Ok((hint, None))
+}
+
+/// Generate a per-agent path for an auto-saved reminder body. Uses
+/// `unix_nanos` plus the agent name to keep collisions infinitesimal
+/// across the agent's own state subtree (we're not stamping a hostname
+/// since hive-c0re is single-host).
+fn auto_reminder_path(agent: &str) -> String {
+    let ts_ns = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map_or(0, |d| d.as_nanos());
+    format!("/agents/{agent}/state/reminders/auto-{ts_ns}.md")
+}
+
+/// Resolve the target agent name for a *named* `GetLooseEnds` /
+/// `CountPendingReminders` / `ReminderRollup` query. Rules:
+///
+/// - `None` (or `Some(caller)`) → the caller's own threads (always allowed).
+/// - `Some(descendant)` in the caller's subtree (the root's subtree is the whole hive) → allowed, no extra capability.
+/// - `Some(other)` outside the subtree → requires the `query_agent_state` capability; error otherwise.
+/// - `Some("*")` → rejected here; the hive-wide sweep is handled by `handle_get_loose_ends` under the same `query_agent_state` gate.
+pub(super) fn resolve_agent_state_target<'a>(
+    caller: &'a str,
+    target: Option<&'a str>,
+) -> Result<&'a str, String> {
+    match target {
+        None => Ok(caller),
+        Some("*") => Err(
+            "hive-wide query (agent=\"*\") is only valid for loose-ends; \
+             not available for this query"
+                .to_owned(),
+        ),
+        Some(name) => {
+            // Own subtree (the root covers all) is visible without extra
+            // capability; `is_descendant_of` returns true for `name == caller`.
+            if crate::topology::is_descendant_of(name, caller) {
+                return Ok(name);
+            }
+            if crate::capabilities::has_cap(caller, hive_sh4re::Capability::QueryAgentState) {
+                Ok(name)
+            } else {
+                Err(format!(
+                    "agent `{caller}` cannot query `{name}`: not in its subtree and \
+                     `query_agent_state` capability is not granted"
+                ))
+            }
+        }
+    }
+}
+
+/// Resolve the `due_at` unix timestamp for a Remind request. Returns
+/// distinct error messages for each failure mode (overflow on
+/// `InSeconds`, pre-epoch clock, `i64` cast wrap) so the caller can tell
+/// what went wrong without inspecting the chain.
+fn resolve_due_at(timing: &hive_sh4re::ReminderTiming) -> anyhow::Result<i64> {
+    use hive_sh4re::ReminderTiming;
+    match timing {
+        ReminderTiming::InSeconds { seconds } => {
+            let now = std::time::SystemTime::now();
+            let future = now
+                .checked_add(std::time::Duration::from_secs(*seconds))
+                .ok_or_else(|| {
+                    anyhow::anyhow!("InSeconds overflow: {seconds}s exceeds system time range")
+                })?;
+            let duration = future
+                .duration_since(std::time::UNIX_EPOCH)
+                .map_err(|e| anyhow::anyhow!("system time before UNIX_EPOCH: {e}"))?;
+            i64::try_from(duration.as_secs())
+                .map_err(|e| anyhow::anyhow!("unix timestamp exceeds i64 range: {e}"))
+        }
+        ReminderTiming::At { unix_timestamp } => Ok(*unix_timestamp),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn auto_reminder_path_format() {
+        let p = auto_reminder_path("damocles");
+        assert!(p.starts_with("/agents/damocles/state/reminders/auto-"));
+        assert!(
+            std::path::Path::new(&p)
+                .extension()
+                .is_some_and(|ext| ext.eq_ignore_ascii_case("md"))
+        );
+    }
+
+    #[test]
+    fn prepare_remind_storage_passthrough_under_cap() {
+        let (msg, fp) = prepare_remind_storage("foo", "small body", None).unwrap();
+        assert_eq!(msg, "small body");
+        assert_eq!(fp, None);
+    }
+
+    #[test]
+    fn prepare_remind_storage_passthrough_with_caller_file_path() {
+        let (msg, fp) =
+            prepare_remind_storage("foo", "small", Some("/agents/foo/state/x.md")).unwrap();
+        assert_eq!(msg, "small");
+        assert_eq!(fp.as_deref(), Some("/agents/foo/state/x.md"));
+    }
+
+    #[test]
+    fn resolve_agent_state_target_self_and_default_are_free() {
+        // No topology/capability state needed for these: `None` and the
+        // caller's own name resolve to the caller (`is_descendant_of` short-
+        // circuits to true when candidate == ancestor); `"*"` is rejected
+        // (the hive-wide sweep is handled by the loose-ends caller instead).
+        assert_eq!(resolve_agent_state_target("iris", None), Ok("iris"));
+        assert_eq!(resolve_agent_state_target("iris", Some("iris")), Ok("iris"));
+        assert!(resolve_agent_state_target("iris", Some("*")).is_err());
+    }
+}

hive-c0re/src/socket_server/schedules.rs

@@ -0,0 +1,404 @@
+//! Scheduled-prompt request handlers (`ListSchedules` /
+//! `RequestSchedulePrompt` / `CancelSchedule` / `EditSchedule` /
+//! `FireScheduleNow`), their shared ownership check, and the
+//! schedule-to-wire mapping reused by the dashboard
+//! (`schedule_to_wire_public` / `filter_ghost_schedule_targets`).
+
+use std::sync::Arc;
+
+use hive_sh4re::AgentResponse;
+
+use crate::coordinator::Coordinator;
+
+/// `ListSchedules` — snapshot every scheduled prompt onto the wire.
+pub(super) fn handle_list_schedules(coord: &Arc<Coordinator>) -> AgentResponse {
+    match coord.scheduled_prompts.list() {
+        Ok(schedules) => AgentResponse::Schedules {
+            schedules: schedules.into_iter().map(schedule_to_wire).collect(),
+        },
+        Err(e) => AgentResponse::Err {
+            message: format!("list scheduled prompts: {e:#}"),
+        },
+    }
+}
+
+/// 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.
+pub(super) fn handle_request_schedule_prompt(
+    coord: &Arc<Coordinator>,
+    requester: &str,
+    payload: &hive_sh4re::SchedulePromptPayload,
+) -> AgentResponse {
+    if payload.targets.is_empty() {
+        return AgentResponse::Err {
+            message: "schedule must have at least one target".into(),
+        };
+    }
+    if payload.body.trim().is_empty() {
+        return AgentResponse::Err {
+            message: "schedule body must be non-empty".into(),
+        };
+    }
+    if let Some(0) = payload.interval_seconds {
+        return AgentResponse::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 AgentResponse::Err {
+                message: format!("encode SchedulePromptPayload: {e:#}"),
+            };
+        }
+    };
+    let id = match coord.approvals.submit_kind(
+        requester,
+        hive_sh4re::ApprovalKind::SchedulePrompt,
+        &commit_ref,
+        payload.description.as_deref(),
+        requester,
+    ) {
+        Ok(id) => id,
+        Err(e) => {
+            return AgentResponse::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(crate::coordinator::ApprovalAdded {
+        id,
+        agent: requester,
+        approval_kind: "schedule_prompt",
+        sha_short: None,
+        diff: None,
+        description: payload.description.clone(),
+        pr_number: None,
+    });
+    AgentResponse::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.
+pub(super) fn handle_cancel_schedule(
+    coord: &Arc<Coordinator>,
+    requester: &str,
+    schedule_id: i64,
+    targets: Option<&[String]>,
+) -> AgentResponse {
+    let schedule = match coord.scheduled_prompts.get(schedule_id) {
+        Ok(Some(s)) => s,
+        Ok(None) => {
+            return AgentResponse::Err {
+                message: format!("schedule {schedule_id} not found"),
+            };
+        }
+        Err(e) => {
+            return AgentResponse::Err {
+                message: format!("read schedule {schedule_id}: {e:#}"),
+            };
+        }
+    };
+    if !cancel_authorized(requester, &schedule.owner) {
+        return AgentResponse::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(()) => {
+            coord.emit_schedules_snapshot();
+            AgentResponse::Ok
+        }
+        Err(message) => AgentResponse::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`.
+pub(super) async fn handle_fire_schedule_now(
+    coord: &Arc<Coordinator>,
+    requester: &str,
+    schedule_id: i64,
+) -> AgentResponse {
+    let schedule = match coord.scheduled_prompts.get(schedule_id) {
+        Ok(Some(s)) => s,
+        Ok(None) => {
+            return AgentResponse::Err {
+                message: format!("schedule {schedule_id} not found"),
+            };
+        }
+        Err(e) => {
+            return AgentResponse::Err {
+                message: format!("read schedule {schedule_id}: {e:#}"),
+            };
+        }
+    };
+    if !cancel_authorized(requester, &schedule.owner) {
+        return AgentResponse::Err {
+            message: format!(
+                "not authorized: {requester} cannot fire schedule owned by {owner}",
+                owner = schedule.owner
+            ),
+        };
+    }
+    // MCP fire_schedule_now stays no-reset (cadence intact); the
+    // reset-timer option is a dashboard-dialog affordance.
+    match crate::scheduled_prompts_worker::fire_now(coord, schedule_id, false).await {
+        Ok(_report) => {
+            coord.emit_schedules_snapshot();
+            AgentResponse::Ok
+        }
+        Err(e) => AgentResponse::Err {
+            message: format!("fire schedule {schedule_id} now: {e:#}"),
+        },
+    }
+}
+
+/// Field-named PATCH payload for [`handle_edit_schedule`]. Every
+/// field is "leave alone" when `None`; the double-`Option` fields
+/// additionally distinguish clear (`Some(None)`) from set
+/// (`Some(Some(v))`).
+#[allow(
+    clippy::option_option,
+    reason = "double-Option carries three-state PATCH semantics: outer None = \
+              leave alone, Some(None) = clear, Some(Some(v)) = set"
+)]
+pub(super) struct EditSchedulePatch {
+    pub(super) body: Option<String>,
+    pub(super) description: Option<Option<String>>,
+    pub(super) interval_seconds: Option<Option<u64>>,
+    pub(super) next_fire_at_unix: Option<i64>,
+    pub(super) targets_add: Option<Vec<String>>,
+    pub(super) targets_remove: Option<Vec<String>>,
+}
+
+/// 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.
+pub(super) fn handle_edit_schedule(
+    coord: &Arc<Coordinator>,
+    requester: &str,
+    schedule_id: i64,
+    patch: EditSchedulePatch,
+) -> AgentResponse {
+    let EditSchedulePatch {
+        body,
+        description,
+        interval_seconds,
+        next_fire_at_unix,
+        targets_add,
+        targets_remove,
+    } = patch;
+    let schedule = match coord.scheduled_prompts.get(schedule_id) {
+        Ok(Some(s)) => s,
+        Ok(None) => {
+            return AgentResponse::Err {
+                message: format!("schedule {schedule_id} not found"),
+            };
+        }
+        Err(e) => {
+            return AgentResponse::Err {
+                message: format!("read schedule {schedule_id}: {e:#}"),
+            };
+        }
+    };
+    if !cancel_authorized(requester, &schedule.owner) {
+        return AgentResponse::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(()) => {
+            coord.emit_schedules_snapshot();
+            AgentResponse::Ok
+        }
+        Err(e) => AgentResponse::Err {
+            message: format!("edit schedule {schedule_id}: {e:#}"),
+        },
+    }
+}
+
+/// Permission check for `CancelSchedule` on the manager surface.
+/// `requester` (always `ruth` 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 socket-server file.
+pub fn schedule_to_wire_public(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSchedule {
+    schedule_to_wire(s)
+}
+
+/// Drop schedule targets that point at agents which no longer exist, so
+/// the dashboard's schedule table doesn't render ghost columns for
+/// destroyed agents. `live` is the set of logical agent names from the
+/// last `nixos-container list` scan (stopped agents included, destroyed
+/// ones absent); the `operator` pseudo-target is always retained since
+/// it isn't a container. Applied only to the dashboard wire paths
+/// (`api_schedules` + the `SchedulesChanged` SSE emit) — the
+/// manager-facing `list_schedules` stays unfiltered so agents can still
+/// see and cancel stale targets. This is a view filter: the underlying
+/// schedule rows keep every target, so a re-spawned agent's targets
+/// reappear on their own.
+pub(crate) fn filter_ghost_schedule_targets(
+    schedules: &mut [hive_sh4re::WireSchedule],
+    live: &std::collections::HashSet<String>,
+) {
+    for s in schedules.iter_mut() {
+        s.targets
+            .retain(|t| t.target == hive_sh4re::OPERATOR_RECIPIENT || live.contains(&t.target));
+    }
+}
+
+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: hive_sh4re::wire_time::from_secs(s.next_fire_at_unix),
+        created_at_unix: hive_sh4re::wire_time::from_secs(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.map(hive_sh4re::wire_time::from_secs),
+        paused_at_unix: s.paused_at_unix.map(hive_sh4re::wire_time::from_secs),
+        description: s.description,
+        targets: s
+            .targets
+            .into_iter()
+            .map(|t| hive_sh4re::WireScheduleTarget {
+                target: t.target,
+                cancelled_at_unix: t.cancelled_at_unix.map(hive_sh4re::wire_time::from_secs),
+                last_fired_at_unix: t.last_fired_at_unix.map(hive_sh4re::wire_time::from_secs),
+                last_result: t.last_result,
+            })
+            .collect(),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn target(name: &str) -> hive_sh4re::WireScheduleTarget {
+        hive_sh4re::WireScheduleTarget {
+            target: name.to_owned(),
+            cancelled_at_unix: None,
+            last_fired_at_unix: None,
+            last_result: None,
+        }
+    }
+
+    fn schedule(targets: &[&str]) -> hive_sh4re::WireSchedule {
+        hive_sh4re::WireSchedule {
+            id: 1,
+            owner: "operator".to_owned(),
+            body: "ping".to_owned(),
+            interval_seconds: None,
+            next_fire_at_unix: hive_sh4re::wire_time::from_secs(0),
+            created_at_unix: hive_sh4re::wire_time::from_secs(0),
+            source: hive_sh4re::WireScheduleSource::Operator,
+            cancelled_at_unix: None,
+            paused_at_unix: None,
+            description: None,
+            targets: targets.iter().map(|t| target(t)).collect(),
+        }
+    }
+
+    #[test]
+    fn ghost_filter_drops_dead_agents_keeps_live_and_operator() {
+        let live: std::collections::HashSet<String> = ["iris".to_owned(), "damocles".to_owned()]
+            .into_iter()
+            .collect();
+        let mut schedules = vec![schedule(&["iris", "ghost", "operator", "damocles"])];
+        filter_ghost_schedule_targets(&mut schedules, &live);
+        let kept: Vec<&str> = schedules[0]
+            .targets
+            .iter()
+            .map(|t| t.target.as_str())
+            .collect();
+        // `ghost` (destroyed) dropped; live agents + operator pseudo-target kept.
+        assert_eq!(kept, vec!["iris", "operator", "damocles"]);
+    }
+
+    #[test]
+    fn ghost_filter_can_empty_targets_when_all_dead() {
+        let live: std::collections::HashSet<String> = std::collections::HashSet::new();
+        let mut schedules = vec![schedule(&["gone1", "gone2"])];
+        filter_ghost_schedule_targets(&mut schedules, &live);
+        // operator is never in the live set but is always retained; here
+        // there's no operator target, so everything drops.
+        assert!(schedules[0].targets.is_empty());
+    }
+}