diff --git a/hive-c0re/src/container_view.rs b/hive-c0re/src/container_view.rs
index c0719ca7..28821cda 100644
--- a/hive-c0re/src/container_view.rs
+++ b/hive-c0re/src/container_view.rs
@@ -137,7 +137,7 @@ fn auth_failed_sentinel(name: &str) -> bool {
 
 /// Read the agent's free-text status and the Unix timestamp when it was last set
 /// (derived from the file's mtime). Returns `(None, None)` when the file is absent
-/// or empty. `pub` so `socket_server` and `manager_server` can populate `AgentMeta`.
+/// or empty. `pub` so `socket_server` and `socket_server` can populate `AgentMeta`.
 ///
 /// NB: callers building `AgentMeta` for a *stopped* container should
 /// clear the result — the on-disk status is a stale snapshot from
diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs
index 039a126d..383e2a23 100644
--- a/hive-c0re/src/coordinator.rs
+++ b/hive-c0re/src/coordinator.rs
@@ -522,7 +522,7 @@ impl Coordinator {
         let mut schedules: Vec<hive_sh4re::WireSchedule> = match self.scheduled_prompts.list() {
             Ok(rows) => rows
                 .into_iter()
-                .map(crate::manager_server::schedule_to_wire_public)
+                .map(crate::socket_server::schedule_to_wire_public)
                 .collect(),
             Err(e) => {
                 tracing::warn!(error = ?e, "emit_schedules_snapshot: list failed");
@@ -534,7 +534,7 @@ impl Coordinator {
         // momentarily contended we emit unfiltered rather than block this
         // sync path — the next snapshot / page reload corrects it.
         if let Some(live) = self.live_container_names_blocking() {
-            crate::manager_server::filter_ghost_schedule_targets(&mut schedules, &live);
+            crate::socket_server::filter_ghost_schedule_targets(&mut schedules, &live);
         }
         self.emit_dashboard_event(DashboardEvent::SchedulesChanged {
             seq: self.next_seq(),
@@ -1330,7 +1330,7 @@ impl Coordinator {
     }
 
     /// Ensure a runtime dir + (for sub-agents) per-agent socket exists. For
-    /// the manager, `manager_server::start` owns the socket — just return
+    /// the manager, `socket_server::start_manager` owns the socket — just return
     /// the dir. For sub-agents this is `register_agent` (creates a fresh
     /// listener bound to `socket_path(name)`). Source directory of the
     /// `/run/hive/mcp.sock` bind that ends up in `set_nspawn_flags`.
diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs
index b19855cd..4b4d7507 100644
--- a/hive-c0re/src/dashboard.rs
+++ b/hive-c0re/src/dashboard.rs
@@ -42,7 +42,7 @@ mod topology;
 mod webhook;
 
 // Pre-computed at approval-submit time by the manager-socket handler
-// (`manager_server.rs`) and embedded in the `ApprovalAdded` event, so
+// (`socket_server.rs`) and embedded in the `ApprovalAdded` event, so
 // re-exported at the module root to preserve the `crate::dashboard::approval_diff`
 // path across the submodule split.
 pub(crate) use approvals::approval_diff;
diff --git a/hive-c0re/src/dashboard/lifecycle_ops.rs b/hive-c0re/src/dashboard/lifecycle_ops.rs
index e11aa503..b8b8b3a4 100644
--- a/hive-c0re/src/dashboard/lifecycle_ops.rs
+++ b/hive-c0re/src/dashboard/lifecycle_ops.rs
@@ -76,7 +76,7 @@ pub(super) async fn post_kill(
     // host-side approval queue without the manager up, and
     // operator-driven meta-input updates work from the dashboard
     // either way. The MCP-surface self-kill guard in
-    // `manager_server.rs::ManagerRequest::Kill` stays in place: a
+    // `socket_server.rs::ManagerRequest::Kill` stays in place: a
     // manager calling Kill on its own container is self-suicide
     // mid-call, not a legitimate operator action.
     state.coord.rebuild_queue.enqueue(
diff --git a/hive-c0re/src/dashboard/schedules.rs b/hive-c0re/src/dashboard/schedules.rs
index b22f92d3..535ae9c0 100644
--- a/hive-c0re/src/dashboard/schedules.rs
+++ b/hive-c0re/src/dashboard/schedules.rs
@@ -21,7 +21,7 @@ pub(super) async fn api_schedules(State(state): State<AppState>) -> Response {
         Ok(rows) => {
             let mut wire: Vec<hive_sh4re::WireSchedule> = rows
                 .into_iter()
-                .map(crate::manager_server::schedule_to_wire_public)
+                .map(crate::socket_server::schedule_to_wire_public)
                 .collect();
             // Drop ghost targets (agents that no longer exist) so the
             // table never shows dead columns. Uses the reliable async
@@ -33,7 +33,7 @@ pub(super) async fn api_schedules(State(state): State<AppState>) -> Response {
                 .into_iter()
                 .map(|c| c.name)
                 .collect();
-            crate::manager_server::filter_ghost_schedule_targets(&mut wire, &live);
+            crate::socket_server::filter_ghost_schedule_targets(&mut wire, &live);
             axum::Json(wire).into_response()
         }
         Err(e) => error_response(&format!("scheduled_prompts list: {e:#}")),
@@ -214,7 +214,7 @@ pub(super) async fn patch_schedule(
     }
     match state.coord.scheduled_prompts.get(id) {
         Ok(Some(s)) => {
-            let wire = crate::manager_server::schedule_to_wire_public(s);
+            let wire = crate::socket_server::schedule_to_wire_public(s);
             state.coord.emit_schedules_snapshot();
             axum::Json(wire).into_response()
         }
diff --git a/hive-c0re/src/lib.rs b/hive-c0re/src/lib.rs
index 2fe98c75..53f66241 100644
--- a/hive-c0re/src/lib.rs
+++ b/hive-c0re/src/lib.rs
@@ -38,7 +38,6 @@ pub mod knowledge;
 pub mod lifecycle;
 pub mod limits;
 pub mod loose_ends;
-pub mod manager_server;
 pub mod matrix;
 pub mod meta;
 pub mod migrate;
diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs
index cd9d3a20..0dbad747 100644
--- a/hive-c0re/src/main.rs
+++ b/hive-c0re/src/main.rs
@@ -13,8 +13,8 @@ use hive_sh4re::{HostRequest, HostResponse};
 use hive_c0re::coordinator::{Coordinator, HiveEnv, ServeConfig};
 use hive_c0re::{
     agent_sockets, auto_update, bash_tasks_vacuum, broker, client, crash_watch, dashboard,
-    dashboard_events, events_vacuum, forge, knowledge, manager_server, matrix, migrate,
-    rebuild_queue, reminder_scheduler, scheduled_prompts_worker, server,
+    dashboard_events, events_vacuum, forge, knowledge, matrix, migrate, rebuild_queue,
+    reminder_scheduler, scheduled_prompts_worker, server, socket_server,
 };
 
 #[derive(Parser)]
@@ -265,7 +265,7 @@ async fn cmd_serve(
     // Copy value first for the dashboard + knowledge-webhook tasks.
     let dashboard_port = env.dashboard_port;
     let coord = Arc::new(Coordinator::open(&db, env, model_prices)?);
-    manager_server::start(coord.clone())?;
+    socket_server::start_manager(coord.clone())?;
     // Idempotent pre-flight: rewrite pre-meta-layout applied
     // repos, ensure proposed repos carry the `applied`
     // remote, bootstrap the meta repo, repoint containers at
diff --git a/hive-c0re/src/manager_server.rs b/hive-c0re/src/manager_server.rs
deleted file mode 100644
index fed755ef..00000000
--- a/hive-c0re/src/manager_server.rs
+++ /dev/null
@@ -1,1029 +0,0 @@
-//! 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};
-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<()> {
-    use std::os::unix::fs::PermissionsExt as _;
-    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 root user (non-root) can connect;
-    // the bind source dir is manager-only on host. See socket_server.rs.
-    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?;
-    }
-}
-
-async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResponse {
-    // Delegate all variants shared with the agent socket to the common
-    // handler. `privileged = true`: every request here arrived on the
-    // manager socket, which is the trust boundary — so the shared guards
-    // grant manager-level authority from the socket, not from matching the
-    // `MANAGER_AGENT` name. `MANAGER_AGENT` is still passed as the actor
-    // name for attribution/routing (notifications, ownership), not authz.
-    if let Some(resp) = crate::socket_server::dispatch_shared(req, MANAGER_AGENT, true, coord).await
-    {
-        return resp;
-    }
-    match req {
-        ManagerRequest::RequestInitConfig { name, description } => {
-            handle_manager_init_config(coord, name, description.clone())
-        }
-        ManagerRequest::Kill { name } => handle_manager_kill(coord, name).await,
-        ManagerRequest::Start { name } => handle_manager_start(coord, name).await,
-        ManagerRequest::Restart { name } => handle_manager_restart(coord, name),
-        ManagerRequest::Update { name } => handle_manager_update(coord, name),
-        ManagerRequest::RequestUpdateMetaInputs {
-            inputs,
-            description,
-        } => handle_request_update_meta_inputs(coord, inputs, description.as_deref()),
-        ManagerRequest::RequestSchedulePrompt(payload) => {
-            handle_request_schedule_prompt(coord, MANAGER_AGENT, payload)
-        }
-        ManagerRequest::CancelSchedule { id, targets } => {
-            handle_cancel_schedule(coord, 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,
-            MANAGER_AGENT,
-            *id,
-            EditSchedulePatch {
-                body: body.clone(),
-                description: description.clone(),
-                interval_seconds: *interval_seconds,
-                next_fire_at_unix: *next_fire_at_unix,
-                targets_add: targets_add.clone(),
-                targets_remove: targets_remove.clone(),
-            },
-        ),
-        ManagerRequest::ListSchedules => handle_list_schedules(coord),
-        ManagerRequest::FireScheduleNow { id } => {
-            handle_fire_schedule_now(coord, MANAGER_AGENT, *id).await
-        }
-        ManagerRequest::GetLogs { agent, lines } => handle_get_logs(agent, *lines).await,
-        ManagerRequest::RequestApplyCommit {
-            agent,
-            commit_ref,
-            description,
-        } => handle_manager_apply_commit(coord, agent, commit_ref, description.as_deref()).await,
-        ManagerRequest::GetLooseEnds { agent } => {
-            handle_manager_loose_ends(coord, agent.as_deref())
-        }
-        ManagerRequest::CountPendingReminders { agent } => {
-            handle_manager_count_pending_reminders(coord, agent.as_deref())
-        }
-        ManagerRequest::ReminderRollup { since_secs, agent } => {
-            handle_manager_reminder_rollup(coord, agent.as_deref(), *since_secs)
-        }
-        _ => ManagerResponse::Err {
-            message: "request not handled on manager socket".to_owned(),
-        },
-    }
-}
-
-/// `RequestInitConfig` (manager socket) — queue an `InitConfig`
-/// approval. No topology check: the manager can act on any agent.
-fn handle_manager_init_config(
-    coord: &Arc<Coordinator>,
-    name: &str,
-    description: Option<String>,
-) -> ManagerResponse {
-    tracing::info!(%name, "manager: request_init_config");
-    // No explicit parent edge from the privileged socket — the new
-    // agent takes `topology::reconcile`'s default position on first
-    // spawn. The agent socket is the path that records an explicit
-    // requester-as-parent edge.
-    match submit_init_config(coord, name, None, description) {
-        Ok(_id) => ManagerResponse::Ok,
-        Err(e) => ManagerResponse::Err {
-            message: format!("{e:#}"),
-        },
-    }
-}
-
-/// `Kill` (manager socket) — kill the container, unregister it, notify.
-async fn handle_manager_kill(coord: &Arc<Coordinator>, name: &str) -> ManagerResponse {
-    tracing::info!(%name, "manager: kill");
-    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.to_owned(),
-            });
-            ManagerResponse::Ok
-        }
-        Err(e) => ManagerResponse::Err {
-            message: format!("{e:#}"),
-        },
-    }
-}
-
-/// `Start` (manager socket) — start the container, kick its next turn.
-async fn handle_manager_start(coord: &Arc<Coordinator>, name: &str) -> ManagerResponse {
-    tracing::info!(%name, "manager: start");
-    match lifecycle::start(name).await {
-        Ok(()) => {
-            coord.kick_agent(name, "container started");
-            ManagerResponse::Ok
-        }
-        Err(e) => ManagerResponse::Err {
-            message: format!("{e:#}"),
-        },
-    }
-}
-
-/// `Restart` (manager socket) — enqueue a restart.
-fn handle_manager_restart(coord: &Arc<Coordinator>, name: &str) -> ManagerResponse {
-    tracing::info!(%name, "manager: enqueue restart");
-    coord.rebuild_queue.enqueue(
-        crate::rebuild_queue::QueueKind::Restart,
-        name.to_owned(),
-        crate::rebuild_queue::QueueSource::Manual,
-        "manager `restart` tool".to_owned(),
-        None,
-    );
-    coord.emit_rebuild_queue_snapshot();
-    ManagerResponse::Ok
-}
-
-/// `Update` (manager socket) — enqueue a rebuild.
-fn handle_manager_update(coord: &Arc<Coordinator>, name: &str) -> ManagerResponse {
-    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
-}
-
-/// `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).
-fn handle_request_update_meta_inputs(
-    coord: &Arc<Coordinator>,
-    inputs: &[String],
-    description: Option<&str>,
-) -> ManagerResponse {
-    let label = if inputs.is_empty() {
-        "all inputs".to_string()
-    } else {
-        inputs.join(", ")
-    };
-    tracing::info!(%label, "manager: request_update_meta_inputs");
-    let commit_ref = serde_json::to_string(inputs).unwrap_or_default();
-    let id = match coord
-        .approvals
-        .submit_kind(
-            MANAGER_AGENT,
-            hive_sh4re::ApprovalKind::UpdateMetaInputs,
-            &commit_ref,
-            description,
-        )
-        .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,
-        MANAGER_AGENT,
-        "update_meta_inputs",
-        None,
-        None,
-        description.map(str::to_owned),
-    );
-    ManagerResponse::Ok
-}
-
-/// `ListSchedules` — snapshot every scheduled prompt onto the wire.
-fn handle_list_schedules(coord: &Arc<Coordinator>) -> ManagerResponse {
-    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:#}"),
-        },
-    }
-}
-
-/// `GetLogs` — read a child container's journal via hive-priv (the
-/// `-M` read needs root). `journalctl -M` wants the `h-<name>` machine
-/// name, which `container_name` derives.
-async fn handle_get_logs(agent: &str, lines: Option<u32>) -> ManagerResponse {
-    let n = lines.unwrap_or(50);
-    let machine = crate::lifecycle::container_name(agent);
-    tracing::info!(%agent, %machine, %n, "manager: get_logs");
-    match crate::priv_client::read_container_journal(
-        &machine,
-        hive_sh4re::priv_proto::JournalQuery {
-            lines: n,
-            ..Default::default()
-        },
-    )
-    .await
-    {
-        Ok((stdout, stderr)) => {
-            let content = if stdout.is_empty() { stderr } else { stdout };
-            ManagerResponse::Logs { content }
-        }
-        Err(e) => ManagerResponse::Err {
-            message: format!("get_logs: {e:#}"),
-        },
-    }
-}
-
-/// `RequestApplyCommit` (manager socket) — queue an apply-commit
-/// approval + plant the `proposal/<id>` tag.
-async fn handle_manager_apply_commit(
-    coord: &Arc<Coordinator>,
-    agent: &str,
-    commit_ref: &str,
-    description: Option<&str>,
-) -> ManagerResponse {
-    tracing::info!(%agent, %commit_ref, "manager: request_apply_commit");
-    match submit_apply_commit(coord, agent, commit_ref, description).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:#}"),
-        },
-    }
-}
-
-/// `GetLooseEnds` (manager socket) — manager's own threads by default,
-/// a named agent's when given, or hive-wide for `"*"` (which requires
-/// the `query_agent_state` capability).
-fn handle_manager_loose_ends(coord: &Arc<Coordinator>, agent: Option<&str>) -> ManagerResponse {
-    let result = match agent {
-        Some("*") => {
-            if !crate::capabilities::has_cap(MANAGER_AGENT, hive_sh4re::Capability::QueryAgentState)
-            {
-                return ManagerResponse::Err {
-                    message: "query_agent_state capability required for hive-wide loose ends"
-                        .into(),
-                };
-            }
-            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:#}"),
-        },
-    }
-}
-
-/// `CountPendingReminders` (manager socket) — count pending reminders
-/// for the target (defaults to the manager itself).
-fn handle_manager_count_pending_reminders(
-    coord: &Arc<Coordinator>,
-    agent: Option<&str>,
-) -> ManagerResponse {
-    let target = agent.unwrap_or(MANAGER_AGENT);
-    match coord.broker.count_pending_reminders_for(target) {
-        Ok(count) => ManagerResponse::PendingRemindersCount { count },
-        Err(e) => ManagerResponse::Err {
-            message: format!("{e:#}"),
-        },
-    }
-}
-
-/// `ReminderRollup` (manager socket) — roll up reminders fired in the
-/// last `since_secs` for the target (defaults to the manager itself).
-fn handle_manager_reminder_rollup(
-    coord: &Arc<Coordinator>,
-    agent: Option<&str>,
-    since_secs: u64,
-) -> ManagerResponse {
-    let target = agent.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:#}"),
-        },
-    }
-}
-
-/// `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. The agent socket passes the requesting agent. `None` (the
-/// privileged manager socket) writes no explicit edge — the new agent
-/// lands at `topology::reconcile`'s default position when it first
-/// spawns, so no caller has to name a specific root agent here.
-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(),
-        )
-        .map_err(|e| anyhow::anyhow!("queue approval row: {e:#}"))?;
-    tracing::info!(%id, %name, "init_config approval queued");
-    coord.emit_approval_added(id, name, "init_config", None, None, description);
-    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>,
-) -> 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(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(
-        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(()) => {
-            coord.emit_schedules_snapshot();
-            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
-            ),
-        };
-    }
-    // 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();
-            ManagerResponse::Ok
-        }
-        Err(e) => ManagerResponse::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"
-)]
-struct EditSchedulePatch {
-    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>>,
-}
-
-/// 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.
-fn handle_edit_schedule(
-    coord: &Arc<Coordinator>,
-    requester: &str,
-    schedule_id: i64,
-    patch: EditSchedulePatch,
-) -> ManagerResponse {
-    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 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(()) => {
-            coord.emit_schedules_snapshot();
-            ManagerResponse::Ok
-        }
-        Err(e) => ManagerResponse::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 manager-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: 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::{filter_ghost_schedule_targets, validate_commit_ref};
-
-    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: 0,
-            created_at_unix: 0,
-            source: hive_sh4re::WireScheduleSource::Operator,
-            cancelled_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());
-    }
-
-    #[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());
-    }
-}
diff --git a/hive-c0re/src/questions.rs b/hive-c0re/src/questions.rs
index 0a13039a..4e23f8eb 100644
--- a/hive-c0re/src/questions.rs
+++ b/hive-c0re/src/questions.rs
@@ -21,7 +21,7 @@ use std::sync::Arc;
 use crate::approvals::kind_to_str;
 use crate::coordinator::Coordinator;
 use crate::limits;
-use crate::manager_server::spawn_question_watchdog;
+use crate::socket_server::spawn_question_watchdog;
 
 /// Cap on how long an asker can demand an answer before the watchdog
 /// auto-resolves with `[expired]`. Six hours mirrors typical agent
diff --git a/hive-c0re/src/socket_server.rs b/hive-c0re/src/socket_server.rs
index 6ab72491..7b414212 100644
--- a/hive-c0re/src/socket_server.rs
+++ b/hive-c0re/src/socket_server.rs
@@ -1,12 +1,17 @@
-//! Per-agent socket listener. Each socket file's existence on disk
-//! authenticates the caller: connecting to `<.../agents/foo/mcp.sock>` means
-//! you are `foo`.
+//! Unix-socket request server, shared by the per-agent sockets and the
+//! manager socket. The socket file's existence on disk authenticates the
+//! caller: connecting to `<.../agents/foo/mcp.sock>` means you are `foo`
+//! (non-privileged); connecting to the manager socket grants privileged
+//! authority. Both transports run the same [`serve`] / [`dispatch`] code,
+//! parameterised by a `privileged: bool` carried from the listener — the
+//! privilege gate (and the topology guards on lifecycle verbs) keys off
+//! that flag, not off matching the `MANAGER_AGENT` name.
 
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
 use anyhow::{Context, Result};
-use hive_sh4re::{AgentRequest, AgentResponse, Message};
+use hive_sh4re::{AgentRequest, AgentResponse, MANAGER_AGENT, Message};
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 use tokio::net::{UnixListener, UnixStream};
 use tokio::task::JoinHandle;
@@ -49,7 +54,8 @@ pub fn start(agent: &str, socket_path: &Path, coord: Arc<Coordinator>) -> Result
                     let agent = agent.clone();
                     let coord = coord.clone();
                     tokio::spawn(async move {
-                        if let Err(e) = serve(stream, agent, coord).await {
+                        // Per-agent socket: never privileged.
+                        if let Err(e) = serve(stream, agent, false, coord).await {
                             tracing::warn!(error = ?e, "agent connection failed");
                         }
                     });
@@ -64,7 +70,56 @@ pub fn start(agent: &str, socket_path: &Path, coord: Arc<Coordinator>) -> Result
     Ok(AgentSocket { path, handle })
 }
 
-async fn serve(stream: UnixStream, agent: String, coord: Arc<Coordinator>) -> Result<()> {
+/// Bind + serve the manager socket. The manager socket is the privileged
+/// trust boundary: every connection that lands here runs `dispatch` with
+/// `privileged = true`. `MANAGER_AGENT` is passed as the actor name for
+/// attribution/routing (notifications, ownership), not for authorisation —
+/// authority comes from the socket, per #1834.
+pub fn start_manager(coord: Arc<Coordinator>) -> Result<()> {
+    use std::os::unix::fs::PermissionsExt as _;
+    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 root user (non-root) can connect; the bind
+    // source dir is manager-only on host (see the per-agent socket above).
+    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 {
+                        // Manager socket: privileged.
+                        if let Err(e) = serve(stream, MANAGER_AGENT.to_owned(), true, 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,
+    agent: String,
+    privileged: bool,
+    coord: Arc<Coordinator>,
+) -> Result<()> {
     let (read, mut write) = stream.into_split();
     let mut reader = BufReader::new(read);
     let mut line = String::new();
@@ -75,7 +130,7 @@ async fn serve(stream: UnixStream, agent: String, coord: Arc<Coordinator>) -> Re
             return Ok(());
         }
         let resp = match serde_json::from_str::<AgentRequest>(line.trim()) {
-            Ok(req) => dispatch(&req, &agent, &coord).await,
+            Ok(req) => dispatch(&req, &agent, privileged, &coord).await,
             Err(e) => AgentResponse::Err {
                 message: format!("parse error: {e}"),
             },
@@ -119,8 +174,8 @@ pub(crate) fn recv_timeout(wait_seconds: Option<u64>) -> std::time::Duration {
 /// semantics (e.g. `GetLooseEnds` / `CountPendingReminders` / `ReminderRollup`
 /// where the manager can target other agents) or for manager-only variants.
 ///
-/// Both `socket_server::dispatch` and `manager_server::dispatch` call this
-/// first; each then handles its own remaining arms.
+/// The unified `dispatch` calls this first (for both the privileged and
+/// non-privileged paths); the remaining arms are handled there.
 pub(crate) async fn dispatch_shared(
     req: &hive_sh4re::Request,
     agent: &str,
@@ -434,30 +489,30 @@ fn handle_requeue_inflight(coord: &Arc<Coordinator>, agent: &str) -> hive_sh4re:
     }
 }
 
-async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) -> AgentResponse {
-    // Regular agent socket: never privileged. Privilege is reserved for
-    // requests arriving on the manager socket (see `manager_server`).
-    if let Some(resp) = dispatch_shared(req, agent, false, coord).await {
+/// Unified dispatch for both the per-agent sockets (`privileged = false`)
+/// and the manager socket (`privileged = true`). Shared variants go through
+/// [`dispatch_shared`]; lifecycle/config + agent-state-query verbs apply the
+/// topology/capability guards on the non-privileged path and skip them when
+/// privileged; the schedule / meta-input / log verbs are privileged-only.
+async fn dispatch(
+    req: &AgentRequest,
+    agent: &str,
+    privileged: bool,
+    coord: &Arc<Coordinator>,
+) -> AgentResponse {
+    if let Some(resp) = dispatch_shared(req, agent, privileged, coord).await {
         return resp;
     }
     match req {
-        AgentRequest::GetLooseEnds { agent: target } => {
-            handle_get_loose_ends(coord, agent, target.as_deref())
-        }
-        AgentRequest::CountPendingReminders { agent: target } => {
-            handle_count_pending_reminders(coord, agent, target.as_deref())
-        }
-        AgentRequest::ReminderRollup {
-            since_secs,
-            agent: target,
-        } => handle_reminder_rollup(coord, agent, target.as_deref(), *since_secs),
-        AgentRequest::Start { name } => handle_start_child(coord, agent, name).await,
-        AgentRequest::Restart { name } => handle_restart_child(coord, agent, name).await,
-        AgentRequest::Kill { name } => handle_kill_child(coord, agent, name).await,
-        AgentRequest::Update { name } => handle_update_child(coord, agent, name),
+        // Lifecycle + config: topology-gated when `!privileged`, ungated
+        // (any agent) when privileged.
+        AgentRequest::Start { name } => handle_start(coord, agent, name, privileged).await,
+        AgentRequest::Restart { name } => handle_restart(coord, agent, name, privileged).await,
+        AgentRequest::Kill { name } => handle_kill(coord, agent, name, privileged).await,
+        AgentRequest::Update { name } => handle_update(coord, agent, name, privileged),
         AgentRequest::ListDescendants => handle_list_descendants(agent).await,
         AgentRequest::RequestInitConfig { name, description } => {
-            handle_request_init_config(coord, agent, name, description.clone())
+            handle_request_init_config(coord, agent, name, description.clone(), privileged)
         }
         AgentRequest::RequestApplyCommit {
             agent: target_agent,
@@ -470,12 +525,86 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) ->
                 target_agent,
                 commit_ref,
                 description.as_deref(),
+                privileged,
             )
             .await
         }
-        // Manager-only variants are not valid on the agent socket.
+        // Agent-state queries: own/child/cap-gated when `!privileged`;
+        // any-agent + hive-wide (`"*"`) when privileged.
+        AgentRequest::GetLooseEnds { agent: target } => {
+            handle_get_loose_ends(coord, agent, target.as_deref(), privileged)
+        }
+        AgentRequest::CountPendingReminders { agent: target } => {
+            handle_count_pending_reminders(coord, agent, target.as_deref(), privileged)
+        }
+        AgentRequest::ReminderRollup {
+            since_secs,
+            agent: target,
+        } => handle_reminder_rollup(coord, agent, target.as_deref(), *since_secs, privileged),
+        // Everything else is privileged-only (scheduling / meta-inputs /
+        // logs) or unknown — gated + handled in `dispatch_privileged_only`.
+        _ => dispatch_privileged_only(req, agent, privileged, coord).await,
+    }
+}
+
+/// Handle the privileged-only verbs (scheduling, meta-input updates,
+/// container logs). Reached as the fallback arm of [`dispatch`]: rejects the
+/// whole group up front when `!privileged` (these never appear on a per-agent
+/// socket), then matches the individual verbs. Any other variant is a
+/// host-admin-only / unknown request that's invalid on either socket.
+async fn dispatch_privileged_only(
+    req: &AgentRequest,
+    agent: &str,
+    privileged: bool,
+    coord: &Arc<Coordinator>,
+) -> AgentResponse {
+    if !privileged {
+        return AgentResponse::Err {
+            message: "request not available on the agent socket (privileged / manager-only)"
+                .to_owned(),
+        };
+    }
+    match req {
+        AgentRequest::RequestUpdateMetaInputs {
+            inputs,
+            description,
+        } => handle_request_update_meta_inputs(coord, inputs, description.as_deref()),
+        AgentRequest::RequestSchedulePrompt(payload) => {
+            handle_request_schedule_prompt(coord, agent, payload)
+        }
+        AgentRequest::CancelSchedule { id, targets } => {
+            handle_cancel_schedule(coord, agent, *id, targets.as_deref())
+        }
+        AgentRequest::EditSchedule {
+            id,
+            body,
+            description,
+            interval_seconds,
+            next_fire_at_unix,
+            targets_add,
+            targets_remove,
+        } => handle_edit_schedule(
+            coord,
+            agent,
+            *id,
+            EditSchedulePatch {
+                body: body.clone(),
+                description: description.clone(),
+                interval_seconds: *interval_seconds,
+                next_fire_at_unix: *next_fire_at_unix,
+                targets_add: targets_add.clone(),
+                targets_remove: targets_remove.clone(),
+            },
+        ),
+        AgentRequest::ListSchedules => handle_list_schedules(coord),
+        AgentRequest::FireScheduleNow { id } => handle_fire_schedule_now(coord, agent, *id).await,
+        AgentRequest::GetLogs {
+            agent: target,
+            lines,
+        } => handle_get_logs(target, *lines).await,
+        // Host-admin-only / unknown variants: never valid on either socket.
         _ => AgentResponse::Err {
-            message: "request not supported on agent socket".to_owned(),
+            message: "request not handled on this socket".to_owned(),
         },
     }
 }
@@ -535,21 +664,59 @@ fn require_new_child(agent: &str, target: &str, action: &str) -> Option<AgentRes
     }
 }
 
-/// `GetLooseEnds` — resolve the (optionally cross-agent) target then
-/// read its loose ends.
+/// `GetLooseEnds` — read the target's loose ends. The privileged (manager)
+/// socket may target any named agent, defaults to itself, and may pass
+/// `"*"` for a hive-wide sweep (gated on `query_agent_state`). The
+/// non-privileged path resolves through `resolve_agent_state_target`
+/// (own / direct-child free; other agents need `query_agent_state`; `"*"`
+/// is rejected).
 fn handle_get_loose_ends(
     coord: &Arc<Coordinator>,
     agent: &str,
     target: Option<&str>,
+    privileged: bool,
 ) -> AgentResponse {
-    match resolve_agent_state_target(agent, target) {
-        Ok(name) => match crate::loose_ends::for_agent(coord, name) {
-            Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends },
-            Err(e) => AgentResponse::Err {
-                message: format!("{e:#}"),
-            },
+    let result = if privileged {
+        match target {
+            Some("*") => {
+                if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::QueryAgentState) {
+                    return AgentResponse::Err {
+                        message: "query_agent_state capability required for hive-wide loose ends"
+                            .to_owned(),
+                    };
+                }
+                crate::loose_ends::hive_wide(coord)
+            }
+            Some(name) => crate::loose_ends::for_agent(coord, name),
+            None => crate::loose_ends::for_agent(coord, agent),
+        }
+    } else {
+        match resolve_agent_state_target(agent, target) {
+            Ok(name) => crate::loose_ends::for_agent(coord, name),
+            Err(message) => return AgentResponse::Err { message },
+        }
+    };
+    match result {
+        Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends },
+        Err(e) => AgentResponse::Err {
+            message: format!("{e:#}"),
         },
-        Err(message) => AgentResponse::Err { message },
+    }
+}
+
+/// Resolve the target agent for `CountPendingReminders` / `ReminderRollup`.
+/// Privileged (manager) callers may name any agent (or default to
+/// themselves); non-privileged callers go through the topology/capability
+/// gate in `resolve_agent_state_target`.
+fn resolve_reminder_target<'a>(
+    caller: &'a str,
+    target: Option<&'a str>,
+    privileged: bool,
+) -> Result<&'a str, String> {
+    if privileged {
+        Ok(target.unwrap_or(caller))
+    } else {
+        resolve_agent_state_target(caller, target)
     }
 }
 
@@ -559,8 +726,9 @@ fn handle_count_pending_reminders(
     coord: &Arc<Coordinator>,
     agent: &str,
     target: Option<&str>,
+    privileged: bool,
 ) -> AgentResponse {
-    match resolve_agent_state_target(agent, target) {
+    match resolve_reminder_target(agent, target, privileged) {
         Ok(name) => match coord.broker.count_pending_reminders_for(name) {
             Ok(count) => AgentResponse::PendingRemindersCount { count },
             Err(e) => AgentResponse::Err {
@@ -578,8 +746,9 @@ fn handle_reminder_rollup(
     agent: &str,
     target: Option<&str>,
     since_secs: u64,
+    privileged: bool,
 ) -> AgentResponse {
-    match resolve_agent_state_target(agent, target) {
+    match resolve_reminder_target(agent, target, privileged) {
         Ok(name) => match coord.broker.reminder_rollup_for(name, since_secs) {
             Ok(stats) => AgentResponse::ReminderRollup(stats),
             Err(e) => AgentResponse::Err {
@@ -590,12 +759,19 @@ fn handle_reminder_rollup(
     }
 }
 
-/// `Start` — start a direct-child container, kicking its next turn.
-async fn handle_start_child(coord: &Arc<Coordinator>, agent: &str, name: &str) -> AgentResponse {
-    if let Some(err) = require_child(agent, name, "start") {
+/// `Start` — start a container, kicking its next turn. Non-privileged
+/// callers may only start a direct child; the privileged (manager) socket
+/// may start any agent.
+async fn handle_start(
+    coord: &Arc<Coordinator>,
+    agent: &str,
+    name: &str,
+    privileged: bool,
+) -> AgentResponse {
+    if !privileged && let Some(err) = require_child(agent, name, "start") {
         return err;
     }
-    tracing::info!(%agent, %name, "agent: start child");
+    tracing::info!(%agent, %name, %privileged, "start container");
     match crate::lifecycle::start(name).await {
         Ok(()) => {
             coord.kick_agent(name, "container started");
@@ -607,10 +783,17 @@ async fn handle_start_child(coord: &Arc<Coordinator>, agent: &str, name: &str) -
     }
 }
 
-/// `Restart` — enqueue a restart for a direct-child container.
-/// Topology parenthood is the only authorisation criterion — no
-/// capability flag needed.
-async fn handle_restart_child(coord: &Arc<Coordinator>, agent: &str, name: &str) -> AgentResponse {
+/// `Restart` — enqueue a restart for a container. Non-privileged callers
+/// may only restart a direct child; the privileged (manager) socket may
+/// restart any agent. The infra-container branch is orthogonal: it is gated
+/// on the `infra_admin` capability (applies to privileged + non-privileged
+/// callers alike) and audited, so it stays ahead of the topology guard.
+async fn handle_restart(
+    coord: &Arc<Coordinator>,
+    agent: &str,
+    name: &str,
+    privileged: bool,
+) -> 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
@@ -620,10 +803,10 @@ async fn handle_restart_child(coord: &Arc<Coordinator>, agent: &str, name: &str)
     if let Ok(container) = name.parse::<hive_sh4re::priv_proto::InfraContainer>() {
         return handle_restart_infra(coord, agent, container).await;
     }
-    if let Some(err) = require_child(agent, name, "restart") {
+    if !privileged && let Some(err) = require_child(agent, name, "restart") {
         return err;
     }
-    tracing::info!(%agent, %name, "agent: enqueue restart for child");
+    tracing::info!(%agent, %name, %privileged, "enqueue restart");
     coord.rebuild_queue.enqueue(
         crate::rebuild_queue::QueueKind::Restart,
         name.to_owned(),
@@ -686,13 +869,19 @@ async fn handle_restart_infra(
     }
 }
 
-/// `Kill` — kill a direct-child container, unregister it, notify the
-/// manager.
-async fn handle_kill_child(coord: &Arc<Coordinator>, agent: &str, name: &str) -> AgentResponse {
-    if let Some(err) = require_child(agent, name, "kill") {
+/// `Kill` — kill a container, unregister it, notify the manager.
+/// Non-privileged callers may only kill a direct child; the privileged
+/// (manager) socket may kill any agent.
+async fn handle_kill(
+    coord: &Arc<Coordinator>,
+    agent: &str,
+    name: &str,
+    privileged: bool,
+) -> AgentResponse {
+    if !privileged && let Some(err) = require_child(agent, name, "kill") {
         return err;
     }
-    tracing::info!(%agent, %name, "agent: kill child");
+    tracing::info!(%agent, %name, %privileged, "kill container");
     let result: anyhow::Result<()> = async {
         crate::lifecycle::kill(name).await?;
         coord.unregister_agent(name);
@@ -712,12 +901,19 @@ async fn handle_kill_child(coord: &Arc<Coordinator>, agent: &str, name: &str) ->
     }
 }
 
-/// `Update` — enqueue a rebuild for a direct-child container.
-fn handle_update_child(coord: &Arc<Coordinator>, agent: &str, name: &str) -> AgentResponse {
-    if let Some(err) = require_child(agent, name, "rebuild") {
+/// `Update` — enqueue a rebuild for a container. Non-privileged callers may
+/// only rebuild a direct child; the privileged (manager) socket may rebuild
+/// any agent.
+fn handle_update(
+    coord: &Arc<Coordinator>,
+    agent: &str,
+    name: &str,
+    privileged: bool,
+) -> AgentResponse {
+    if !privileged && let Some(err) = require_child(agent, name, "rebuild") {
         return err;
     }
-    tracing::info!(%agent, %name, "agent: enqueue rebuild for child");
+    tracing::info!(%agent, %name, %privileged, "enqueue rebuild");
     coord.rebuild_queue.enqueue(
         crate::rebuild_queue::QueueKind::Rebuild,
         name.to_owned(),
@@ -767,19 +963,24 @@ async fn handle_list_descendants(agent: &str) -> AgentResponse {
     AgentResponse::Containers { containers }
 }
 
-/// `RequestInitConfig` — queue an `InitConfig` approval for a
-/// direct-child agent.
+/// `RequestInitConfig` — queue an `InitConfig` approval for an agent.
+/// Non-privileged callers may only init a (brand-new or existing) direct
+/// child and are recorded as its parent; the privileged (manager) socket
+/// may init any agent and records no explicit parent edge (the new agent
+/// lands at `topology::reconcile`'s default position on first spawn).
 fn handle_request_init_config(
     coord: &Arc<Coordinator>,
     agent: &str,
     name: &str,
     description: Option<String>,
+    privileged: bool,
 ) -> AgentResponse {
-    if let Some(err) = require_new_child(agent, name, "request_init_config for") {
+    if !privileged && let Some(err) = require_new_child(agent, name, "request_init_config for") {
         return err;
     }
-    tracing::info!(%agent, %name, "agent: request_init_config for child");
-    match crate::manager_server::submit_init_config(coord, name, Some(agent), description) {
+    tracing::info!(%agent, %name, %privileged, "request_init_config");
+    let parent = if privileged { None } else { Some(agent) };
+    match submit_init_config(coord, name, parent, description) {
         Ok(_id) => AgentResponse::Ok,
         Err(e) => AgentResponse::Err {
             message: format!("{e:#}"),
@@ -787,24 +988,26 @@ fn handle_request_init_config(
     }
 }
 
-/// `RequestApplyCommit` — queue an apply-commit approval for a
-/// direct-child agent.
+/// `RequestApplyCommit` — queue an apply-commit approval for an agent.
+/// Non-privileged callers may only target a direct child; the privileged
+/// (manager) socket may target any agent.
 async fn handle_request_apply_commit(
     coord: &Arc<Coordinator>,
     agent: &str,
     target_agent: &str,
     commit_ref: &str,
     description: Option<&str>,
+    privileged: bool,
 ) -> AgentResponse {
-    if let Some(err) = require_new_child(agent, target_agent, "request_apply_commit for") {
+    if !privileged
+        && let Some(err) = require_new_child(agent, target_agent, "request_apply_commit for")
+    {
         return err;
     }
-    tracing::info!(%agent, %target_agent, %commit_ref, "agent: request_apply_commit for child");
-    match crate::manager_server::submit_apply_commit(coord, target_agent, commit_ref, description)
-        .await
-    {
+    tracing::info!(%agent, %target_agent, %commit_ref, %privileged, "request_apply_commit");
+    match submit_apply_commit(coord, target_agent, commit_ref, description).await {
         Ok((id, sha)) => {
-            tracing::info!(%id, %target_agent, %sha, "agent: apply_commit approval queued");
+            tracing::info!(%id, %target_agent, %sha, "apply_commit approval queued");
             AgentResponse::Ok
         }
         Err(e) => AgentResponse::Err {
@@ -953,7 +1156,7 @@ pub(crate) fn fan_out_send(
 /// Common Send handler shared between dispatch arms. Applies the
 /// 4 KiB body cap, then routes broadcast (`to == "*"`) / children fan-out
 /// (`to == "<children>"`) / unicast through their respective broker calls.
-/// `pub(crate)` so `dispatch_shared` (and via it, `manager_server`) can use it.
+/// `pub(crate)` so `dispatch_shared` can use it across both socket paths.
 pub(crate) fn handle_send(
     coord: &Arc<Coordinator>,
     agent: &str,
@@ -1217,6 +1420,652 @@ fn resolve_due_at(timing: &hive_sh4re::ReminderTiming) -> anyhow::Result<i64> {
     }
 }
 
+// ---------------------------------------------------------------------------
+// Privileged-only handlers + submit/schedule helpers (manager socket).
+// Reached only via the `require_privileged!()`-gated arms in `dispatch`, or
+// re-used by the agent-socket lifecycle handlers (`submit_init_config` /
+// `submit_apply_commit`) and the dashboard (`schedule_to_wire_public` /
+// `filter_ghost_schedule_targets`).
+// ---------------------------------------------------------------------------
+
+/// `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).
+fn handle_request_update_meta_inputs(
+    coord: &Arc<Coordinator>,
+    inputs: &[String],
+    description: Option<&str>,
+) -> AgentResponse {
+    let label = if inputs.is_empty() {
+        "all inputs".to_string()
+    } else {
+        inputs.join(", ")
+    };
+    tracing::info!(%label, "manager: request_update_meta_inputs");
+    let commit_ref = serde_json::to_string(inputs).unwrap_or_default();
+    let id = match coord
+        .approvals
+        .submit_kind(
+            MANAGER_AGENT,
+            hive_sh4re::ApprovalKind::UpdateMetaInputs,
+            &commit_ref,
+            description,
+        )
+        .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(
+        id,
+        MANAGER_AGENT,
+        "update_meta_inputs",
+        None,
+        None,
+        description.map(str::to_owned),
+    );
+    AgentResponse::Ok
+}
+
+/// `ListSchedules` — snapshot every scheduled prompt onto the wire.
+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:#}"),
+        },
+    }
+}
+
+/// `GetLogs` — read a child container's journal via hive-priv (the
+/// `-M` read needs root). `journalctl -M` wants the `h-<name>` machine
+/// name, which `container_name` derives.
+async fn handle_get_logs(agent: &str, lines: Option<u32>) -> AgentResponse {
+    let n = lines.unwrap_or(50);
+    let machine = crate::lifecycle::container_name(agent);
+    tracing::info!(%agent, %machine, %n, "manager: get_logs");
+    match crate::priv_client::read_container_journal(
+        &machine,
+        hive_sh4re::priv_proto::JournalQuery {
+            lines: n,
+            ..Default::default()
+        },
+    )
+    .await
+    {
+        Ok((stdout, stderr)) => {
+            let content = if stdout.is_empty() { stderr } else { stdout };
+            AgentResponse::Logs { content }
+        }
+        Err(e) => AgentResponse::Err {
+            message: format!("get_logs: {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.
+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(),
+    ) {
+        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(
+        id,
+        requester,
+        "schedule_prompt",
+        None,
+        None,
+        payload.description.clone(),
+    );
+    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.
+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`.
+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"
+)]
+struct EditSchedulePatch {
+    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>>,
+}
+
+/// 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.
+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)
+}
+
+/// `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. The agent socket passes the requesting agent. `None` (the
+/// privileged manager socket) writes no explicit edge — the new agent
+/// lands at `topology::reconcile`'s default position when it first
+/// spawns, so no caller has to name a specific root agent here.
+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(),
+        )
+        .map_err(|e| anyhow::anyhow!("queue approval row: {e:#}"))?;
+    tracing::info!(%id, %name, "init_config approval queued");
+    coord.emit_approval_added(id, name, "init_config", None, None, description);
+    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>,
+) -> 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,
+        )
+        .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(
+        id,
+        agent,
+        "apply_commit",
+        Some(sha_short),
+        Some(diff),
+        description.map(str::to_owned),
+    );
+    Ok((id, sha))
+}
+
+/// 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: 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::*;
@@ -1246,4 +2095,101 @@ mod tests {
         assert_eq!(msg, "small");
         assert_eq!(fp.as_deref(), Some("/agents/foo/state/x.md"));
     }
+
+    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: 0,
+            created_at_unix: 0,
+            source: hive_sh4re::WireScheduleSource::Operator,
+            cancelled_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());
+    }
+
+    #[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());
+    }
+
+    #[test]
+    fn reminder_target_privileged_resolves_any_or_self() {
+        // Privileged (manager) callers may name any agent, or default to
+        // themselves — no topology/capability gate on this path.
+        assert_eq!(
+            resolve_reminder_target("ruth", Some("iris"), true),
+            Ok("iris")
+        );
+        assert_eq!(resolve_reminder_target("ruth", None, true), Ok("ruth"));
+    }
+
+    #[test]
+    fn reminder_target_non_privileged_self_is_free() {
+        // The non-privileged path defers to `resolve_agent_state_target`;
+        // the self / default case needs no topology state.
+        assert_eq!(resolve_reminder_target("iris", None, false), Ok("iris"));
+        assert_eq!(
+            resolve_reminder_target("iris", Some("iris"), false),
+            Ok("iris")
+        );
+    }
 }
diff --git a/hive-c0re/src/topology.rs b/hive-c0re/src/topology.rs
index e2e94476..ce2f879d 100644
--- a/hive-c0re/src/topology.rs
+++ b/hive-c0re/src/topology.rs
@@ -140,7 +140,7 @@ pub fn resolve_recipient_in(
 /// `candidate` upward; the walk terminates at root or on a cycle
 /// (cycle defence: bounded to 32 hops, more than any plausible
 /// hive depth). Used by the cancel-authorization check in
-/// `manager_server::handle_cancel_schedule` to enforce
+/// `socket_server::handle_cancel_schedule` to enforce
 /// "managers can cancel anything their subtree owns."
 #[must_use]
 pub fn is_descendant_of(candidate: &str, ancestor: &str) -> bool {