vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2

fix(#1953): route approval helper-events to the submitter, not the root agent

Browse source
This commit is contained in:
damocles 2026-06-23 20:09:40 +02:00 • committed by mara
commit 3618399d94
7 changed files with 187 additions and 73 deletions
Download patch file Download diff file Expand all files Collapse all files

92
hive-c0re/src/actions.rs
View file

@ -546,15 +546,18 @@ fn finish_approval(
(ApprovalStatus::Failed, Some(note), false) (ApprovalStatus::Failed, Some(note), false)
} }
}; };
coord.notify_manager(&HelperEvent::ApprovalResolved { coord.notify_submitter(
id: approval.id, approval.id,
agent: approval.agent.clone(), &HelperEvent::ApprovalResolved {
commit_ref: approval.commit_ref.clone(), id: approval.id,
status, agent: approval.agent.clone(),
note: note.clone(), commit_ref: approval.commit_ref.clone(),
sha: approval.fetched_sha.clone(), status,
tag: terminal_tag.clone(), note: note.clone(),
}); sha: approval.fetched_sha.clone(),
tag: terminal_tag.clone(),
},
);
// Phase 5b: also fire on the dashboard event channel so the // Phase 5b: also fire on the dashboard event channel so the
// browser moves the row out of pending into history without a // browser moves the row out of pending into history without a
// snapshot refetch. `approved` rows that succeed get the // snapshot refetch. `approved` rows that succeed get the
@ -590,37 +593,49 @@ fn finish_approval(
match approval.kind { match approval.kind {
ApprovalKind::InitConfig => { ApprovalKind::InitConfig => {
if ok { if ok {
coord.notify_manager(&HelperEvent::ConfigReady { coord.notify_submitter(
agent: approval.agent.clone(), approval.id,
}); &HelperEvent::ConfigReady {
agent: approval.agent.clone(),
},
);
} }
} }
ApprovalKind::Spawn => coord.notify_manager(&HelperEvent::Spawned { ApprovalKind::Spawn => coord.notify_submitter(
agent: approval.agent.clone(), approval.id,
ok, &HelperEvent::Spawned {
note,
sha: approval.fetched_sha.clone(),
}),
ApprovalKind::ApplyCommit if is_first_spawn => {
coord.notify_manager(&HelperEvent::Spawned {
agent: approval.agent.clone(), agent: approval.agent.clone(),
ok, ok,
note, note,
sha: approval.fetched_sha.clone(), sha: approval.fetched_sha.clone(),
}); },
),
ApprovalKind::ApplyCommit if is_first_spawn => {
coord.notify_submitter(
approval.id,
&HelperEvent::Spawned {
agent: approval.agent.clone(),
ok,
note,
sha: approval.fetched_sha.clone(),
},
);
} }
// MergeConfigPr ends in a container rebuild just like a // MergeConfigPr ends in a container rebuild just like a
// non-first-spawn ApplyCommit, so both surface the same Rebuilt // non-first-spawn ApplyCommit, so both surface the same Rebuilt
// lifecycle event. (MergeConfigPr is never a first spawn — the // lifecycle event. (MergeConfigPr is never a first spawn — the
// agent already exists — so it never hits the Spawned arm above.) // agent already exists — so it never hits the Spawned arm above.)
ApprovalKind::ApplyCommit | ApprovalKind::MergeConfigPr => { ApprovalKind::ApplyCommit | ApprovalKind::MergeConfigPr => {
coord.notify_manager(&HelperEvent::Rebuilt { coord.notify_submitter(
agent: approval.agent.clone(), approval.id,
ok, &HelperEvent::Rebuilt {
note, agent: approval.agent.clone(),
sha: approval.fetched_sha.clone(), ok,
tag: terminal_tag, note,
}); sha: approval.fetched_sha.clone(),
tag: terminal_tag,
},
);
} }
// UpdateMetaInputs / SchedulePrompt: ApprovalResolved already // UpdateMetaInputs / SchedulePrompt: ApprovalResolved already
// carries the result. No separate lifecycle event needed. // carries the result. No separate lifecycle event needed.
@ -1037,15 +1052,18 @@ pub async fn deny(coord: &Coordinator, id: i64, note: Option<&str>) -> Result<()
let sha_short = sha.as_deref().map(|s| s[..s.len().min(12)].to_owned()); let sha_short = sha.as_deref().map(|s| s[..s.len().min(12)].to_owned());
let description = a.description.clone(); let description = a.description.clone();
let agent_owned = a.agent.clone(); let agent_owned = a.agent.clone();
coord.notify_manager(&HelperEvent::ApprovalResolved { coord.notify_submitter(
id: a.id, a.id,
agent: a.agent, &HelperEvent::ApprovalResolved {
commit_ref: a.commit_ref, id: a.id,
status: ApprovalStatus::Denied, agent: a.agent,
note: note.map(String::from), commit_ref: a.commit_ref,
sha, status: ApprovalStatus::Denied,
tag, note: note.map(String::from),
}); sha,
tag,
},
);
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved { coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
id, id,
agent: &agent_owned, agent: &agent_owned,

83
hive-c0re/src/approvals.rs
View file

@ -70,6 +70,21 @@ fn ensure_fetched_sha_column(conn: &Connection) -> Result<()> {
Ok(()) Ok(())
} }
/// Same shape as `ensure_fetched_sha_column` but for `submitter` — the
/// agent that submitted the approval (the authenticated socket caller).
/// Approval-scoped helper events route to this agent. Legacy rows have
/// NULL; callers fall back to the root agent for those.
fn ensure_submitter_column(conn: &Connection) -> Result<()> {
let has: bool = conn
.prepare("SELECT 1 FROM pragma_table_info('approvals') WHERE name = 'submitter'")?
.exists([])?;
if !has {
conn.execute_batch("ALTER TABLE approvals ADD COLUMN submitter TEXT;")
.context("add approvals.submitter column")?;
}
Ok(())
}
pub struct Approvals { pub struct Approvals {
conn: Mutex<Connection>, conn: Mutex<Connection>,
} }
@ -87,6 +102,7 @@ impl Approvals {
ensure_kind_column(&conn).context("migrate approvals.kind")?; ensure_kind_column(&conn).context("migrate approvals.kind")?;
ensure_fetched_sha_column(&conn).context("migrate approvals.fetched_sha")?; ensure_fetched_sha_column(&conn).context("migrate approvals.fetched_sha")?;
ensure_description_column(&conn).context("migrate approvals.description")?; ensure_description_column(&conn).context("migrate approvals.description")?;
ensure_submitter_column(&conn).context("migrate approvals.submitter")?;
Ok(Self { Ok(Self {
conn: Mutex::new(conn), conn: Mutex::new(conn),
}) })
@ -98,22 +114,41 @@ impl Approvals {
kind: ApprovalKind, kind: ApprovalKind,
commit_ref: &str, commit_ref: &str,
description: Option<&str>, description: Option<&str>,
submitter: &str,
) -> Result<i64> { ) -> Result<i64> {
let conn = self.conn.lock().unwrap(); let conn = self.conn.lock().unwrap();
conn.execute( conn.execute(
"INSERT INTO approvals (agent, kind, commit_ref, requested_at, status, description) "INSERT INTO approvals
VALUES (?1, ?2, ?3, ?4, 'pending', ?5)", (agent, kind, commit_ref, requested_at, status, description, submitter)
VALUES (?1, ?2, ?3, ?4, 'pending', ?5, ?6)",
params![ params![
agent, agent,
kind_to_str(kind), kind_to_str(kind),
commit_ref, commit_ref,
now_unix(), now_unix(),
description description,
submitter
], ],
)?; )?;
Ok(conn.last_insert_rowid()) Ok(conn.last_insert_rowid())
} }
/// The agent that submitted approval `id` (the authenticated socket
/// caller at submit time). `None` for legacy rows predating the
/// `submitter` column — callers route those to the root agent.
pub fn submitter_of(&self, id: i64) -> Result<Option<String>> {
let conn = self.conn.lock().unwrap();
let submitter: Option<String> = conn
.query_row(
"SELECT submitter FROM approvals WHERE id = ?1",
params![id],
|row| row.get(0),
)
.optional()?
.flatten();
Ok(submitter)
}
/// Record the canonical sha hive-c0re fetched from the proposed repo /// Record the canonical sha hive-c0re fetched from the proposed repo
/// into applied at submission time. Idempotent on identical values. /// into applied at submission time. Idempotent on identical values.
pub fn set_fetched_sha(&self, id: i64, sha: &str) -> Result<()> { pub fn set_fetched_sha(&self, id: i64, sha: &str) -> Result<()> {
@ -428,7 +463,13 @@ mod tests {
// approval then vanished from the dashboard. // approval then vanished from the dashboard.
let (_dir, _path, db) = open_temp(); let (_dir, _path, db) = open_temp();
let id = db let id = db
.submit_kind("bitburner", ApprovalKind::InitConfig, "", Some("scaffold")) .submit_kind(
"bitburner",
ApprovalKind::InitConfig,
"",
Some("scaffold"),
"bitburner",
)
.expect("submit init_config"); .expect("submit init_config");
let pending = db let pending = db
.pending() .pending()
@ -441,10 +482,11 @@ mod tests {
#[test] #[test]
fn mixed_kinds_all_listed() { fn mixed_kinds_all_listed() {
let (_dir, _path, db) = open_temp(); let (_dir, _path, db) = open_temp();
db.submit_kind("a", ApprovalKind::ApplyCommit, "deadbeef", None) db.submit_kind("a", ApprovalKind::ApplyCommit, "deadbeef", None, "a")
.unwrap(); .unwrap();
db.submit_kind("b", ApprovalKind::Spawn, "", None).unwrap(); db.submit_kind("b", ApprovalKind::Spawn, "", None, "b")
db.submit_kind("c", ApprovalKind::InitConfig, "", None) .unwrap();
db.submit_kind("c", ApprovalKind::InitConfig, "", None, "c")
.unwrap(); .unwrap();
let pending = db.pending().expect("pending"); let pending = db.pending().expect("pending");
assert_eq!(pending.len(), 3, "all three kinds must be visible"); assert_eq!(pending.len(), 3, "all three kinds must be visible");
@ -462,6 +504,7 @@ mod tests {
ApprovalKind::ApplyCommit, ApprovalKind::ApplyCommit,
"cafef00d", "cafef00d",
Some("test"), Some("test"),
"bitburner",
) )
.unwrap(); .unwrap();
let row = db.mark_cancelled(id, "manager").expect("cancel"); let row = db.mark_cancelled(id, "manager").expect("cancel");
@ -481,7 +524,7 @@ mod tests {
// final — re-cancelling errors instead of silently overwriting. // final — re-cancelling errors instead of silently overwriting.
let (_dir, _path, db) = open_temp(); let (_dir, _path, db) = open_temp();
let id = db let id = db
.submit_kind("a", ApprovalKind::Spawn, "deadbeef", None) .submit_kind("a", ApprovalKind::Spawn, "deadbeef", None, "a")
.unwrap(); .unwrap();
db.mark_cancelled(id, "manager").expect("first cancel"); db.mark_cancelled(id, "manager").expect("first cancel");
let err = db let err = db
@ -496,7 +539,7 @@ mod tests {
// whole list — collect_lenient skips it instead of failing. // whole list — collect_lenient skips it instead of failing.
let (_dir, path, db) = open_temp(); let (_dir, path, db) = open_temp();
let good = db let good = db
.submit_kind("good", ApprovalKind::ApplyCommit, "cafe", None) .submit_kind("good", ApprovalKind::ApplyCommit, "cafe", None, "good")
.unwrap(); .unwrap();
let raw = Connection::open(&path).unwrap(); let raw = Connection::open(&path).unwrap();
raw.execute( raw.execute(
@ -511,4 +554,26 @@ mod tests {
assert_eq!(pending.len(), 1); assert_eq!(pending.len(), 1);
assert_eq!(pending[0].id, good); assert_eq!(pending[0].id, good);
} }
#[test]
fn submitter_round_trips_and_legacy_reads_none() {
// A submitted approval records its submitter; a legacy row
// (inserted without the column) reads back as None so callers
// fall back to the root agent.
let (_dir, path, db) = open_temp();
let id = db
.submit_kind("child", ApprovalKind::ApplyCommit, "cafe", None, "parent")
.unwrap();
assert_eq!(db.submitter_of(id).unwrap().as_deref(), Some("parent"));
let raw = Connection::open(&path).unwrap();
raw.execute(
"INSERT INTO approvals (agent, kind, commit_ref, requested_at, status)
VALUES ('old', 'apply_commit', '', 0, 'pending')",
[],
)
.unwrap();
let legacy_id = raw.last_insert_rowid();
assert_eq!(db.submitter_of(legacy_id).unwrap(), None);
}
} }

14
hive-c0re/src/coordinator.rs
View file

@ -1295,6 +1295,20 @@ impl Coordinator {
self.notify_agent(hive_sh4re::MANAGER_AGENT, event); self.notify_agent(hive_sh4re::MANAGER_AGENT, event);
} }
/// Route an approval-scoped helper event to the agent that submitted
/// approval `approval_id` (the authenticated socket caller at submit
/// time). Legacy rows with no recorded submitter — and any lookup
/// failure — fall back to the root agent, preserving the prior
/// always-root behaviour.
pub fn notify_submitter(&self, approval_id: i64, event: &hive_sh4re::HelperEvent) {
let target = self
.approvals
.submitter_of(approval_id)
.unwrap_or_default()
.unwrap_or_else(|| hive_sh4re::MANAGER_AGENT.to_owned());
self.notify_agent(&target, event);
}
/// Push a `HelperEvent` into an arbitrary agent's inbox. Encoded /// Push a `HelperEvent` into an arbitrary agent's inbox. Encoded
/// the same way as `notify_manager` (sender = `SYSTEM_SENDER`, /// the same way as `notify_manager` (sender = `SYSTEM_SENDER`,
/// body = JSON-encoded event). Used to route `QuestionAnswered` /// body = JSON-encoded event). Used to route `QuestionAnswered`

12
hive-c0re/src/dashboard.rs
View file

@ -1650,11 +1650,13 @@ async fn post_request_spawn(
if name.is_empty() { if name.is_empty() {
return error_response("spawn: `name` required"); return error_response("spawn: `name` required");
} }
match state match state.coord.approvals.submit_kind(
.coord &name,
.approvals hive_sh4re::ApprovalKind::Spawn,
.submit_kind(&name, hive_sh4re::ApprovalKind::Spawn, "", None) "",
{ None,
hive_sh4re::MANAGER_AGENT,
) {
Ok(id) => { Ok(id) => {
tracing::info!(%id, %name, "operator: spawn approval queued via dashboard"); tracing::info!(%id, %name, "operator: spawn approval queued via dashboard");
// Phase 5b: notify the dashboard event channel so live // Phase 5b: notify the dashboard event channel so live

38
hive-c0re/src/loose_ends.rs
View file

@ -21,10 +21,10 @@ use hive_sh4re::{LooseEnd, MANAGER_AGENT};
use crate::coordinator::Coordinator; use crate::coordinator::Coordinator;
/// Open threads pending against `agent`: /// Open threads pending against `agent`:
/// - pending approvals where this agent is the submitter (only ever /// - pending approvals where this agent is the submitter (a parent
/// true for the manager — sub-agents don't submit approvals — but /// agent with the `approvals` group submits for its children; the
/// we keep the rule per-agent so the manager's MCP surface gets /// root submits for top-level agents). Legacy rows with no recorded
/// the same shape via a different code path); /// submitter count as the root's;
/// - unanswered questions where `agent` is the asker (waiting on /// - unanswered questions where `agent` is the asker (waiting on
/// someone) OR the target (owes a reply); /// someone) OR the target (owes a reply);
/// - pending reminders this agent scheduled (`owner == self`). /// - pending reminders this agent scheduled (`owner == self`).
@ -35,20 +35,24 @@ use crate::coordinator::Coordinator;
pub fn for_agent(coord: &Coordinator, agent: &str) -> Result<Vec<LooseEnd>> { pub fn for_agent(coord: &Coordinator, agent: &str) -> Result<Vec<LooseEnd>> {
let now = now_unix(); let now = now_unix();
let mut out = Vec::new(); let mut out = Vec::new();
// Approvals are only submitted by the manager today. When that // Show each pending approval to the agent that submitted it. The
// expands (e.g. sub-agents propose changes to their own configs), // submitter column is NULL for rows predating it; those count as
// teach the approvals table to track the submitter and filter // the root agent's.
// here on that column — for now MANAGER_AGENT == sole submitter. for a in coord.approvals.pending()? {
if agent == MANAGER_AGENT { let submitter = coord
for a in coord.approvals.pending()? { .approvals
out.push(LooseEnd::Approval { .submitter_of(a.id)?
id: a.id, .unwrap_or_else(|| MANAGER_AGENT.to_owned());
agent: a.agent, if submitter != agent {
commit_ref: a.commit_ref, continue;
description: a.description,
age_seconds: saturating_age(now, a.requested_at),
});
} }
out.push(LooseEnd::Approval {
id: a.id,
agent: a.agent,
commit_ref: a.commit_ref,
description: a.description,
age_seconds: saturating_age(now, a.requested_at),
});
} }
for q in coord.questions.pending_all()? { for q in coord.questions.pending_all()? {
let role_match = q.asker == agent || q.target.as_deref() == Some(agent); let role_match = q.asker == agent || q.target.as_deref() == Some(agent);

11
hive-c0re/src/server.rs
View file

@ -80,10 +80,13 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
HostRequest::Spawn { name } => handle_spawn(&coord, name).await?, HostRequest::Spawn { name } => handle_spawn(&coord, name).await?,
HostRequest::RequestSpawn { name } => { HostRequest::RequestSpawn { name } => {
tracing::info!(%name, "request_spawn"); tracing::info!(%name, "request_spawn");
let id = let id = coord.approvals.submit_kind(
coord name,
.approvals hive_sh4re::ApprovalKind::Spawn,
.submit_kind(name, hive_sh4re::ApprovalKind::Spawn, "", None)?; "",
None,
hive_sh4re::MANAGER_AGENT,
)?;
tracing::info!(%id, %name, "spawn approval queued"); tracing::info!(%id, %name, "spawn approval queued");
HostResponse::success() HostResponse::success()
} }

10
hive-c0re/src/socket_server.rs
View file

@ -989,7 +989,7 @@ async fn handle_request_apply_commit(
return err; return err;
} }
tracing::info!(%agent, %target_agent, %commit_ref, "request_apply_commit"); tracing::info!(%agent, %target_agent, %commit_ref, "request_apply_commit");
match submit_apply_commit(coord, target_agent, commit_ref, description).await { match submit_apply_commit(coord, target_agent, commit_ref, description, agent).await {
Ok((id, sha)) => { Ok((id, sha)) => {
tracing::info!(%id, %target_agent, %sha, "apply_commit approval queued"); tracing::info!(%id, %target_agent, %sha, "apply_commit approval queued");
AgentResponse::Ok AgentResponse::Ok
@ -1427,6 +1427,7 @@ fn handle_request_update_meta_inputs(
hive_sh4re::ApprovalKind::UpdateMetaInputs, hive_sh4re::ApprovalKind::UpdateMetaInputs,
&commit_ref, &commit_ref,
description, description,
requester,
) )
.map_err(|e| anyhow::anyhow!("{e:#}")) .map_err(|e| anyhow::anyhow!("{e:#}"))
{ {
@ -1527,6 +1528,7 @@ fn handle_request_schedule_prompt(
hive_sh4re::ApprovalKind::SchedulePrompt, hive_sh4re::ApprovalKind::SchedulePrompt,
&commit_ref, &commit_ref,
payload.description.as_deref(), payload.description.as_deref(),
requester,
) { ) {
Ok(id) => id, Ok(id) => id,
Err(e) => { Err(e) => {
@ -1801,6 +1803,10 @@ pub(crate) fn submit_init_config(
hive_sh4re::ApprovalKind::InitConfig, hive_sh4re::ApprovalKind::InitConfig,
parent.unwrap_or(""), parent.unwrap_or(""),
description.as_deref(), description.as_deref(),
// `parent` is the requesting agent (becomes the new child's
// parent); it's also the submitter the approval events route
// back to. No declared parent = operator/root path.
parent.unwrap_or(hive_sh4re::MANAGER_AGENT),
) )
.map_err(|e| anyhow::anyhow!("queue approval row: {e:#}"))?; .map_err(|e| anyhow::anyhow!("queue approval row: {e:#}"))?;
tracing::info!(%id, %name, "init_config approval queued"); tracing::info!(%id, %name, "init_config approval queued");
@ -1832,6 +1838,7 @@ pub(crate) async fn submit_apply_commit(
agent: &str, agent: &str,
commit_ref: &str, commit_ref: &str,
description: Option<&str>, description: Option<&str>,
submitter: &str,
) -> anyhow::Result<(i64, String)> { ) -> anyhow::Result<(i64, String)> {
validate_commit_ref(commit_ref)?; validate_commit_ref(commit_ref)?;
let proposed_dir = crate::coordinator::Coordinator::agent_proposed_dir(agent); let proposed_dir = crate::coordinator::Coordinator::agent_proposed_dir(agent);
@ -1859,6 +1866,7 @@ pub(crate) async fn submit_apply_commit(
hive_sh4re::ApprovalKind::ApplyCommit, hive_sh4re::ApprovalKind::ApplyCommit,
commit_ref, commit_ref,
description, description,
submitter,
) )
.map_err(|e| anyhow::anyhow!("queue approval row: {e:#}"))?; .map_err(|e| anyhow::anyhow!("queue approval row: {e:#}"))?;
let tag = format!("proposal/{id}"); let tag = format!("proposal/{id}");