Watch
0
0
Fork
You've already forked hyperhive
0
hyperhive/hive-c0re/src/stores/approvals.rs
atlas efbfec6d01 config PRs: remove the hive's config-PR webhook, poll and core merge
An operator's merge on the forge deploys a config PR through
swarm-controller's DeployRequest{rev}. The hive-side path that queued a
MergeConfigPr approval and merged the PR as `core` goes:

- the `/webhook/config-pr` receiver, its HMAC secret, the WebhookRegister
  boot node and the org-hook registration; the hive vhost's `/webhook/`
  location
- the 5-minute config-PR poll
- ApprovalKind::MergeConfigPr, its dashboard card, and the deploy DAG it
  drove (DeployWindow, MergeVerify, DeployApply, FinalizeDeploy,
  DeployTail), with verify_commit, the two-phase meta deploy, rollback
  refs, the PR-failure comment and forge/pr_merge.rs
- `fetched_sha`, `sha_short`/`pr_number` on approval events, and
  `sha`/`tag` on HelperEvent::ApprovalResolved: only the merge path set
  them

`config_repo`, `merged_pr_for_commit` and `post_pr_comment` move to
forge/pr_comment.rs for the merged-rev deploy's refusal comment.
Approvals v5 drops stored `merge_config_pr` rows; a test reopens a v4
database holding them.

Closes #4850
2026-10-02 23:13:03 +02:00

549 lines
20 KiB
Rust

//! Approval queue. Requests are submitted by an agent
//! (`RequestSchedulePrompt`); the user approves/denies via the host admin
//! CLI; on approval the host runs the corresponding action.
//!
//! `UpdateMetaInputs` rows are legacy: the MCP tool that queued them was
//! removed and nothing produces the kind any more. The variant and
//! its (de)serialization stay so pre-existing rows still read back.
use std::path::Path;
use std::sync::Mutex;
use anyhow::{Context, Result, bail};
use chrono::Utc;
use hive_sh4re::approvals::{Approval, ApprovalKind, ApprovalStatus};
use rusqlite::{Connection, OptionalExtension, params};
use crate::db::Migration;
const SCHEMA: &str = r"
CREATE TABLE IF NOT EXISTS approvals (
id INTEGER PRIMARY KEY AUTOINCREMENT,
agent TEXT NOT NULL,
commit_ref TEXT NOT NULL,
requested_at INTEGER NOT NULL,
status TEXT NOT NULL,
resolved_at INTEGER,
note TEXT
);
CREATE INDEX IF NOT EXISTS idx_approvals_pending
ON approvals (id) WHERE status = 'pending';
";
/// Ordered schema migrations tracked in `schema_versions` (key `"approvals"`).
/// Each migration declares the column it adds, so a legacy DB (fully or
/// partially migrated before versioning) converges by skipping the migrations
/// whose column already exists. New columns go here as v5, v6, …
const MIGRATIONS: &[Migration] = &[
// v1: `kind` (pre-Phase-8 dbs): legacy rows default to `apply_commit`.
Migration {
sql: "ALTER TABLE approvals ADD COLUMN \
kind TEXT NOT NULL DEFAULT 'apply_commit'",
adds_column: Some(("approvals", "kind")),
},
// v2: `description`: manager-supplied note on the dashboard card.
Migration {
sql: "ALTER TABLE approvals ADD COLUMN description TEXT",
adds_column: Some(("approvals", "description")),
},
// v3: `fetched_sha`: canonical sha hive-c0re resolved at submit time.
Migration {
sql: "ALTER TABLE approvals ADD COLUMN fetched_sha TEXT",
adds_column: Some(("approvals", "fetched_sha")),
},
// v4: `submitter`: authenticated agent that submitted the approval.
// Legacy rows are NULL → callers fall back to the root agent.
Migration {
sql: "ALTER TABLE approvals ADD COLUMN submitter TEXT",
adds_column: Some(("approvals", "submitter")),
},
// v5: drop `merge_config_pr` rows. Config PRs merge on the forge, so no
// approval of that kind can be acted on, and `row_to_approval` rejects
// the kind.
Migration {
sql: "DELETE FROM approvals WHERE kind = 'merge_config_pr'",
adds_column: None,
},
];
pub struct Approvals {
conn: Mutex<Connection>,
}
impl Approvals {
pub fn open(path: &Path) -> Result<Self> {
let conn = crate::db::open(path, "approvals")?;
conn.execute_batch(SCHEMA)
.context("apply approvals schema")?;
crate::db::apply_versioned_migrations(&conn, "approvals", MIGRATIONS)?;
Ok(Self {
conn: Mutex::new(conn),
})
}
/// Insert a new pending approval row.
pub fn submit_kind(
&self,
agent: &str,
kind: ApprovalKind,
commit_ref: &str,
description: Option<&str>,
submitter: &str,
) -> Result<i64> {
let conn = self.conn.lock().unwrap();
conn.execute(
"INSERT INTO approvals
(agent, kind, commit_ref, requested_at, status, description, submitter)
VALUES (?1, ?2, ?3, ?4, 'pending', ?5, ?6)",
params![
agent,
<&str>::from(kind),
commit_ref,
Utc::now().timestamp(),
description,
submitter,
],
)?;
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.
///
/// # Errors
///
/// Returns an error if the sqlite prepare/query fails. A missing row
/// or a `NULL` submitter is not an error — both yield `Ok(None)`.
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)
}
/// Last `limit` resolved approvals (approved / denied / failed),
/// newest-first. Drives the history tab on the dashboard.
pub fn recent_resolved(&self, limit: u64) -> Result<Vec<Approval>> {
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT id, agent, kind, commit_ref, requested_at, status, resolved_at, note, description
FROM approvals
WHERE status IN ('approved', 'denied', 'failed', 'cancelled')
ORDER BY resolved_at DESC, id DESC
LIMIT ?1",
)?;
let rows = stmt.query_map([limit], row_to_approval)?;
Ok(collect_lenient(rows))
}
pub fn pending(&self) -> Result<Vec<Approval>> {
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT id, agent, kind, commit_ref, requested_at, status, resolved_at, note, description
FROM approvals
WHERE status = 'pending'
ORDER BY id ASC",
)?;
let rows = stmt.query_map([], row_to_approval)?;
Ok(collect_lenient(rows))
}
pub fn get(&self, id: i64) -> Result<Option<Approval>> {
let conn = self.conn.lock().unwrap();
conn.query_row(
"SELECT id, agent, kind, commit_ref, requested_at, status, resolved_at, note, description
FROM approvals WHERE id = ?1",
params![id],
row_to_approval,
)
.optional()
.map_err(Into::into)
}
/// Mark pending -> approved (or fail if not pending). Returns the (now-updated)
/// approval so the caller can run the action and pass the agent name.
pub fn mark_approved(&self, id: i64) -> Result<Approval> {
let conn = self.conn.lock().unwrap();
let row: Option<ApprovalLookup> = conn
.query_row(
ApprovalLookup::SELECT,
params![id],
ApprovalLookup::from_row,
)
.optional()?;
let Some(row) = row else {
bail!("approval {id} not found");
};
if row.status != "pending" {
bail!("approval {id} is {}, not pending", row.status);
}
let resolved_at = Utc::now().timestamp();
conn.execute(
"UPDATE approvals SET status = 'approved', resolved_at = ?1 WHERE id = ?2",
params![resolved_at, id],
)?;
Ok(Approval {
id,
agent: row.agent,
kind: kind_from_str(&row.kind)?,
commit_ref: row.commit_ref,
requested_at: hive_sh4re::wire_time::from_secs(row.requested_at),
status: ApprovalStatus::Approved,
resolved_at: Some(hive_sh4re::wire_time::from_secs(resolved_at)),
note: None,
description: row.description,
})
}
pub fn mark_denied(&self, id: i64, note: Option<&str>) -> Result<()> {
let conn = self.conn.lock().unwrap();
let affected = conn.execute(
"UPDATE approvals SET status = 'denied', resolved_at = ?1, note = ?2
WHERE id = ?3 AND status = 'pending'",
params![Utc::now().timestamp(), note, id],
)?;
if affected == 0 {
bail!("approval {id} not pending");
}
Ok(())
}
pub fn mark_failed(&self, id: i64, note: &str) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute(
"UPDATE approvals SET status = 'failed', resolved_at = ?1, note = ?2 WHERE id = ?3",
params![Utc::now().timestamp(), note, id],
)?;
Ok(())
}
/// Withdraw a pending approval. Returns the now-updated
/// row so the caller can emit `ApprovalResolved` with the right
/// kind / agent. Errors if the approval isn't pending — once
/// it's approved/denied/failed/cancelled, the resolution is final.
pub fn mark_cancelled(&self, id: i64, canceller: &str) -> Result<Approval> {
let mut conn = self.conn.lock().unwrap();
let tx = conn.transaction()?;
let row: Option<ApprovalLookup> = tx
.query_row(
ApprovalLookup::SELECT,
params![id],
ApprovalLookup::from_row,
)
.optional()?;
let Some(row) = row else {
bail!("approval {id} not found");
};
if row.status != "pending" {
bail!("approval {id} is {}, not pending", row.status);
}
let resolved_at = Utc::now().timestamp();
let note = format!("cancelled by {canceller}");
tx.execute(
"UPDATE approvals SET status = 'cancelled', resolved_at = ?1, note = ?2 WHERE id = ?3",
params![resolved_at, note, id],
)?;
tx.commit()?;
Ok(Approval {
id,
agent: row.agent,
kind: kind_from_str(&row.kind)?,
commit_ref: row.commit_ref,
requested_at: hive_sh4re::wire_time::from_secs(row.requested_at),
status: ApprovalStatus::Cancelled,
resolved_at: Some(hive_sh4re::wire_time::from_secs(resolved_at)),
note: Some(note),
description: row.description,
})
}
/// Mark every pending approval for `agent` as failed (returns rows affected).
/// Used by `destroy` to clear the queue of an agent that no longer exists.
pub fn fail_pending_for_agent(&self, agent: &str, note: &str) -> Result<usize> {
let conn = self.conn.lock().unwrap();
let n = conn.execute(
"UPDATE approvals SET status = 'failed', resolved_at = ?1, note = ?2
WHERE agent = ?3 AND status = 'pending'",
params![Utc::now().timestamp(), note, agent],
)?;
Ok(n)
}
}
/// Columns needed to rebuild an [`Approval`] after a status transition,
/// shared by `mark_approved` / `mark_cancelled`. Replaces a 7-field
/// tuple that tripped `clippy::type_complexity` and was duplicated
/// across both callers (one suppressed the lint, the other aliased the
/// tuple) — one named projection + mapper now backs both.
struct ApprovalLookup {
agent: hive_types::Ident,
kind: String,
commit_ref: String,
requested_at: i64,
status: String,
description: Option<String>,
}
impl ApprovalLookup {
/// The single-row lookup by id (`?1`). Column order matches
/// [`ApprovalLookup::from_row`].
const SELECT: &str = "SELECT agent, kind, commit_ref, requested_at, status, description \
FROM approvals WHERE id = ?1";
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 {
agent,
kind: row.get(1)?,
commit_ref: row.get(2)?,
requested_at: row.get(3)?,
status: row.get(4)?,
description: row.get(5)?,
})
}
}
/// Collect approval rows, dropping any that fail to deserialize. A single
/// malformed / unknown-kind row must never blank the whole list:
/// `collect::<Result<Vec>>()` is all-or-nothing, so one bad row used to make
/// `pending()` / `recent_resolved()` error out wholesale — the dashboard then
/// rendered an empty approvals queue.
///
/// Drops are aggregated by error message and logged **once per call** rather
/// than one line per row: a batch of legacy unknown-kind rows (e.g. the
/// retired `apply_commit` approvals from the removed non-PR config flow) sit
/// `pending` forever and would otherwise flood the journal with an identical
/// warning on every dashboard render. Aggregating keeps the signal (how many,
/// which error) without the flood.
fn collect_lenient(rows: impl Iterator<Item = rusqlite::Result<Approval>>) -> Vec<Approval> {
let mut out = Vec::new();
let mut dropped: std::collections::HashMap<String, usize> = std::collections::HashMap::new();
for r in rows {
match r {
Ok(a) => out.push(a),
Err(e) => *dropped.entry(e.to_string()).or_default() += 1,
}
}
if !dropped.is_empty() {
let total: usize = dropped.values().sum();
tracing::warn!(
dropped = total,
by_error = ?dropped,
"skipped unparseable approval rows"
);
}
out
}
fn row_to_approval(row: &rusqlite::Row<'_>) -> rusqlite::Result<Approval> {
// Column order: id, agent, kind, commit_ref, requested_at, status, resolved_at, note, description.
let kind: String = row.get(2)?;
let kind = match kind.as_str() {
"update_meta_inputs" => ApprovalKind::UpdateMetaInputs,
"schedule_prompt" => ApprovalKind::SchedulePrompt,
other => {
return Err(rusqlite::Error::FromSqlConversionFailure(
2,
rusqlite::types::Type::Text,
format!("unknown approval kind '{other}'").into(),
));
}
};
let status: String = row.get(5)?;
let status = match status.as_str() {
"pending" => ApprovalStatus::Pending,
"approved" => ApprovalStatus::Approved,
"denied" => ApprovalStatus::Denied,
"failed" => ApprovalStatus::Failed,
"cancelled" => ApprovalStatus::Cancelled,
other => {
return Err(rusqlite::Error::FromSqlConversionFailure(
5,
rusqlite::types::Type::Text,
format!("unknown approval status '{other}'").into(),
));
}
};
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 {
id: row.get(0)?,
agent,
kind,
commit_ref: row.get(3)?,
requested_at: hive_sh4re::wire_time::from_secs(row.get(4)?),
status,
resolved_at: row
.get::<_, Option<i64>>(6)?
.map(hive_sh4re::wire_time::from_secs),
note: row.get(7)?,
description: row.get(8)?,
})
}
fn kind_from_str(s: &str) -> Result<ApprovalKind> {
Ok(match s {
"update_meta_inputs" => ApprovalKind::UpdateMetaInputs,
"schedule_prompt" => ApprovalKind::SchedulePrompt,
other => bail!("unknown approval kind '{other}'"),
})
}
#[cfg(test)]
mod tests {
use super::*;
use hive_sh4re::approvals::ApprovalKind;
fn open_temp() -> (tempfile::TempDir, std::path::PathBuf, Approvals) {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("approvals.sqlite");
let db = Approvals::open(&path).expect("open approvals db");
(dir, path, db)
}
#[test]
fn mixed_kinds_all_listed() {
let (_dir, _path, db) = open_temp();
db.submit_kind("b", ApprovalKind::SchedulePrompt, "", None, "b")
.unwrap();
db.submit_kind("c", ApprovalKind::UpdateMetaInputs, "[]", None, "c")
.unwrap();
let pending = db.pending().expect("pending");
assert_eq!(pending.len(), 2, "both kinds must be visible");
}
#[test]
fn mark_cancelled_transitions_pending_row() {
// Manager withdraws a pending approval. Row leaves pending(),
// shows up in recent_resolved() with the cancelled status + a
// "cancelled by <who>" note.
let (_dir, _path, db) = open_temp();
let id = db
.submit_kind(
"bitburner",
ApprovalKind::SchedulePrompt,
"{}",
Some("test"),
"bitburner",
)
.unwrap();
let row = db.mark_cancelled(id, "manager").expect("cancel");
assert_eq!(row.id, id);
assert!(matches!(row.status, ApprovalStatus::Cancelled));
assert_eq!(row.note.as_deref(), Some("cancelled by manager"));
assert!(row.resolved_at.is_some());
assert!(db.pending().unwrap().is_empty(), "row leaves pending");
let resolved = db.recent_resolved(10).unwrap();
assert_eq!(resolved.len(), 1);
assert!(matches!(resolved[0].status, ApprovalStatus::Cancelled));
}
#[test]
fn mark_cancelled_refuses_already_resolved_row() {
// Once approved/denied/failed/cancelled the resolution is
// final — re-cancelling errors instead of silently overwriting.
let (_dir, _path, db) = open_temp();
let id = db
.submit_kind("a", ApprovalKind::SchedulePrompt, "{}", None, "a")
.unwrap();
db.mark_cancelled(id, "manager").expect("first cancel");
let err = db
.mark_cancelled(id, "manager")
.expect_err("second cancel must fail");
assert!(err.to_string().contains("not pending"), "{err}");
}
#[test]
fn unknown_kind_row_is_skipped_not_fatal() {
// A single malformed / future-kind row must not blank the
// whole list — collect_lenient skips it instead of failing.
let (_dir, path, db) = open_temp();
let good = db
.submit_kind("good", ApprovalKind::SchedulePrompt, "{}", None, "good")
.unwrap();
let raw = Connection::open(&path).unwrap();
raw.execute(
"INSERT INTO approvals (agent, kind, commit_ref, requested_at, status)
VALUES ('weird', 'from_the_future', '', 0, 'pending')",
[],
)
.unwrap();
let pending = db
.pending()
.expect("pending() must survive an unparseable row");
assert_eq!(pending.len(), 1);
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::SchedulePrompt, "{}", 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', 'schedule_prompt', '', 0, 'pending')",
[],
)
.unwrap();
let legacy_id = raw.last_insert_rowid();
assert_eq!(db.submitter_of(legacy_id).unwrap(), None);
}
/// A database written before v5 still holds `merge_config_pr` rows,
/// pending and resolved. Opening it drops them and keeps every other row,
/// so the lists and `get` read cleanly.
#[test]
fn a_pre_v5_merge_config_pr_row_is_dropped_on_open() {
let (_dir, path, db) = open_temp();
let kept = db
.submit_kind("iris", ApprovalKind::SchedulePrompt, "{}", None, "iris")
.unwrap();
drop(db);
let raw = Connection::open(&path).unwrap();
raw.execute_batch(
"INSERT INTO approvals (agent, kind, commit_ref, requested_at, status, fetched_sha)
VALUES ('iris', 'merge_config_pr', '42', 0, 'pending', 'abc123');
INSERT INTO approvals (agent, kind, commit_ref, requested_at, status, resolved_at)
VALUES ('iris', 'merge_config_pr', '41', 0, 'approved', 1);
UPDATE schema_versions SET version = 4 WHERE store = 'approvals';",
)
.unwrap();
let old: i64 = raw.last_insert_rowid();
drop(raw);
let db = Approvals::open(&path).expect("a pre-v5 db opens");
let pending = db.pending().expect("pending");
assert_eq!(pending.len(), 1);
assert_eq!(pending[0].id, kept);
assert!(db.recent_resolved(10).unwrap().is_empty());
assert!(db.get(old).unwrap().is_none());
}
}