manager: receive forge notifications (ManagerRequest::Wake)
This commit is contained in:
parent
d5009cd175
commit
3b44410427
4 changed files with 46 additions and 11 deletions
|
|
@ -82,6 +82,7 @@ async fn main() -> Result<()> {
|
||||||
files.clone(),
|
files.clone(),
|
||||||
turn_lock.clone(),
|
turn_lock.clone(),
|
||||||
));
|
));
|
||||||
|
tokio::spawn(hive_ag3nt::forge_notify::run(cli.socket.clone(), true));
|
||||||
match initial {
|
match initial {
|
||||||
LoginState::Online => {
|
LoginState::Online => {
|
||||||
serve(
|
serve(
|
||||||
|
|
|
||||||
|
|
@ -27,10 +27,15 @@ const HTTP_TIMEOUT_SECS: u64 = 10;
|
||||||
/// Maximum characters of a body/comment to include in the wake message.
|
/// Maximum characters of a body/comment to include in the wake message.
|
||||||
const BODY_TRUNCATE: usize = 500;
|
const BODY_TRUNCATE: usize = 500;
|
||||||
|
|
||||||
/// Spawn point: called once from `hive-ag3nt serve`. Returns immediately if
|
/// Spawn point: called once from `hive-ag3nt serve` (agent) or
|
||||||
/// the forge is not configured for this agent. Otherwise loops forever,
|
/// `hive-m1nd serve` (manager). Returns immediately if the forge is not
|
||||||
/// polling every `POLL_INTERVAL_SECS` seconds. Errors are never fatal.
|
/// configured. Otherwise loops forever, polling every
|
||||||
pub async fn run(socket: PathBuf) {
|
/// `POLL_INTERVAL_SECS` seconds. Errors are never fatal.
|
||||||
|
///
|
||||||
|
/// `is_manager`: when true, wakes the inbox via `ManagerRequest::Wake`
|
||||||
|
/// instead of `AgentRequest::Wake` (the manager socket rejects the agent
|
||||||
|
/// request type).
|
||||||
|
pub async fn run(socket: PathBuf, is_manager: bool) {
|
||||||
let forge_url = match std::env::var("HIVE_FORGE_URL") {
|
let forge_url = match std::env::var("HIVE_FORGE_URL") {
|
||||||
Ok(u) if !u.is_empty() => u,
|
Ok(u) if !u.is_empty() => u,
|
||||||
_ => {
|
_ => {
|
||||||
|
|
@ -77,7 +82,7 @@ pub async fn run(socket: PathBuf) {
|
||||||
|
|
||||||
loop {
|
loop {
|
||||||
interval.tick().await;
|
interval.tick().await;
|
||||||
poll_once(&client, &forge_url, &token, &socket).await;
|
poll_once(&client, &forge_url, &token, &socket, is_manager).await;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -208,7 +213,7 @@ async fn format_notification(
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn poll_once(client: &reqwest::Client, forge_url: &str, token: &str, socket: &Path) {
|
async fn poll_once(client: &reqwest::Client, forge_url: &str, token: &str, socket: &Path, is_manager: bool) {
|
||||||
let url = format!("{forge_url}/api/v1/notifications?all=false&limit=50");
|
let url = format!("{forge_url}/api/v1/notifications?all=false&limit=50");
|
||||||
let resp = match client
|
let resp = match client
|
||||||
.get(&url)
|
.get(&url)
|
||||||
|
|
@ -250,12 +255,25 @@ async fn poll_once(client: &reqwest::Client, forge_url: &str, token: &str, socke
|
||||||
|
|
||||||
let body = format_notification(client, token, notif).await;
|
let body = format_notification(client, token, notif).await;
|
||||||
|
|
||||||
let req = hive_sh4re::AgentRequest::Wake {
|
let delivered = if is_manager {
|
||||||
from: "forge".to_owned(),
|
let req = hive_sh4re::ManagerRequest::Wake {
|
||||||
body,
|
from: "forge".to_owned(),
|
||||||
|
body,
|
||||||
|
};
|
||||||
|
crate::client::request::<_, hive_sh4re::ManagerResponse>(socket, &req)
|
||||||
|
.await
|
||||||
|
.map(|_| ())
|
||||||
|
} else {
|
||||||
|
let req = hive_sh4re::AgentRequest::Wake {
|
||||||
|
from: "forge".to_owned(),
|
||||||
|
body,
|
||||||
|
};
|
||||||
|
crate::client::request::<_, hive_sh4re::AgentResponse>(socket, &req)
|
||||||
|
.await
|
||||||
|
.map(|_| ())
|
||||||
};
|
};
|
||||||
match crate::client::request::<_, hive_sh4re::AgentResponse>(socket, &req).await {
|
match delivered {
|
||||||
Ok(_) => {
|
Ok(()) => {
|
||||||
debug!(%id, "forge_notify: delivered");
|
debug!(%id, "forge_notify: delivered");
|
||||||
}
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
|
|
|
||||||
|
|
@ -117,6 +117,17 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
ManagerRequest::Wake { from, body } => match coord.broker.send(&Message {
|
||||||
|
from: from.clone(),
|
||||||
|
to: MANAGER_AGENT.to_owned(),
|
||||||
|
body: body.clone(),
|
||||||
|
in_reply_to: None,
|
||||||
|
}) {
|
||||||
|
Ok(()) => ManagerResponse::Ok,
|
||||||
|
Err(e) => ManagerResponse::Err {
|
||||||
|
message: format!("{e:#}"),
|
||||||
|
},
|
||||||
|
},
|
||||||
ManagerRequest::OperatorMsg { body } => match coord.broker.send(&Message {
|
ManagerRequest::OperatorMsg { body } => match coord.broker.send(&Message {
|
||||||
from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
|
from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
|
||||||
to: MANAGER_AGENT.to_owned(),
|
to: MANAGER_AGENT.to_owned(),
|
||||||
|
|
|
||||||
|
|
@ -837,6 +837,11 @@ pub enum ManagerRequest {
|
||||||
/// Mirror of `AgentRequest::RequeueInflight` on the manager
|
/// Mirror of `AgentRequest::RequeueInflight` on the manager
|
||||||
/// surface — fired exactly once on manager harness boot.
|
/// surface — fired exactly once on manager harness boot.
|
||||||
RequeueInflight,
|
RequeueInflight,
|
||||||
|
/// Mirror of `AgentRequest::Wake` on the manager surface. Used by
|
||||||
|
/// in-container background tasks (e.g. `forge_notify`) to push a
|
||||||
|
/// message into the manager's own broker inbox. `from` is caller-
|
||||||
|
/// chosen; `body` becomes the wake prompt body.
|
||||||
|
Wake { from: String, body: String },
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue