Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4989bcdb5e | ||
|
|
76647415af | ||
|
|
8c1979f05c | ||
|
|
3df565789c |
19 changed files with 158 additions and 107 deletions
3
Cargo.lock
generated
3
Cargo.lock
generated
|
|
@ -1556,6 +1556,7 @@ dependencies = [
|
||||||
"hive-agent-sock",
|
"hive-agent-sock",
|
||||||
"hive-core-agent-sock",
|
"hive-core-agent-sock",
|
||||||
"hive-sh4re",
|
"hive-sh4re",
|
||||||
|
"hive-types",
|
||||||
"rmcp",
|
"rmcp",
|
||||||
"serde",
|
"serde",
|
||||||
"serde_json",
|
"serde_json",
|
||||||
|
|
@ -1662,6 +1663,7 @@ name = "hive-core-agent-sock"
|
||||||
version = "0.1.0"
|
version = "0.1.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"hive-sh4re",
|
"hive-sh4re",
|
||||||
|
"hive-types",
|
||||||
"serde",
|
"serde",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
@ -1766,6 +1768,7 @@ name = "hive-sh4re"
|
||||||
version = "0.1.0"
|
version = "0.1.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"chrono",
|
"chrono",
|
||||||
|
"hive-types",
|
||||||
"schemars",
|
"schemars",
|
||||||
"serde",
|
"serde",
|
||||||
"serde_json",
|
"serde_json",
|
||||||
|
|
|
||||||
|
|
@ -17,6 +17,7 @@ clap.workspace = true
|
||||||
hive-agent-sock.workspace = true
|
hive-agent-sock.workspace = true
|
||||||
hive-core-agent-sock.workspace = true
|
hive-core-agent-sock.workspace = true
|
||||||
hive-sh4re.workspace = true
|
hive-sh4re.workspace = true
|
||||||
|
hive-types.workspace = true
|
||||||
rmcp.workspace = true
|
rmcp.workspace = true
|
||||||
serde.workspace = true
|
serde.workspace = true
|
||||||
serde_json.workspace = true
|
serde_json.workspace = true
|
||||||
|
|
|
||||||
|
|
@ -181,13 +181,17 @@ impl AgentServer {
|
||||||
async fn ask(&self, Parameters(args): Parameters<AskArgs>) -> String {
|
async fn ask(&self, Parameters(args): Parameters<AskArgs>) -> String {
|
||||||
let log = format!("{args:?}");
|
let log = format!("{args:?}");
|
||||||
run_tool_envelope("ask", log, async move {
|
run_tool_envelope("ask", log, async move {
|
||||||
|
let to = match args.to.map(|t| hive_types::Ident::parse(&t)).transpose() {
|
||||||
|
Ok(to) => to,
|
||||||
|
Err(reason) => return format!("invalid `to` agent name: {reason}"),
|
||||||
|
};
|
||||||
let (resp, retries) = self
|
let (resp, retries) = self
|
||||||
.dispatch(hive_core_agent_sock::Request::Ask {
|
.dispatch(hive_core_agent_sock::Request::Ask {
|
||||||
question: args.question,
|
question: args.question,
|
||||||
options: args.options,
|
options: args.options,
|
||||||
multi: args.multi,
|
multi: args.multi,
|
||||||
ttl_seconds: args.ttl_seconds,
|
ttl_seconds: args.ttl_seconds,
|
||||||
to: args.to,
|
to,
|
||||||
})
|
})
|
||||||
.await;
|
.await;
|
||||||
let s = match resp {
|
let s = match resp {
|
||||||
|
|
@ -388,8 +392,12 @@ impl AgentServer {
|
||||||
async fn get_agent_meta(&self, Parameters(args): Parameters<GetAgentMetaArgs>) -> String {
|
async fn get_agent_meta(&self, Parameters(args): Parameters<GetAgentMetaArgs>) -> String {
|
||||||
let log = args.name.clone().unwrap_or_else(|| "<self>".to_owned());
|
let log = args.name.clone().unwrap_or_else(|| "<self>".to_owned());
|
||||||
run_tool_envelope("get_agent_meta", log, async move {
|
run_tool_envelope("get_agent_meta", log, async move {
|
||||||
|
let name = match args.name.map(|n| hive_types::Ident::parse(&n)).transpose() {
|
||||||
|
Ok(name) => name,
|
||||||
|
Err(reason) => return format!("invalid agent name: {reason}"),
|
||||||
|
};
|
||||||
let (resp, retries) = self
|
let (resp, retries) = self
|
||||||
.dispatch(hive_core_agent_sock::Request::GetAgentMeta { name: args.name })
|
.dispatch(hive_core_agent_sock::Request::GetAgentMeta { name })
|
||||||
.await;
|
.await;
|
||||||
annotate_retries(format_agent_meta(resp), retries)
|
annotate_retries(format_agent_meta(resp), retries)
|
||||||
})
|
})
|
||||||
|
|
|
||||||
|
|
@ -41,12 +41,9 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
|
||||||
// Sub-second git seed + forge-remote wire. Routing through
|
// Sub-second git seed + forge-remote wire. Routing through
|
||||||
// the queue would surface a queue card that's gone before
|
// the queue would surface a queue card that's gone before
|
||||||
// the operator's eyes refocus. Run inline.
|
// the operator's eyes refocus. Run inline.
|
||||||
let agent = hive_types::Ident::parse(&approval.agent).map_err(|e| {
|
let proposed_dir = Coordinator::agent_proposed_dir(&approval.agent);
|
||||||
anyhow::anyhow!("approval {} has invalid agent name: {e}", approval.id)
|
let claude_dir = Coordinator::agent_claude_dir(&approval.agent);
|
||||||
})?;
|
let notes_dir = Coordinator::agent_notes_dir(&approval.agent);
|
||||||
let proposed_dir = Coordinator::agent_proposed_dir(&agent);
|
|
||||||
let claude_dir = Coordinator::agent_claude_dir(&agent);
|
|
||||||
let notes_dir = Coordinator::agent_notes_dir(&agent);
|
|
||||||
run_approval_init_config(&coord, approval, proposed_dir, claude_dir, notes_dir).await
|
run_approval_init_config(&coord, approval, proposed_dir, claude_dir, notes_dir).await
|
||||||
}
|
}
|
||||||
ApprovalKind::UpdateMetaInputs => {
|
ApprovalKind::UpdateMetaInputs => {
|
||||||
|
|
@ -75,11 +72,14 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
|
||||||
ApprovalKind::Spawn => {
|
ApprovalKind::Spawn => {
|
||||||
// The spawn's tail `Reconcile` starts the container, so the
|
// The spawn's tail `Reconcile` starts the container, so the
|
||||||
// new agent's power intent is `Up` from the outset.
|
// new agent's power intent is `Up` from the outset.
|
||||||
if let Err(e) = coord.power.set(&approval.agent, crate::power::Wanted::Up) {
|
if let Err(e) = coord
|
||||||
|
.power
|
||||||
|
.set(approval.agent.as_str(), crate::power::Wanted::Up)
|
||||||
|
{
|
||||||
tracing::warn!(agent = %approval.agent, error = ?e, "agent_power: seed on spawn failed");
|
tracing::warn!(agent = %approval.agent, error = ?e, "agent_power: seed on spawn failed");
|
||||||
}
|
}
|
||||||
let submitted = coord.job_queue.submit(crate::job_queue::templates::spawn(
|
let submitted = coord.job_queue.submit(crate::job_queue::templates::spawn(
|
||||||
&approval.agent,
|
approval.agent.as_str(),
|
||||||
id,
|
id,
|
||||||
format!("approval #{id} spawn"),
|
format!("approval #{id} spawn"),
|
||||||
));
|
));
|
||||||
|
|
@ -110,7 +110,7 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
|
||||||
// deploy tail).
|
// deploy tail).
|
||||||
enqueue_approval_rebuild(
|
enqueue_approval_rebuild(
|
||||||
&coord,
|
&coord,
|
||||||
&approval.agent,
|
approval.agent.as_str(),
|
||||||
id,
|
id,
|
||||||
format!("approval #{id} merge config pr"),
|
format!("approval #{id} merge config pr"),
|
||||||
);
|
);
|
||||||
|
|
@ -158,8 +158,8 @@ pub async fn run_approval_merge_config_pr(
|
||||||
approval_id: i64,
|
approval_id: i64,
|
||||||
) -> Result<()> {
|
) -> Result<()> {
|
||||||
let approval = fetch_approval_for_worker(coord, approval_id, ApprovalKind::MergeConfigPr)?;
|
let approval = fetch_approval_for_worker(coord, approval_id, ApprovalKind::MergeConfigPr)?;
|
||||||
let agent_dir = crate::paths::agent_runtime_dir(&approval.agent);
|
let agent_dir = crate::paths::agent_runtime_dir(approval.agent.as_str());
|
||||||
let applied_dir = crate::paths::applied_dir(&approval.agent);
|
let applied_dir = crate::paths::applied_dir(approval.agent.as_str());
|
||||||
// Captured up front to scope the failure-comment's build-log lookup to
|
// Captured up front to scope the failure-comment's build-log lookup to
|
||||||
// rows this deploy produced (see `post_merge_failure_to_pr`).
|
// rows this deploy produced (see `post_merge_failure_to_pr`).
|
||||||
let since_ts = hive_sh4re::wire_time::now_unix();
|
let since_ts = hive_sh4re::wire_time::now_unix();
|
||||||
|
|
@ -172,7 +172,7 @@ pub async fn run_approval_merge_config_pr(
|
||||||
// ff'd by the merge, so only the tag refspec actually lands; best-effort,
|
// ff'd by the merge, so only the tag refspec actually lands; best-effort,
|
||||||
// never fails the approval.
|
// never fails the approval.
|
||||||
coord.set_queue_step(queue_entry_id, "forge push");
|
coord.set_queue_step(queue_entry_id, "forge push");
|
||||||
if let Err(e) = crate::forge::push_config(&approval.agent).await {
|
if let Err(e) = crate::forge::push_config(approval.agent.as_str()).await {
|
||||||
tracing::warn!(agent = %approval.agent, error = ?e, "forge: push_config after merge failed");
|
tracing::warn!(agent = %approval.agent, error = ?e, "forge: push_config after merge failed");
|
||||||
}
|
}
|
||||||
// On a failed deploy, surface the failing build log back onto the PR so
|
// On a failed deploy, surface the failing build log back onto the PR so
|
||||||
|
|
@ -208,11 +208,11 @@ async fn post_merge_failure_to_pr(
|
||||||
let Ok(pr) = approval.commit_ref.parse::<u64>() else {
|
let Ok(pr) = approval.commit_ref.parse::<u64>() else {
|
||||||
return;
|
return;
|
||||||
};
|
};
|
||||||
let repo = crate::forge::config_repo(&approval.agent);
|
let repo = crate::forge::config_repo(approval.agent.as_str());
|
||||||
|
|
||||||
let log_section = coord
|
let log_section = coord
|
||||||
.build_logs
|
.build_logs
|
||||||
.list_recent_for_agent(&approval.agent, 10)
|
.list_recent_for_agent(approval.agent.as_str(), 10)
|
||||||
.ok()
|
.ok()
|
||||||
.and_then(|rows| {
|
.and_then(|rows| {
|
||||||
rows.into_iter()
|
rows.into_iter()
|
||||||
|
|
@ -300,7 +300,7 @@ async fn run_merge_config_pr(
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
let repo = crate::forge::config_repo(&approval.agent);
|
let repo = crate::forge::config_repo(approval.agent.as_str());
|
||||||
|
|
||||||
// 1. Drift gate: the live PR head must still equal what was reviewed.
|
// 1. Drift gate: the live PR head must still equal what was reviewed.
|
||||||
coord.set_queue_step(queue_entry_id, "verify PR head");
|
coord.set_queue_step(queue_entry_id, "verify PR head");
|
||||||
|
|
@ -328,7 +328,9 @@ async fn run_merge_config_pr(
|
||||||
|
|
||||||
// 3. Eval-verify BEFORE the irreversible push (bad nix fails fast here).
|
// 3. Eval-verify BEFORE the irreversible push (bad nix fails fast here).
|
||||||
coord.set_queue_step(queue_entry_id, "verify proposal (eval)");
|
coord.set_queue_step(queue_entry_id, "verify proposal (eval)");
|
||||||
if let Err(e) = crate::meta::verify_commit(&approval.agent, applied_dir, &reviewed).await {
|
if let Err(e) =
|
||||||
|
crate::meta::verify_commit(approval.agent.as_str(), applied_dir, &reviewed).await
|
||||||
|
{
|
||||||
return (
|
return (
|
||||||
Err(anyhow::anyhow!("verify merge head {reviewed}: {e:#}")),
|
Err(anyhow::anyhow!("verify merge head {reviewed}: {e:#}")),
|
||||||
None,
|
None,
|
||||||
|
|
@ -364,7 +366,7 @@ async fn run_merge_config_pr(
|
||||||
// 5. Deploy tail. target == finalize == the reviewed head.
|
// 5. Deploy tail. target == finalize == the reviewed head.
|
||||||
deploy_applied_target(
|
deploy_applied_target(
|
||||||
coord,
|
coord,
|
||||||
&approval.agent,
|
approval.agent.as_str(),
|
||||||
agent_dir,
|
agent_dir,
|
||||||
applied_dir,
|
applied_dir,
|
||||||
&reviewed,
|
&reviewed,
|
||||||
|
|
@ -391,7 +393,7 @@ async fn run_approval_schedule_prompt(
|
||||||
coord
|
coord
|
||||||
.scheduled_prompts
|
.scheduled_prompts
|
||||||
.submit(&crate::scheduled_prompts::NewSchedule {
|
.submit(&crate::scheduled_prompts::NewSchedule {
|
||||||
owner: approval.agent.clone(),
|
owner: approval.agent.to_string(),
|
||||||
targets: payload.targets,
|
targets: payload.targets,
|
||||||
body: payload.body,
|
body: payload.body,
|
||||||
first_fire_at_unix: payload.first_fire_at_unix,
|
first_fire_at_unix: payload.first_fire_at_unix,
|
||||||
|
|
@ -451,7 +453,7 @@ pub(crate) async fn resolve_approval_dag(
|
||||||
// access) — warn-only, then the resolution events + a rescan so
|
// access) — warn-only, then the resolution events + a rescan so
|
||||||
// the dashboard reflects the post-spawn state either way.
|
// the dashboard reflects the post-spawn state either way.
|
||||||
if result.is_ok() {
|
if result.is_ok() {
|
||||||
forge_after_first_spawn(coord, &approval.agent).await;
|
forge_after_first_spawn(coord, approval.agent.as_str()).await;
|
||||||
} else {
|
} else {
|
||||||
coord.rescan_containers_and_emit().await;
|
coord.rescan_containers_and_emit().await;
|
||||||
crate::dashboard::emit_tombstones_snapshot(coord).await;
|
crate::dashboard::emit_tombstones_snapshot(coord).await;
|
||||||
|
|
@ -528,7 +530,7 @@ async fn run_approval_init_config(
|
||||||
// and let `topology::reconcile` assign the default position on
|
// and let `topology::reconcile` assign the default position on
|
||||||
// first spawn, so this path never names a specific root agent.
|
// first spawn, so this path never names a specific root agent.
|
||||||
if !approval.commit_ref.is_empty() {
|
if !approval.commit_ref.is_empty() {
|
||||||
crate::topology::add_child(&approval.agent, &approval.commit_ref)
|
crate::topology::add_child(approval.agent.as_str(), &approval.commit_ref)
|
||||||
.map_err(|e| anyhow::anyhow!("topology add_child: {e}"))?;
|
.map_err(|e| anyhow::anyhow!("topology add_child: {e}"))?;
|
||||||
}
|
}
|
||||||
// Create the agent's state root as a btrfs subvolume FIRST, before
|
// Create the agent's state root as a btrfs subvolume FIRST, before
|
||||||
|
|
@ -538,15 +540,15 @@ async fn run_approval_init_config(
|
||||||
// materialise the state root as a plain directory — after which
|
// materialise the state root as a plain directory — after which
|
||||||
// the subvolume create is silently skipped and the agent never
|
// the subvolume create is silently skipped and the agent never
|
||||||
// lands on a subvolume (no quota, no snapshot). Order matters.
|
// lands on a subvolume (no quota, no snapshot). Order matters.
|
||||||
lifecycle::ensure_agent_state_subvolume(&approval.agent).await?;
|
lifecycle::ensure_agent_state_subvolume(approval.agent.as_str()).await?;
|
||||||
lifecycle::setup_proposed(&proposed_dir, &approval.agent).await?;
|
lifecycle::setup_proposed(&proposed_dir, approval.agent.as_str()).await?;
|
||||||
lifecycle::ensure_claude_dir(&claude_dir)?;
|
lifecycle::ensure_claude_dir(&claude_dir)?;
|
||||||
lifecycle::ensure_state_dir(¬es_dir)?;
|
lifecycle::ensure_state_dir(¬es_dir)?;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
.await;
|
.await;
|
||||||
if result.is_ok()
|
if result.is_ok()
|
||||||
&& let Err(e) = crate::forge::ensure_meta_remote(&approval.agent).await
|
&& let Err(e) = crate::forge::ensure_meta_remote(approval.agent.as_str()).await
|
||||||
{
|
{
|
||||||
tracing::warn!(agent = %approval.agent, error = ?e, "forge: ensure_meta_remote after init_config failed");
|
tracing::warn!(agent = %approval.agent, error = ?e, "forge: ensure_meta_remote after init_config failed");
|
||||||
}
|
}
|
||||||
|
|
@ -571,7 +573,7 @@ fn finish_approval(
|
||||||
approval.id,
|
approval.id,
|
||||||
&HelperEvent::ApprovalResolved {
|
&HelperEvent::ApprovalResolved {
|
||||||
id: approval.id,
|
id: approval.id,
|
||||||
agent: approval.agent.clone(),
|
agent: approval.agent.to_string(),
|
||||||
commit_ref: approval.commit_ref.clone(),
|
commit_ref: approval.commit_ref.clone(),
|
||||||
status,
|
status,
|
||||||
note: note.clone(),
|
note: note.clone(),
|
||||||
|
|
@ -592,7 +594,7 @@ fn finish_approval(
|
||||||
let status_str = if ok { "approved" } else { "failed" };
|
let status_str = if ok { "approved" } else { "failed" };
|
||||||
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
|
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
|
||||||
id: approval.id,
|
id: approval.id,
|
||||||
agent: &approval.agent,
|
agent: approval.agent.as_str(),
|
||||||
approval_kind,
|
approval_kind,
|
||||||
sha_short,
|
sha_short,
|
||||||
status: status_str,
|
status: status_str,
|
||||||
|
|
@ -610,7 +612,7 @@ fn finish_approval(
|
||||||
coord.notify_submitter(
|
coord.notify_submitter(
|
||||||
approval.id,
|
approval.id,
|
||||||
&HelperEvent::ConfigReady {
|
&HelperEvent::ConfigReady {
|
||||||
agent: approval.agent.clone(),
|
agent: approval.agent.to_string(),
|
||||||
},
|
},
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
@ -618,7 +620,7 @@ fn finish_approval(
|
||||||
ApprovalKind::Spawn => coord.notify_submitter(
|
ApprovalKind::Spawn => coord.notify_submitter(
|
||||||
approval.id,
|
approval.id,
|
||||||
&HelperEvent::Spawned {
|
&HelperEvent::Spawned {
|
||||||
agent: approval.agent.clone(),
|
agent: approval.agent.to_string(),
|
||||||
ok,
|
ok,
|
||||||
note,
|
note,
|
||||||
},
|
},
|
||||||
|
|
@ -630,7 +632,7 @@ fn finish_approval(
|
||||||
coord.notify_submitter(
|
coord.notify_submitter(
|
||||||
approval.id,
|
approval.id,
|
||||||
&HelperEvent::Rebuilt {
|
&HelperEvent::Rebuilt {
|
||||||
agent: approval.agent.clone(),
|
agent: approval.agent.to_string(),
|
||||||
ok,
|
ok,
|
||||||
note,
|
note,
|
||||||
sha: approval.fetched_sha.clone(),
|
sha: approval.fetched_sha.clone(),
|
||||||
|
|
@ -881,7 +883,7 @@ pub fn deny(coord: &Coordinator, id: i64, note: Option<&str>) -> Result<()> {
|
||||||
a.id,
|
a.id,
|
||||||
&HelperEvent::ApprovalResolved {
|
&HelperEvent::ApprovalResolved {
|
||||||
id: a.id,
|
id: a.id,
|
||||||
agent: a.agent,
|
agent: a.agent.to_string(),
|
||||||
commit_ref: a.commit_ref,
|
commit_ref: a.commit_ref,
|
||||||
status: ApprovalStatus::Denied,
|
status: ApprovalStatus::Denied,
|
||||||
note: note.map(String::from),
|
note: note.map(String::from),
|
||||||
|
|
@ -892,7 +894,7 @@ pub fn deny(coord: &Coordinator, id: i64, note: Option<&str>) -> Result<()> {
|
||||||
);
|
);
|
||||||
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
|
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
|
||||||
id,
|
id,
|
||||||
agent: &agent_owned,
|
agent: agent_owned.as_str(),
|
||||||
approval_kind,
|
approval_kind,
|
||||||
sha_short,
|
sha_short,
|
||||||
status: "denied",
|
status: "denied",
|
||||||
|
|
|
||||||
|
|
@ -1025,7 +1025,7 @@ impl Coordinator {
|
||||||
// doesn't bubble out and unwind the topology write.
|
// doesn't bubble out and unwind the topology write.
|
||||||
if let Some(op) = old_parent.as_deref() {
|
if let Some(op) = old_parent.as_deref() {
|
||||||
let _ = self.broker.send(&hive_sh4re::Message {
|
let _ = self.broker.send(&hive_sh4re::Message {
|
||||||
from: hive_sh4re::SYSTEM_SENDER.to_owned(),
|
from: hive_sh4re::trusted_sender(hive_sh4re::SYSTEM_SENDER),
|
||||||
to: op.to_owned(),
|
to: op.to_owned(),
|
||||||
body: format!("{child} moved out of your subtree to {new_label}"),
|
body: format!("{child} moved out of your subtree to {new_label}"),
|
||||||
in_reply_to: None,
|
in_reply_to: None,
|
||||||
|
|
@ -1033,7 +1033,7 @@ impl Coordinator {
|
||||||
}
|
}
|
||||||
if let Some(np) = new_parent {
|
if let Some(np) = new_parent {
|
||||||
let _ = self.broker.send(&hive_sh4re::Message {
|
let _ = self.broker.send(&hive_sh4re::Message {
|
||||||
from: hive_sh4re::SYSTEM_SENDER.to_owned(),
|
from: hive_sh4re::trusted_sender(hive_sh4re::SYSTEM_SENDER),
|
||||||
to: np.to_owned(),
|
to: np.to_owned(),
|
||||||
body: format!(
|
body: format!(
|
||||||
"{child} just moved into your subtree (was previously under {old_label})"
|
"{child} just moved into your subtree (was previously under {old_label})"
|
||||||
|
|
@ -1087,7 +1087,7 @@ impl Coordinator {
|
||||||
let new_label = new_parent.unwrap_or("<root>");
|
let new_label = new_parent.unwrap_or("<root>");
|
||||||
if let Some(op) = old_parent.as_deref() {
|
if let Some(op) = old_parent.as_deref() {
|
||||||
let _ = self.broker.send(&hive_sh4re::Message {
|
let _ = self.broker.send(&hive_sh4re::Message {
|
||||||
from: hive_sh4re::SYSTEM_SENDER.to_owned(),
|
from: hive_sh4re::trusted_sender(hive_sh4re::SYSTEM_SENDER),
|
||||||
to: op.to_owned(),
|
to: op.to_owned(),
|
||||||
body: format!("{child} moved out of your subtree to {new_label}"),
|
body: format!("{child} moved out of your subtree to {new_label}"),
|
||||||
in_reply_to: None,
|
in_reply_to: None,
|
||||||
|
|
@ -1095,7 +1095,7 @@ impl Coordinator {
|
||||||
}
|
}
|
||||||
if let Some(np) = new_parent {
|
if let Some(np) = new_parent {
|
||||||
let _ = self.broker.send(&hive_sh4re::Message {
|
let _ = self.broker.send(&hive_sh4re::Message {
|
||||||
from: hive_sh4re::SYSTEM_SENDER.to_owned(),
|
from: hive_sh4re::trusted_sender(hive_sh4re::SYSTEM_SENDER),
|
||||||
to: np.to_owned(),
|
to: np.to_owned(),
|
||||||
body: format!(
|
body: format!(
|
||||||
"{child} just moved into your subtree (was previously under {old_label})"
|
"{child} just moved into your subtree (was previously under {old_label})"
|
||||||
|
|
@ -1349,7 +1349,7 @@ impl Coordinator {
|
||||||
still in your window."
|
still in your window."
|
||||||
);
|
);
|
||||||
if let Err(e) = self.broker.send(&hive_sh4re::Message {
|
if let Err(e) = self.broker.send(&hive_sh4re::Message {
|
||||||
from: hive_sh4re::SYSTEM_SENDER.to_owned(),
|
from: hive_sh4re::trusted_sender(hive_sh4re::SYSTEM_SENDER),
|
||||||
to: name.to_owned(),
|
to: name.to_owned(),
|
||||||
body,
|
body,
|
||||||
in_reply_to: None,
|
in_reply_to: None,
|
||||||
|
|
@ -1402,7 +1402,7 @@ impl Coordinator {
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
if let Err(e) = self.broker.send(&hive_sh4re::Message {
|
if let Err(e) = self.broker.send(&hive_sh4re::Message {
|
||||||
from: from.to_owned(),
|
from: hive_sh4re::trusted_sender(from),
|
||||||
to: agent.to_owned(),
|
to: agent.to_owned(),
|
||||||
body,
|
body,
|
||||||
in_reply_to: None,
|
in_reply_to: None,
|
||||||
|
|
@ -1424,7 +1424,7 @@ impl Coordinator {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
if let Err(e) = self.broker.send(&hive_sh4re::Message {
|
if let Err(e) = self.broker.send(&hive_sh4re::Message {
|
||||||
from: from.to_owned(),
|
from: hive_sh4re::trusted_sender(from),
|
||||||
to: agent_name.clone(),
|
to: agent_name.clone(),
|
||||||
body: broadcast_body.clone(),
|
body: broadcast_body.clone(),
|
||||||
in_reply_to: None,
|
in_reply_to: None,
|
||||||
|
|
|
||||||
|
|
@ -66,12 +66,7 @@ pub(super) fn gc_orphans(coord: &Coordinator, approvals: Vec<Approval>) -> Vec<A
|
||||||
) {
|
) {
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
let Ok(agent) = hive_types::Ident::parse(&a.agent) else {
|
if Coordinator::agent_proposed_dir(&a.agent).exists() {
|
||||||
let _ = coord.approvals.mark_failed(a.id, "invalid agent name");
|
|
||||||
tracing::warn!(id = a.id, agent = %a.agent, "auto-failed approval with invalid agent name");
|
|
||||||
return false;
|
|
||||||
};
|
|
||||||
if Coordinator::agent_proposed_dir(&agent).exists() {
|
|
||||||
true
|
true
|
||||||
} else {
|
} else {
|
||||||
let note = "agent state dir missing";
|
let note = "agent state dir missing";
|
||||||
|
|
@ -83,7 +78,7 @@ pub(super) fn gc_orphans(coord: &Coordinator, approvals: Vec<Approval>) -> Vec<A
|
||||||
.map(|s| s[..s.len().min(12)].to_owned());
|
.map(|s| s[..s.len().min(12)].to_owned());
|
||||||
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
|
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
|
||||||
id: a.id,
|
id: a.id,
|
||||||
agent: &a.agent,
|
agent: a.agent.as_str(),
|
||||||
approval_kind: a.kind.as_str(),
|
approval_kind: a.kind.as_str(),
|
||||||
sha_short,
|
sha_short,
|
||||||
status: "failed",
|
status: "failed",
|
||||||
|
|
|
||||||
|
|
@ -164,7 +164,7 @@ pub(super) async fn post_op_send(
|
||||||
));
|
));
|
||||||
}
|
}
|
||||||
} else if let Err(e) = state.coord.broker.send(&hive_sh4re::Message {
|
} else if let Err(e) = state.coord.broker.send(&hive_sh4re::Message {
|
||||||
from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
|
from: hive_sh4re::trusted_sender(hive_sh4re::OPERATOR_RECIPIENT),
|
||||||
to: to.clone(),
|
to: to.clone(),
|
||||||
body,
|
body,
|
||||||
in_reply_to: None,
|
in_reply_to: None,
|
||||||
|
|
|
||||||
|
|
@ -552,7 +552,7 @@ fn history_view(a: Approval) -> ApprovalHistoryView {
|
||||||
let kind = a.kind.as_str();
|
let kind = a.kind.as_str();
|
||||||
ApprovalHistoryView {
|
ApprovalHistoryView {
|
||||||
id: a.id,
|
id: a.id,
|
||||||
agent: a.agent,
|
agent: a.agent.to_string(),
|
||||||
kind,
|
kind,
|
||||||
sha_short,
|
sha_short,
|
||||||
status,
|
status,
|
||||||
|
|
@ -567,7 +567,7 @@ fn build_approval_views(approvals: Vec<Approval>) -> Vec<ApprovalView> {
|
||||||
out.push(match a.kind {
|
out.push(match a.kind {
|
||||||
hive_sh4re::ApprovalKind::Spawn => ApprovalView {
|
hive_sh4re::ApprovalKind::Spawn => ApprovalView {
|
||||||
id: a.id,
|
id: a.id,
|
||||||
agent: a.agent,
|
agent: a.agent.to_string(),
|
||||||
kind: "spawn",
|
kind: "spawn",
|
||||||
sha_short: None,
|
sha_short: None,
|
||||||
description: a.description,
|
description: a.description,
|
||||||
|
|
@ -577,7 +577,7 @@ fn build_approval_views(approvals: Vec<Approval>) -> Vec<ApprovalView> {
|
||||||
},
|
},
|
||||||
hive_sh4re::ApprovalKind::InitConfig => ApprovalView {
|
hive_sh4re::ApprovalKind::InitConfig => ApprovalView {
|
||||||
id: a.id,
|
id: a.id,
|
||||||
agent: a.agent,
|
agent: a.agent.to_string(),
|
||||||
kind: "init_config",
|
kind: "init_config",
|
||||||
sha_short: None,
|
sha_short: None,
|
||||||
description: a.description,
|
description: a.description,
|
||||||
|
|
@ -587,7 +587,7 @@ fn build_approval_views(approvals: Vec<Approval>) -> Vec<ApprovalView> {
|
||||||
},
|
},
|
||||||
hive_sh4re::ApprovalKind::UpdateMetaInputs => ApprovalView {
|
hive_sh4re::ApprovalKind::UpdateMetaInputs => ApprovalView {
|
||||||
id: a.id,
|
id: a.id,
|
||||||
agent: a.agent,
|
agent: a.agent.to_string(),
|
||||||
kind: "update_meta_inputs",
|
kind: "update_meta_inputs",
|
||||||
sha_short: None,
|
sha_short: None,
|
||||||
description: a.description,
|
description: a.description,
|
||||||
|
|
@ -597,7 +597,7 @@ fn build_approval_views(approvals: Vec<Approval>) -> Vec<ApprovalView> {
|
||||||
},
|
},
|
||||||
hive_sh4re::ApprovalKind::SchedulePrompt => ApprovalView {
|
hive_sh4re::ApprovalKind::SchedulePrompt => ApprovalView {
|
||||||
id: a.id,
|
id: a.id,
|
||||||
agent: a.agent,
|
agent: a.agent.to_string(),
|
||||||
kind: "schedule_prompt",
|
kind: "schedule_prompt",
|
||||||
sha_short: None,
|
sha_short: None,
|
||||||
description: a.description,
|
description: a.description,
|
||||||
|
|
@ -618,7 +618,7 @@ fn build_approval_views(approvals: Vec<Approval>) -> Vec<ApprovalView> {
|
||||||
let pr_number = a.commit_ref.parse::<u64>().ok();
|
let pr_number = a.commit_ref.parse::<u64>().ok();
|
||||||
ApprovalView {
|
ApprovalView {
|
||||||
id: a.id,
|
id: a.id,
|
||||||
agent: a.agent,
|
agent: a.agent.to_string(),
|
||||||
kind: "merge_config_pr",
|
kind: "merge_config_pr",
|
||||||
sha_short: sha,
|
sha_short: sha,
|
||||||
description: a.description,
|
description: a.description,
|
||||||
|
|
|
||||||
|
|
@ -145,13 +145,15 @@ fn reconcile_stale_config_pr_approvals(
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
for a in pending {
|
for a in pending {
|
||||||
if a.kind != hive_sh4re::ApprovalKind::MergeConfigPr || !scanned_agents.contains(&a.agent) {
|
if a.kind != hive_sh4re::ApprovalKind::MergeConfigPr
|
||||||
|
|| !scanned_agents.contains(a.agent.as_str())
|
||||||
|
{
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
let Ok(pr_number) = a.commit_ref.parse::<u64>() else {
|
let Ok(pr_number) = a.commit_ref.parse::<u64>() else {
|
||||||
continue;
|
continue;
|
||||||
};
|
};
|
||||||
if open_prs.contains(&(a.agent.clone(), pr_number)) {
|
if open_prs.contains(&(a.agent.to_string(), pr_number)) {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
match coord
|
match coord
|
||||||
|
|
@ -165,7 +167,7 @@ fn reconcile_stale_config_pr_approvals(
|
||||||
);
|
);
|
||||||
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
|
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
|
||||||
id: a.id,
|
id: a.id,
|
||||||
agent: &a.agent,
|
agent: a.agent.as_str(),
|
||||||
approval_kind: "merge_config_pr",
|
approval_kind: "merge_config_pr",
|
||||||
sha_short: a
|
sha_short: a
|
||||||
.fetched_sha
|
.fetched_sha
|
||||||
|
|
|
||||||
|
|
@ -66,7 +66,7 @@ pub fn for_agent(coord: &Coordinator, agent: &str) -> Result<Vec<LooseEnd>> {
|
||||||
}
|
}
|
||||||
out.push(LooseEnd::Approval {
|
out.push(LooseEnd::Approval {
|
||||||
id: a.id,
|
id: a.id,
|
||||||
agent: a.agent,
|
agent: a.agent.to_string(),
|
||||||
commit_ref: a.commit_ref,
|
commit_ref: a.commit_ref,
|
||||||
description: a.description,
|
description: a.description,
|
||||||
age_seconds: saturating_age(now, a.requested_at.timestamp()),
|
age_seconds: saturating_age(now, a.requested_at.timestamp()),
|
||||||
|
|
@ -110,7 +110,7 @@ pub fn hive_wide(coord: &Coordinator) -> Result<Vec<LooseEnd>> {
|
||||||
for a in coord.approvals.pending()? {
|
for a in coord.approvals.pending()? {
|
||||||
out.push(LooseEnd::Approval {
|
out.push(LooseEnd::Approval {
|
||||||
id: a.id,
|
id: a.id,
|
||||||
agent: a.agent,
|
agent: a.agent.to_string(),
|
||||||
commit_ref: a.commit_ref,
|
commit_ref: a.commit_ref,
|
||||||
description: a.description,
|
description: a.description,
|
||||||
age_seconds: saturating_age(now, a.requested_at.timestamp()),
|
age_seconds: saturating_age(now, a.requested_at.timestamp()),
|
||||||
|
|
|
||||||
|
|
@ -227,7 +227,7 @@ pub fn handle_cancel_loose_end(
|
||||||
.map(|s| s[..s.len().min(12)].to_owned());
|
.map(|s| s[..s.len().min(12)].to_owned());
|
||||||
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
|
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
|
||||||
id: approval.id,
|
id: approval.id,
|
||||||
agent: &approval.agent,
|
agent: approval.agent.as_str(),
|
||||||
approval_kind: approval.kind.as_str(),
|
approval_kind: approval.kind.as_str(),
|
||||||
sha_short,
|
sha_short,
|
||||||
status: "cancelled",
|
status: "cancelled",
|
||||||
|
|
|
||||||
|
|
@ -222,7 +222,7 @@ pub(crate) async fn dispatch_shared(
|
||||||
options,
|
options,
|
||||||
*multi,
|
*multi,
|
||||||
*ttl_seconds,
|
*ttl_seconds,
|
||||||
to.as_deref(),
|
to.as_ref().map(hive_types::Ident::as_str),
|
||||||
)
|
)
|
||||||
.map_or_else(
|
.map_or_else(
|
||||||
|message| hive_core_agent_sock::Response::Err { message },
|
|message| hive_core_agent_sock::Response::Err { message },
|
||||||
|
|
@ -241,7 +241,7 @@ pub(crate) async fn dispatch_shared(
|
||||||
} => handle_remind(coord, agent, message, timing, file_path.as_deref()),
|
} => handle_remind(coord, agent, message, timing, file_path.as_deref()),
|
||||||
hive_core_agent_sock::Request::SetStatus { text } => handle_set_status(coord, text),
|
hive_core_agent_sock::Request::SetStatus { text } => handle_set_status(coord, text),
|
||||||
hive_core_agent_sock::Request::GetAgentMeta { name } => {
|
hive_core_agent_sock::Request::GetAgentMeta { name } => {
|
||||||
handle_get_agent_meta(coord, agent, name.as_deref()).await
|
handle_get_agent_meta(coord, agent, name.as_ref()).await
|
||||||
}
|
}
|
||||||
hive_core_agent_sock::Request::CancelLooseEnd { kind, id } => {
|
hive_core_agent_sock::Request::CancelLooseEnd { kind, id } => {
|
||||||
crate::questions::handle_cancel_loose_end(coord, agent, *kind, *id).map_or_else(
|
crate::questions::handle_cancel_loose_end(coord, agent, *kind, *id).map_or_else(
|
||||||
|
|
@ -321,7 +321,7 @@ async fn handle_recv(
|
||||||
messages: deliveries
|
messages: deliveries
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.map(|d| hive_sh4re::DeliveredMessage {
|
.map(|d| hive_sh4re::DeliveredMessage {
|
||||||
from: d.message.from,
|
from: d.message.from.to_string(),
|
||||||
body: d.message.body,
|
body: d.message.body,
|
||||||
id: d.id,
|
id: d.id,
|
||||||
redelivered: d.redelivered,
|
redelivered: d.redelivered,
|
||||||
|
|
@ -347,7 +347,7 @@ fn handle_wake(
|
||||||
body: &str,
|
body: &str,
|
||||||
) -> hive_core_agent_sock::Response {
|
) -> hive_core_agent_sock::Response {
|
||||||
match coord.broker.send(&Message {
|
match coord.broker.send(&Message {
|
||||||
from: from.to_owned(),
|
from: hive_sh4re::trusted_sender(from),
|
||||||
to: agent.to_owned(),
|
to: agent.to_owned(),
|
||||||
body: body.to_owned(),
|
body: body.to_owned(),
|
||||||
in_reply_to: None,
|
in_reply_to: None,
|
||||||
|
|
@ -418,42 +418,43 @@ async fn handle_create_repo(agent: &str, repo: &str) -> hive_core_agent_sock::Re
|
||||||
async fn handle_get_agent_meta(
|
async fn handle_get_agent_meta(
|
||||||
coord: &Arc<Coordinator>,
|
coord: &Arc<Coordinator>,
|
||||||
agent: &str,
|
agent: &str,
|
||||||
name: Option<&str>,
|
name: Option<&hive_types::Ident>,
|
||||||
) -> hive_core_agent_sock::Response {
|
) -> hive_core_agent_sock::Response {
|
||||||
let target = name.unwrap_or(agent);
|
// `name` arrives pre-validated by serde (wire field is `Ident`). The
|
||||||
// `name` is agent-supplied and flows into filesystem reads below
|
// `None` default (target == caller) still needs a parse since `agent`
|
||||||
// (`read_agent_status_live`, `read_agent_matrix_identities` →
|
// is a plain `&str` here — but it's the caller's own authenticated
|
||||||
// `agent_notes_dir(target)`), where a `../` component would traverse at
|
// name, already valid in practice.
|
||||||
// the OS level. Validate it before any path is built. The `None` default
|
let target_id = match name {
|
||||||
// (`target == agent`) is the caller's own authenticated name, already
|
Some(id) => id.clone(),
|
||||||
// valid — but validating unconditionally is simplest and harmless.
|
None => match hive_types::Ident::parse(agent) {
|
||||||
let target_id = match hive_types::Ident::parse(target) {
|
Ok(id) => id,
|
||||||
Ok(id) => id,
|
Err(reason) => {
|
||||||
Err(reason) => {
|
return hive_core_agent_sock::Response::Err {
|
||||||
return hive_core_agent_sock::Response::Err {
|
message: format!("get_agent_meta: invalid agent name {agent:?}: {reason}"),
|
||||||
message: format!("get_agent_meta: invalid agent name {target:?}: {reason}"),
|
};
|
||||||
};
|
}
|
||||||
}
|
},
|
||||||
};
|
};
|
||||||
let (status_text, status_set_at, running) =
|
let (status_text, status_set_at, running) =
|
||||||
crate::container_view::read_agent_status_live(&target_id).await;
|
crate::container_view::read_agent_status_live(&target_id).await;
|
||||||
let (hive_name, swarm_name) = crate::container_view::hive_swarm_names();
|
let (hive_name, swarm_name) = crate::container_view::hive_swarm_names();
|
||||||
|
// Matrix identities are public handles (`name` / `user_id`
|
||||||
|
// `@user:server` / `homeserver`) — the access token lives separately
|
||||||
|
// in the agent's `matrix-token` and is never part of this response.
|
||||||
|
// Peer visibility is intentional: it lets an agent verify/contact
|
||||||
|
// another on a public matrix instance. The `Ident::parse` gate
|
||||||
|
// above is what closes the real vector here (path traversal via `../`
|
||||||
|
// in an agent-supplied name).
|
||||||
|
let matrix_accounts = read_agent_matrix_identities(&target_id);
|
||||||
hive_core_agent_sock::Response::AgentMeta {
|
hive_core_agent_sock::Response::AgentMeta {
|
||||||
name: target.to_owned(),
|
name: target_id.into_string(),
|
||||||
running,
|
running,
|
||||||
hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake),
|
hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake),
|
||||||
status_text,
|
status_text,
|
||||||
status_set_at,
|
status_set_at,
|
||||||
hive_name,
|
hive_name,
|
||||||
swarm_name,
|
swarm_name,
|
||||||
// Matrix identities are public handles (`name` / `user_id`
|
matrix_accounts,
|
||||||
// `@user:server` / `homeserver`) — the access token lives separately
|
|
||||||
// in the agent's `matrix-token` and is never part of this response.
|
|
||||||
// Peer visibility is intentional: it lets an agent verify/contact
|
|
||||||
// another on a public matrix instance. The `Ident::parse` gate
|
|
||||||
// above is what closes the real vector here (path traversal via `../`
|
|
||||||
// in an agent-supplied name).
|
|
||||||
matrix_accounts: read_agent_matrix_identities(&target_id),
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -489,7 +490,7 @@ fn handle_operator_msg(
|
||||||
body: &str,
|
body: &str,
|
||||||
) -> hive_core_agent_sock::Response {
|
) -> hive_core_agent_sock::Response {
|
||||||
match coord.broker.send(&Message {
|
match coord.broker.send(&Message {
|
||||||
from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
|
from: hive_sh4re::trusted_sender(hive_sh4re::OPERATOR_RECIPIENT),
|
||||||
to: agent.to_owned(),
|
to: agent.to_owned(),
|
||||||
body: body.to_owned(),
|
body: body.to_owned(),
|
||||||
in_reply_to: None,
|
in_reply_to: None,
|
||||||
|
|
@ -954,7 +955,7 @@ pub(crate) fn fan_out_send(
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
if let Err(e) = coord.broker.send(&Message {
|
if let Err(e) = coord.broker.send(&Message {
|
||||||
from: from.to_owned(),
|
from: hive_sh4re::trusted_sender(from),
|
||||||
to: target.clone(),
|
to: target.clone(),
|
||||||
body: body.to_owned(),
|
body: body.to_owned(),
|
||||||
in_reply_to,
|
in_reply_to,
|
||||||
|
|
@ -1039,7 +1040,7 @@ pub(crate) fn handle_send(
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
match coord.broker.send(&Message {
|
match coord.broker.send(&Message {
|
||||||
from: agent.to_owned(),
|
from: hive_sh4re::trusted_sender(agent),
|
||||||
to: resolved,
|
to: resolved,
|
||||||
body: body.to_owned(),
|
body: body.to_owned(),
|
||||||
in_reply_to,
|
in_reply_to,
|
||||||
|
|
|
||||||
|
|
@ -307,7 +307,7 @@ impl Approvals {
|
||||||
/// across both callers (one suppressed the lint, the other aliased the
|
/// across both callers (one suppressed the lint, the other aliased the
|
||||||
/// tuple) — one named projection + mapper now backs both.
|
/// tuple) — one named projection + mapper now backs both.
|
||||||
struct ApprovalLookup {
|
struct ApprovalLookup {
|
||||||
agent: String,
|
agent: hive_types::Ident,
|
||||||
kind: String,
|
kind: String,
|
||||||
commit_ref: String,
|
commit_ref: String,
|
||||||
requested_at: i64,
|
requested_at: i64,
|
||||||
|
|
@ -323,8 +323,16 @@ impl ApprovalLookup {
|
||||||
description FROM approvals WHERE id = ?1";
|
description FROM approvals WHERE id = ?1";
|
||||||
|
|
||||||
fn from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<Self> {
|
fn from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<Self> {
|
||||||
|
let agent: String = row.get(0)?;
|
||||||
|
let agent = hive_types::Ident::parse(&agent).map_err(|e| {
|
||||||
|
rusqlite::Error::FromSqlConversionFailure(
|
||||||
|
0,
|
||||||
|
rusqlite::types::Type::Text,
|
||||||
|
format!("invalid approval agent {agent:?}: {e}").into(),
|
||||||
|
)
|
||||||
|
})?;
|
||||||
Ok(Self {
|
Ok(Self {
|
||||||
agent: row.get(0)?,
|
agent,
|
||||||
kind: row.get(1)?,
|
kind: row.get(1)?,
|
||||||
commit_ref: row.get(2)?,
|
commit_ref: row.get(2)?,
|
||||||
requested_at: row.get(3)?,
|
requested_at: row.get(3)?,
|
||||||
|
|
@ -399,9 +407,17 @@ fn row_to_approval(row: &rusqlite::Row<'_>) -> rusqlite::Result<Approval> {
|
||||||
));
|
));
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
let agent: String = row.get(1)?;
|
||||||
|
let agent = hive_types::Ident::parse(&agent).map_err(|e| {
|
||||||
|
rusqlite::Error::FromSqlConversionFailure(
|
||||||
|
1,
|
||||||
|
rusqlite::types::Type::Text,
|
||||||
|
format!("invalid approval agent {agent:?}: {e}").into(),
|
||||||
|
)
|
||||||
|
})?;
|
||||||
Ok(Approval {
|
Ok(Approval {
|
||||||
id: row.get(0)?,
|
id: row.get(0)?,
|
||||||
agent: row.get(1)?,
|
agent,
|
||||||
kind,
|
kind,
|
||||||
commit_ref: row.get(3)?,
|
commit_ref: row.get(3)?,
|
||||||
requested_at: hive_sh4re::wire_time::from_secs(row.get(4)?),
|
requested_at: hive_sh4re::wire_time::from_secs(row.get(4)?),
|
||||||
|
|
|
||||||
|
|
@ -277,12 +277,12 @@ impl Broker {
|
||||||
// Operator messages get elevated priority so they surface before
|
// Operator messages get elevated priority so they surface before
|
||||||
// queued wakes (bash completions, forge events, etc.) when the
|
// queued wakes (bash completions, forge events, etc.) when the
|
||||||
// harness pops the next turn driver. All other senders stay at 0.
|
// harness pops the next turn driver. All other senders stay at 0.
|
||||||
let priority: i64 = i64::from(message.from == "operator");
|
let priority: i64 = i64::from(message.from.as_str() == hive_sh4re::OPERATOR_RECIPIENT);
|
||||||
conn.execute(
|
conn.execute(
|
||||||
"INSERT INTO messages (sender, recipient, body, sent_at, in_reply_to, priority) \
|
"INSERT INTO messages (sender, recipient, body, sent_at, in_reply_to, priority) \
|
||||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
|
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
|
||||||
params![
|
params![
|
||||||
message.from,
|
message.from.as_str(),
|
||||||
message.to,
|
message.to,
|
||||||
message.body,
|
message.body,
|
||||||
now,
|
now,
|
||||||
|
|
@ -294,7 +294,7 @@ impl Broker {
|
||||||
drop(conn);
|
drop(conn);
|
||||||
let _ = self.events.send(MessageEvent::Sent {
|
let _ = self.events.send(MessageEvent::Sent {
|
||||||
id: row_id,
|
id: row_id,
|
||||||
from: message.from.clone(),
|
from: message.from.to_string(),
|
||||||
to: message.to.clone(),
|
to: message.to.clone(),
|
||||||
body: message.body.clone(),
|
body: message.body.clone(),
|
||||||
at: now,
|
at: now,
|
||||||
|
|
@ -664,7 +664,7 @@ impl Broker {
|
||||||
id,
|
id,
|
||||||
redelivered,
|
redelivered,
|
||||||
message: Message {
|
message: Message {
|
||||||
from,
|
from: hive_sh4re::trusted_sender(&from),
|
||||||
to,
|
to,
|
||||||
body,
|
body,
|
||||||
in_reply_to,
|
in_reply_to,
|
||||||
|
|
@ -678,7 +678,7 @@ impl Broker {
|
||||||
for d in &deliveries {
|
for d in &deliveries {
|
||||||
let _ = self.events.send(MessageEvent::Delivered {
|
let _ = self.events.send(MessageEvent::Delivered {
|
||||||
id: d.id,
|
id: d.id,
|
||||||
from: d.message.from.clone(),
|
from: d.message.from.to_string(),
|
||||||
to: d.message.to.clone(),
|
to: d.message.to.clone(),
|
||||||
body: d.message.body.clone(),
|
body: d.message.body.clone(),
|
||||||
at: now,
|
at: now,
|
||||||
|
|
@ -1187,7 +1187,7 @@ mod tests {
|
||||||
|
|
||||||
fn msg(from: &str, to: &str, body: &str) -> Message {
|
fn msg(from: &str, to: &str, body: &str) -> Message {
|
||||||
Message {
|
Message {
|
||||||
from: from.to_owned(),
|
from: hive_types::Ident::parse(from).expect("test sender must be a valid ident"),
|
||||||
to: to.to_owned(),
|
to: to.to_owned(),
|
||||||
body: body.to_owned(),
|
body: body.to_owned(),
|
||||||
in_reply_to: None,
|
in_reply_to: None,
|
||||||
|
|
@ -1475,7 +1475,7 @@ mod tests {
|
||||||
assert_eq!(batch.len(), 3);
|
assert_eq!(batch.len(), 3);
|
||||||
// Operator message surfaces first despite arriving last.
|
// Operator message surfaces first despite arriving last.
|
||||||
assert_eq!(batch[0].message.body, "stop what you're doing");
|
assert_eq!(batch[0].message.body, "stop what you're doing");
|
||||||
assert_eq!(batch[0].message.from, "operator");
|
assert_eq!(batch[0].message.from.as_str(), "operator");
|
||||||
// Remaining two in FIFO order.
|
// Remaining two in FIFO order.
|
||||||
assert_eq!(batch[1].message.body, "task done");
|
assert_eq!(batch[1].message.body, "task done");
|
||||||
assert_eq!(batch[2].message.body, "new pr");
|
assert_eq!(batch[2].message.body, "new pr");
|
||||||
|
|
|
||||||
|
|
@ -129,7 +129,7 @@ fn fire_schedule(coord: &Arc<Coordinator>, schedule: &Schedule, now: i64) {
|
||||||
Ok(false) => {}
|
Ok(false) => {}
|
||||||
}
|
}
|
||||||
let msg = Message {
|
let msg = Message {
|
||||||
from: "scheduled".to_owned(),
|
from: hive_sh4re::trusted_sender("scheduled"),
|
||||||
to: target.clone(),
|
to: target.clone(),
|
||||||
body: schedule.body.clone(),
|
body: schedule.body.clone(),
|
||||||
in_reply_to: None,
|
in_reply_to: None,
|
||||||
|
|
@ -227,7 +227,7 @@ fn notify_operator_missing_target(coord: &Coordinator, schedule: &Schedule, targ
|
||||||
body = schedule.body
|
body = schedule.body
|
||||||
);
|
);
|
||||||
let msg = Message {
|
let msg = Message {
|
||||||
from: "scheduled".to_owned(),
|
from: hive_sh4re::trusted_sender("scheduled"),
|
||||||
to: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
|
to: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
|
||||||
body,
|
body,
|
||||||
in_reply_to: None,
|
in_reply_to: None,
|
||||||
|
|
@ -327,7 +327,7 @@ pub async fn fire_now(
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
let msg = Message {
|
let msg = Message {
|
||||||
from: "scheduled".to_owned(),
|
from: hive_sh4re::trusted_sender("scheduled"),
|
||||||
to: target.clone(),
|
to: target.clone(),
|
||||||
body: schedule.body.clone(),
|
body: schedule.body.clone(),
|
||||||
in_reply_to: None,
|
in_reply_to: None,
|
||||||
|
|
|
||||||
|
|
@ -8,4 +8,5 @@ workspace = true
|
||||||
|
|
||||||
[dependencies]
|
[dependencies]
|
||||||
hive-sh4re.workspace = true
|
hive-sh4re.workspace = true
|
||||||
|
hive-types.workspace = true
|
||||||
serde.workspace = true
|
serde.workspace = true
|
||||||
|
|
|
||||||
|
|
@ -11,6 +11,7 @@ use hive_sh4re::{
|
||||||
CancelLooseEndKind, ContainerInfo, DeliveredMessage, InboxRow, JournalPriority, LooseEnd,
|
CancelLooseEndKind, ContainerInfo, DeliveredMessage, InboxRow, JournalPriority, LooseEnd,
|
||||||
MatrixIdentity, ReminderStats, ReminderTiming, SchedulePromptPayload, WireSchedule,
|
MatrixIdentity, ReminderStats, ReminderTiming, SchedulePromptPayload, WireSchedule,
|
||||||
};
|
};
|
||||||
|
use hive_types::Ident;
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
|
|
||||||
/// serde `default` helper for `Response::AgentMeta::running` (absent = true).
|
/// serde `default` helper for `Response::AgentMeta::running` (absent = true).
|
||||||
|
|
@ -73,7 +74,7 @@ pub enum Request {
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
ttl_seconds: Option<u64>,
|
ttl_seconds: Option<u64>,
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
to: Option<String>,
|
to: Option<Ident>,
|
||||||
},
|
},
|
||||||
/// Answer a question previously routed to this agent via
|
/// Answer a question previously routed to this agent via
|
||||||
/// `HelperEvent::QuestionAsked`. Authorised callers + threading
|
/// `HelperEvent::QuestionAsked`. Authorised callers + threading
|
||||||
|
|
@ -139,7 +140,7 @@ pub enum Request {
|
||||||
/// `docs/conventions.md::Agent metadata`.
|
/// `docs/conventions.md::Agent metadata`.
|
||||||
GetAgentMeta {
|
GetAgentMeta {
|
||||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||||
name: Option<String>,
|
name: Option<Ident>,
|
||||||
},
|
},
|
||||||
/// Cancel an open thread the agent owns. Authorisation +
|
/// Cancel an open thread the agent owns. Authorisation +
|
||||||
/// per-kind semantics in
|
/// per-kind semantics in
|
||||||
|
|
|
||||||
|
|
@ -8,6 +8,7 @@ workspace = true
|
||||||
|
|
||||||
[dependencies]
|
[dependencies]
|
||||||
chrono.workspace = true
|
chrono.workspace = true
|
||||||
|
hive-types.workspace = true
|
||||||
schemars.workspace = true
|
schemars.workspace = true
|
||||||
serde.workspace = true
|
serde.workspace = true
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,6 +1,7 @@
|
||||||
//! Wire types shared between `hive-c0re` and the in-container harness.
|
//! Wire types shared between `hive-c0re` and the in-container harness.
|
||||||
|
|
||||||
use chrono::{DateTime, Utc};
|
use chrono::{DateTime, Utc};
|
||||||
|
use hive_types::Ident;
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
|
|
||||||
pub mod assets;
|
pub mod assets;
|
||||||
|
|
@ -52,7 +53,7 @@ pub fn pending_hint(remaining: u64) -> String {
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||||
pub struct Approval {
|
pub struct Approval {
|
||||||
pub id: i64,
|
pub id: i64,
|
||||||
pub agent: String,
|
pub agent: Ident,
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
pub kind: ApprovalKind,
|
pub kind: ApprovalKind,
|
||||||
/// Kind-specific payload (git sha / inputs array / schedule
|
/// Kind-specific payload (git sha / inputs array / schedule
|
||||||
|
|
@ -154,7 +155,7 @@ pub struct ReminderStats {
|
||||||
/// A logical message between agents.
|
/// A logical message between agents.
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||||
pub struct Message {
|
pub struct Message {
|
||||||
pub from: String,
|
pub from: Ident,
|
||||||
pub to: String,
|
pub to: String,
|
||||||
pub body: String,
|
pub body: String,
|
||||||
/// Optional broker row-id of the message this is a reply to.
|
/// Optional broker row-id of the message this is a reply to.
|
||||||
|
|
@ -441,6 +442,25 @@ pub const CHILDREN_RECIPIENT: &str = "<children>";
|
||||||
/// Manager harness recognises this and parses the body as a `HelperEvent`.
|
/// Manager harness recognises this and parses the body as a `HelperEvent`.
|
||||||
pub const SYSTEM_SENDER: &str = "system";
|
pub const SYSTEM_SENDER: &str = "system";
|
||||||
|
|
||||||
|
/// Parse `s` as a [`Ident`] for use as `Message.from`, falling back to
|
||||||
|
/// [`SYSTEM_SENDER`] on the (should-be-unreachable) case that `s` isn't
|
||||||
|
/// ident-shaped. `Message.from` is always either a fixed sentinel literal
|
||||||
|
/// (`SYSTEM_SENDER`, `OPERATOR_RECIPIENT`, `"scheduled"`, …) or an
|
||||||
|
/// already-registered agent's own name reaching this point through
|
||||||
|
/// hive-c0re's internal dispatch — never arbitrary external input — so
|
||||||
|
/// this is a defensive fallback for a programming-bug case, not a
|
||||||
|
/// validation gate.
|
||||||
|
///
|
||||||
|
/// # Panics
|
||||||
|
///
|
||||||
|
/// Never, unless [`SYSTEM_SENDER`] itself stops being ident-shaped (which
|
||||||
|
/// would also be a programming bug, caught by `hive-types`' own tests).
|
||||||
|
#[must_use]
|
||||||
|
pub fn trusted_sender(s: &str) -> Ident {
|
||||||
|
Ident::parse(s)
|
||||||
|
.unwrap_or_else(|_| Ident::parse(SYSTEM_SENDER).expect("SYSTEM_SENDER is a valid Ident"))
|
||||||
|
}
|
||||||
|
|
||||||
/// Out-of-band events the host-side daemon pushes to the manager's inbox.
|
/// Out-of-band events the host-side daemon pushes to the manager's inbox.
|
||||||
/// Serialised as JSON in `Message::body` (sender = `SYSTEM_SENDER`).
|
/// Serialised as JSON in `Message::body` (sender = `SYSTEM_SENDER`).
|
||||||
/// Per-variant triggers + the optional `sha`/`tag` semantics live in
|
/// Per-variant triggers + the optional `sha`/`tag` semantics live in
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue