refactor(#1474): extract dispatch arm logic in manager_server + hive-priv
This commit is contained in:
damocles
mara
parent
c6d9f59c4d
commit
02dcf4d028
2 changed files with 336 additions and 261 deletions
|
|
@ -74,11 +74,6 @@ async fn serve(stream: UnixStream, coord: Arc<Coordinator>) -> Result<()> {
|
|||
}
|
||||
}
|
||||
|
||||
#[allow(
|
||||
clippy::too_many_lines,
|
||||
reason = "flat dispatch table: one match arm per manager-socket request \
|
||||
variant; splitting it would scatter the routing logic"
|
||||
)]
|
||||
async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResponse {
|
||||
// Delegate all variants shared with the agent socket to the common handler.
|
||||
if let Some(resp) = crate::agent_server::dispatch_shared(req, MANAGER_AGENT, coord).await {
|
||||
|
|
@ -86,112 +81,16 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
|
|||
}
|
||||
match req {
|
||||
ManagerRequest::RequestInitConfig { name, description } => {
|
||||
tracing::info!(%name, "manager: request_init_config");
|
||||
match submit_init_config(coord, name, description.clone()) {
|
||||
Ok(_id) => ManagerResponse::Ok,
|
||||
Err(e) => ManagerResponse::Err {
|
||||
message: format!("{e:#}"),
|
||||
},
|
||||
}
|
||||
}
|
||||
ManagerRequest::Kill { name } => {
|
||||
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.clone(),
|
||||
});
|
||||
ManagerResponse::Ok
|
||||
}
|
||||
Err(e) => ManagerResponse::Err {
|
||||
message: format!("{e:#}"),
|
||||
},
|
||||
}
|
||||
}
|
||||
ManagerRequest::Start { name } => {
|
||||
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:#}"),
|
||||
},
|
||||
}
|
||||
}
|
||||
ManagerRequest::Restart { name } => {
|
||||
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
|
||||
}
|
||||
ManagerRequest::Update { name } => {
|
||||
tracing::info!(%name, "manager: enqueue update");
|
||||
coord.rebuild_queue.enqueue(
|
||||
crate::rebuild_queue::QueueKind::Rebuild,
|
||||
name.to_owned(),
|
||||
crate::rebuild_queue::QueueSource::Manual,
|
||||
"manager `update` tool".to_owned(),
|
||||
None,
|
||||
);
|
||||
coord.emit_rebuild_queue_snapshot();
|
||||
ManagerResponse::Ok
|
||||
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,
|
||||
} => {
|
||||
let label = if inputs.is_empty() {
|
||||
"all inputs".to_string()
|
||||
} else {
|
||||
inputs.join(", ")
|
||||
};
|
||||
tracing::info!(%label, "manager: request_update_meta_inputs");
|
||||
// Encode the inputs list as JSON and store it in commit_ref
|
||||
// (there's no git commit involved; the field carries the
|
||||
// payload for the approval handler to decode at run time).
|
||||
let commit_ref = serde_json::to_string(inputs).unwrap_or_default();
|
||||
let id = match coord
|
||||
.approvals
|
||||
.submit_kind(
|
||||
MANAGER_AGENT,
|
||||
hive_sh4re::ApprovalKind::UpdateMetaInputs,
|
||||
&commit_ref,
|
||||
description.as_deref(),
|
||||
)
|
||||
.map_err(|e| anyhow::anyhow!("{e:#}"))
|
||||
{
|
||||
Ok(id) => id,
|
||||
Err(e) => {
|
||||
return ManagerResponse::Err {
|
||||
message: format!("queue update_meta_inputs approval: {e:#}"),
|
||||
};
|
||||
}
|
||||
};
|
||||
tracing::info!(%id, %label, "update_meta_inputs approval queued");
|
||||
coord.emit_approval_added(
|
||||
id,
|
||||
MANAGER_AGENT,
|
||||
"update_meta_inputs",
|
||||
None,
|
||||
None,
|
||||
description.clone(),
|
||||
);
|
||||
ManagerResponse::Ok
|
||||
}
|
||||
} => handle_request_update_meta_inputs(coord, inputs, description.as_deref()),
|
||||
ManagerRequest::RequestSchedulePrompt(payload) => {
|
||||
handle_request_schedule_prompt(coord, MANAGER_AGENT, payload)
|
||||
}
|
||||
|
|
@ -219,101 +118,24 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
|
|||
targets_remove: targets_remove.clone(),
|
||||
},
|
||||
),
|
||||
ManagerRequest::ListSchedules => match coord.scheduled_prompts.list() {
|
||||
Ok(schedules) => ManagerResponse::Schedules {
|
||||
schedules: schedules.into_iter().map(schedule_to_wire).collect(),
|
||||
},
|
||||
Err(e) => ManagerResponse::Err {
|
||||
message: format!("list scheduled prompts: {e:#}"),
|
||||
},
|
||||
},
|
||||
ManagerRequest::ListSchedules => handle_list_schedules(coord),
|
||||
ManagerRequest::FireScheduleNow { id } => {
|
||||
handle_fire_schedule_now(coord, MANAGER_AGENT, *id).await
|
||||
}
|
||||
ManagerRequest::GetLogs { agent, lines } => {
|
||||
let n = lines.unwrap_or(50);
|
||||
// `journalctl -M` wants the container name (`h-<name>`),
|
||||
// not the logical agent name. `container_name` adds the prefix.
|
||||
// The `-M` read needs root, so it goes through hive-priv.
|
||||
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:#}"),
|
||||
},
|
||||
}
|
||||
}
|
||||
ManagerRequest::GetLogs { agent, lines } => handle_get_logs(agent, *lines).await,
|
||||
ManagerRequest::RequestApplyCommit {
|
||||
agent,
|
||||
commit_ref,
|
||||
description,
|
||||
} => {
|
||||
tracing::info!(%agent, %commit_ref, "manager: request_apply_commit");
|
||||
match submit_apply_commit(coord, agent, commit_ref, description.as_deref()).await {
|
||||
Ok((id, sha)) => {
|
||||
tracing::info!(%id, %agent, manager_ref = %commit_ref, %sha, "approval queued + proposal tag planted");
|
||||
ManagerResponse::Ok
|
||||
}
|
||||
Err(e) => ManagerResponse::Err {
|
||||
message: format!("{e:#}"),
|
||||
},
|
||||
}
|
||||
}
|
||||
} => handle_manager_apply_commit(coord, agent, commit_ref, description.as_deref()).await,
|
||||
ManagerRequest::GetLooseEnds { agent } => {
|
||||
let result = match agent.as_deref() {
|
||||
Some("*") => {
|
||||
// Hive-wide query requires query_agent_state capability.
|
||||
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:#}"),
|
||||
},
|
||||
}
|
||||
handle_manager_loose_ends(coord, agent.as_deref())
|
||||
}
|
||||
ManagerRequest::CountPendingReminders { agent } => {
|
||||
let target = agent.as_deref().unwrap_or(MANAGER_AGENT);
|
||||
match coord.broker.count_pending_reminders_for(target) {
|
||||
Ok(count) => ManagerResponse::PendingRemindersCount { count },
|
||||
Err(e) => ManagerResponse::Err {
|
||||
message: format!("{e:#}"),
|
||||
},
|
||||
}
|
||||
handle_manager_count_pending_reminders(coord, agent.as_deref())
|
||||
}
|
||||
ManagerRequest::ReminderRollup { since_secs, agent } => {
|
||||
let target = agent.as_deref().unwrap_or(MANAGER_AGENT);
|
||||
match coord.broker.reminder_rollup_for(target, *since_secs) {
|
||||
Ok(stats) => ManagerResponse::ReminderRollup(stats),
|
||||
Err(e) => ManagerResponse::Err {
|
||||
message: format!("{e:#}"),
|
||||
},
|
||||
}
|
||||
handle_manager_reminder_rollup(coord, agent.as_deref(), *since_secs)
|
||||
}
|
||||
_ => ManagerResponse::Err {
|
||||
message: "request not handled on manager socket".to_owned(),
|
||||
|
|
@ -321,6 +143,245 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
|
|||
}
|
||||
}
|
||||
|
||||
/// `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");
|
||||
match submit_init_config(coord, name, 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
|
||||
|
|
|
|||
Loading…
Reference in a new issue