clippy: zero pedantic warnings across the tree
This commit is contained in:
parent
690cb5ab5b
commit
f9f1346eae
20 changed files with 71 additions and 61 deletions
|
|
@ -148,6 +148,7 @@ async fn main() -> Result<()> {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[allow(clippy::too_many_arguments, clippy::similar_names)]
|
||||||
async fn serve(
|
async fn serve(
|
||||||
socket: &Path,
|
socket: &Path,
|
||||||
interval: Duration,
|
interval: Duration,
|
||||||
|
|
@ -245,13 +246,13 @@ async fn serve(
|
||||||
s.record(&row);
|
s.record(&row);
|
||||||
}
|
}
|
||||||
|
|
||||||
// After turn completes, check if there are pending messages waiting.
|
// After turn completes, log whether messages arrived during
|
||||||
// If so, immediately process them instead of blocking on recv().
|
// the turn — the outer loop will iterate back to recv() on
|
||||||
// This ensures messages queued during the turn are processed ASAP.
|
// its own (the Empty-arm sleep only fires when recv
|
||||||
|
// actually returned Empty), so no explicit continue needed.
|
||||||
let pending = inbox_unread(socket).await;
|
let pending = inbox_unread(socket).await;
|
||||||
if pending > 0 {
|
if pending > 0 {
|
||||||
tracing::info!(%pending, "pending messages after turn; fetching next");
|
tracing::info!(%pending, "pending messages after turn; fetching next");
|
||||||
continue; // Loop back to recv() immediately instead of sleeping
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Ok(AgentResponse::Empty) => {
|
Ok(AgentResponse::Empty) => {
|
||||||
|
|
@ -409,10 +410,11 @@ async fn fetch_agent_post_turn_counts(socket: &Path) -> (Option<u64>, Option<u64
|
||||||
(threads, reminders)
|
(threads, reminders)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Assemble a TurnStatRow from the harness's per-turn state. Shared
|
/// Assemble a `TurnStatRow` from the harness's per-turn state. Shared
|
||||||
/// shape between the agent + manager bin loops (each lives in its own
|
/// shape between the agent + manager bin loops (each lives in its own
|
||||||
/// crate root so this helper is duplicated; the savings of a shared
|
/// crate root so this helper is duplicated; the savings of a shared
|
||||||
/// module aren't worth the cross-crate ceremony at this size).
|
/// module aren't worth the cross-crate ceremony at this size).
|
||||||
|
#[allow(clippy::too_many_arguments)]
|
||||||
fn build_row(
|
fn build_row(
|
||||||
started_at: i64,
|
started_at: i64,
|
||||||
ended_at: i64,
|
ended_at: i64,
|
||||||
|
|
|
||||||
|
|
@ -206,12 +206,12 @@ async fn serve(
|
||||||
);
|
);
|
||||||
s.record(&row);
|
s.record(&row);
|
||||||
}
|
}
|
||||||
// Check for messages that arrived during the turn and loop
|
// Check for messages that arrived during the turn so we
|
||||||
// immediately if any are waiting — mirrors hive-ag3nt behaviour.
|
// surface "draining" in the logs. The loop will already
|
||||||
|
// re-iterate from here — no explicit continue needed.
|
||||||
let pending = inbox_unread(socket).await;
|
let pending = inbox_unread(socket).await;
|
||||||
if pending > 0 {
|
if pending > 0 {
|
||||||
tracing::info!(%pending, "pending messages after turn; fetching next");
|
tracing::info!(%pending, "pending messages after turn; fetching next");
|
||||||
continue;
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Ok(ManagerResponse::Empty) => {
|
Ok(ManagerResponse::Empty) => {
|
||||||
|
|
@ -333,8 +333,9 @@ async fn fetch_manager_post_turn_counts(socket: &Path) -> (Option<u64>, Option<u
|
||||||
(threads, reminders)
|
(threads, reminders)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Manager flavour of the agent's build_row helper. Duplicated rather
|
/// Manager flavour of the agent's `build_row` helper. Duplicated rather
|
||||||
/// than shared to keep each bin self-contained at this size.
|
/// than shared to keep each bin self-contained at this size.
|
||||||
|
#[allow(clippy::too_many_arguments)]
|
||||||
fn build_row(
|
fn build_row(
|
||||||
started_at: i64,
|
started_at: i64,
|
||||||
ended_at: i64,
|
ended_at: i64,
|
||||||
|
|
|
||||||
|
|
@ -234,7 +234,7 @@ impl TokenUsage {
|
||||||
if v.get("type").and_then(|t| t.as_str()) != Some("result") {
|
if v.get("type").and_then(|t| t.as_str()) != Some("result") {
|
||||||
return None;
|
return None;
|
||||||
}
|
}
|
||||||
Self::from_usage_obj(v.get("usage")?)
|
Some(Self::from_usage_obj(v.get("usage")?))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Parse usage from a per-inference `assistant` event's
|
/// Parse usage from a per-inference `assistant` event's
|
||||||
|
|
@ -245,17 +245,17 @@ impl TokenUsage {
|
||||||
if v.get("type").and_then(|t| t.as_str()) != Some("assistant") {
|
if v.get("type").and_then(|t| t.as_str()) != Some("assistant") {
|
||||||
return None;
|
return None;
|
||||||
}
|
}
|
||||||
Self::from_usage_obj(v.get("message")?.get("usage")?)
|
Some(Self::from_usage_obj(v.get("message")?.get("usage")?))
|
||||||
}
|
}
|
||||||
|
|
||||||
fn from_usage_obj(u: &serde_json::Value) -> Option<Self> {
|
fn from_usage_obj(u: &serde_json::Value) -> Self {
|
||||||
let field = |k: &str| u.get(k).and_then(serde_json::Value::as_u64).unwrap_or(0);
|
let field = |k: &str| u.get(k).and_then(serde_json::Value::as_u64).unwrap_or(0);
|
||||||
Some(Self {
|
Self {
|
||||||
input_tokens: field("input_tokens"),
|
input_tokens: field("input_tokens"),
|
||||||
output_tokens: field("output_tokens"),
|
output_tokens: field("output_tokens"),
|
||||||
cache_read_input_tokens: field("cache_read_input_tokens"),
|
cache_read_input_tokens: field("cache_read_input_tokens"),
|
||||||
cache_creation_input_tokens: field("cache_creation_input_tokens"),
|
cache_creation_input_tokens: field("cache_creation_input_tokens"),
|
||||||
})
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -494,7 +494,7 @@ impl Bus {
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Broadcast a status flip (online / needs_login_*). Called by
|
/// Broadcast a status flip (online / `needs_login_*`). Called by
|
||||||
/// the bin entry points + `turn::wait_for_login` + the
|
/// the bin entry points + `turn::wait_for_login` + the
|
||||||
/// `post_login_*` handlers — every site that mutates the
|
/// `post_login_*` handlers — every site that mutates the
|
||||||
/// `Arc<Mutex<LoginState>>` should also call this so the web UI
|
/// `Arc<Mutex<LoginState>>` should also call this so the web UI
|
||||||
|
|
|
||||||
|
|
@ -308,11 +308,13 @@ where
|
||||||
/// content failure and the model would burn a turn retrying it.
|
/// content failure and the model would burn a turn retrying it.
|
||||||
pub fn annotate_retries(mut s: String, retries: u32) -> String {
|
pub fn annotate_retries(mut s: String, retries: u32) -> String {
|
||||||
if retries > 0 {
|
if retries > 0 {
|
||||||
|
use std::fmt::Write as _;
|
||||||
let suffix = if retries == 1 { "retry" } else { "retries" };
|
let suffix = if retries == 1 { "retry" } else { "retries" };
|
||||||
s.push_str(&format!(
|
let _ = write!(
|
||||||
|
s,
|
||||||
"\n\n(note: hive socket connect needed {retries} {suffix} — c0re likely \
|
"\n\n(note: hive socket connect needed {retries} {suffix} — c0re likely \
|
||||||
restarted. Your request did succeed on the final attempt; no action needed.)"
|
restarted. Your request did succeed on the final attempt; no action needed.)"
|
||||||
));
|
);
|
||||||
}
|
}
|
||||||
s
|
s
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -27,9 +27,8 @@ const AUTO_UPDATE_PATH: &str = "/etc/hyperhive/claude-plugins-auto-update.json";
|
||||||
/// "already exists" message and exits non-zero on some versions).
|
/// "already exists" message and exits non-zero on some versions).
|
||||||
/// Required before any `<plugin>@<marketplace>` install can resolve.
|
/// Required before any `<plugin>@<marketplace>` install can resolve.
|
||||||
async fn add_marketplaces() {
|
async fn add_marketplaces() {
|
||||||
let raw = match tokio::fs::read_to_string(MARKETPLACES_PATH).await {
|
let Ok(raw) = tokio::fs::read_to_string(MARKETPLACES_PATH).await else {
|
||||||
Ok(s) => s,
|
return;
|
||||||
Err(_) => return,
|
|
||||||
};
|
};
|
||||||
let sources: Vec<String> = match serde_json::from_str(&raw) {
|
let sources: Vec<String> = match serde_json::from_str(&raw) {
|
||||||
Ok(v) => v,
|
Ok(v) => v,
|
||||||
|
|
@ -107,9 +106,8 @@ async fn update_marketplaces() {
|
||||||
/// journald. The manager itself passes `None` — there's nobody above
|
/// journald. The manager itself passes `None` — there's nobody above
|
||||||
/// it to notify.
|
/// it to notify.
|
||||||
pub async fn install_configured(socket: &Path, notify_recipient: Option<&str>) {
|
pub async fn install_configured(socket: &Path, notify_recipient: Option<&str>) {
|
||||||
let raw = match tokio::fs::read_to_string(PLUGINS_PATH).await {
|
let Ok(raw) = tokio::fs::read_to_string(PLUGINS_PATH).await else {
|
||||||
Ok(s) => s,
|
return;
|
||||||
Err(_) => return,
|
|
||||||
};
|
};
|
||||||
let specs: Vec<String> = match serde_json::from_str(&raw) {
|
let specs: Vec<String> = match serde_json::from_str(&raw) {
|
||||||
Ok(v) => v,
|
Ok(v) => v,
|
||||||
|
|
|
||||||
|
|
@ -222,7 +222,12 @@ pub async fn compact_session(files: &TurnFiles, bus: &Bus) -> Result<()> {
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[allow(clippy::too_many_lines)]
|
||||||
async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<bool> {
|
async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<bool> {
|
||||||
|
// Keep the last STDERR_TAIL_LINES of stderr so a non-zero exit can
|
||||||
|
// include real context in the bail message (and downstream in the
|
||||||
|
// failure notification to the manager) instead of just "exit 1".
|
||||||
|
const STDERR_TAIL_LINES: usize = 20;
|
||||||
let model = bus.model();
|
let model = bus.model();
|
||||||
let resume = !bus.take_skip_continue();
|
let resume = !bus.take_skip_continue();
|
||||||
if !resume {
|
if !resume {
|
||||||
|
|
@ -311,10 +316,6 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<bool>
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
// Keep the last STDERR_TAIL_LINES of stderr so a non-zero exit can
|
|
||||||
// include real context in the bail message (and downstream in the
|
|
||||||
// failure notification to the manager) instead of just "exit 1".
|
|
||||||
const STDERR_TAIL_LINES: usize = 20;
|
|
||||||
let stderr_tail: Arc<Mutex<VecDeque<String>>> =
|
let stderr_tail: Arc<Mutex<VecDeque<String>>> =
|
||||||
Arc::new(Mutex::new(VecDeque::with_capacity(STDERR_TAIL_LINES)));
|
Arc::new(Mutex::new(VecDeque::with_capacity(STDERR_TAIL_LINES)));
|
||||||
let tail_clone = stderr_tail.clone();
|
let tail_clone = stderr_tail.clone();
|
||||||
|
|
|
||||||
|
|
@ -1,8 +1,8 @@
|
||||||
//! Per-turn analytics sink. One sqlite row per claude turn captures:
|
//! Per-turn analytics sink. One sqlite row per claude turn captures:
|
||||||
//! identity (model, wake_from, result_kind), timing (started_at,
|
//! identity (`model`, `wake_from`, `result_kind`), timing (`started_at`,
|
||||||
//! ended_at, duration_ms), cost (token counts), behaviour (tool-call
|
//! `ended_at`, `duration_ms`), cost (token counts), behaviour (tool-call
|
||||||
//! count + per-tool breakdown), and post-turn snapshot metrics
|
//! count + per-tool breakdown), and post-turn snapshot metrics
|
||||||
//! (open_threads_count, open_reminders_count).
|
//! (`open_threads_count`, `open_reminders_count`).
|
||||||
//!
|
//!
|
||||||
//! Lives next to `hyperhive-events.sqlite` in the agent's state dir
|
//! Lives next to `hyperhive-events.sqlite` in the agent's state dir
|
||||||
//! so the host-side state vacuum sweep can reach both. Schema is
|
//! so the host-side state vacuum sweep can reach both. Schema is
|
||||||
|
|
@ -65,7 +65,7 @@ const MIGRATIONS: &[&str] = &[
|
||||||
|
|
||||||
/// One row to be inserted. `Option`-wrapped fields default to NULL
|
/// One row to be inserted. `Option`-wrapped fields default to NULL
|
||||||
/// when the harness couldn't gather them (e.g. socket roundtrip for
|
/// when the harness couldn't gather them (e.g. socket roundtrip for
|
||||||
/// open_threads failed) so a partial row beats no row.
|
/// `open_threads` failed) so a partial row beats no row.
|
||||||
#[derive(Debug, Clone)]
|
#[derive(Debug, Clone)]
|
||||||
pub struct TurnStatRow {
|
pub struct TurnStatRow {
|
||||||
pub started_at: i64,
|
pub started_at: i64,
|
||||||
|
|
|
||||||
|
|
@ -527,11 +527,8 @@ async fn post_compact(State(state): State<AppState>) -> Response {
|
||||||
let lock = state.turn_lock.clone();
|
let lock = state.turn_lock.clone();
|
||||||
// Reject immediately if a normal turn is in flight — concurrent access
|
// Reject immediately if a normal turn is in flight — concurrent access
|
||||||
// to the claude session is unsafe and produces garbled output.
|
// to the claude session is unsafe and produces garbled output.
|
||||||
let guard = match lock.try_lock_owned() {
|
let Ok(guard) = lock.try_lock_owned() else {
|
||||||
Ok(g) => g,
|
|
||||||
Err(_) => {
|
|
||||||
return error_response("turn in flight — wait for it to finish before compacting");
|
return error_response("turn in flight — wait for it to finish before compacting");
|
||||||
}
|
|
||||||
};
|
};
|
||||||
let bus = state.bus.clone();
|
let bus = state.bus.clone();
|
||||||
let files = state.files.clone();
|
let files = state.files.clone();
|
||||||
|
|
|
||||||
|
|
@ -324,7 +324,7 @@ pub async fn destroy(coord: &Arc<Coordinator>, name: &str, purge: bool) -> Resul
|
||||||
tracing::info!(%name, purge, "destroy");
|
tracing::info!(%name, purge, "destroy");
|
||||||
// Guard auto-clears on the success path's final scope exit and on
|
// Guard auto-clears on the success path's final scope exit and on
|
||||||
// every early-return / cancellation along the way.
|
// every early-return / cancellation along the way.
|
||||||
let _guard = coord.transient_guard(name, TransientKind::Destroying);
|
let guard = coord.transient_guard(name, TransientKind::Destroying);
|
||||||
lifecycle::destroy(name).await?;
|
lifecycle::destroy(name).await?;
|
||||||
coord.unregister_agent(name);
|
coord.unregister_agent(name);
|
||||||
let runtime = Coordinator::agent_dir(name);
|
let runtime = Coordinator::agent_dir(name);
|
||||||
|
|
@ -359,7 +359,7 @@ pub async fn destroy(coord: &Arc<Coordinator>, name: &str, purge: bool) -> Resul
|
||||||
"agent destroyed"
|
"agent destroyed"
|
||||||
},
|
},
|
||||||
);
|
);
|
||||||
drop(_guard);
|
drop(guard);
|
||||||
coord.notify_manager(&HelperEvent::Destroyed {
|
coord.notify_manager(&HelperEvent::Destroyed {
|
||||||
agent: name.to_owned(),
|
agent: name.to_owned(),
|
||||||
});
|
});
|
||||||
|
|
|
||||||
|
|
@ -376,7 +376,11 @@ mod tests {
|
||||||
fn auto_reminder_path_format() {
|
fn auto_reminder_path_format() {
|
||||||
let p = auto_reminder_path("damocles");
|
let p = auto_reminder_path("damocles");
|
||||||
assert!(p.starts_with("/agents/damocles/state/reminders/auto-"));
|
assert!(p.starts_with("/agents/damocles/state/reminders/auto-"));
|
||||||
assert!(p.ends_with(".md"));
|
assert!(
|
||||||
|
std::path::Path::new(&p)
|
||||||
|
.extension()
|
||||||
|
.is_some_and(|ext| ext.eq_ignore_ascii_case("md"))
|
||||||
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
|
|
|
||||||
|
|
@ -168,8 +168,11 @@ impl Approvals {
|
||||||
|
|
||||||
/// Mark pending -> approved (or fail if not pending). Returns the (now-updated)
|
/// 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.
|
/// approval so the caller can run the action and pass the agent name.
|
||||||
|
#[allow(clippy::type_complexity)]
|
||||||
pub fn mark_approved(&self, id: i64) -> Result<Approval> {
|
pub fn mark_approved(&self, id: i64) -> Result<Approval> {
|
||||||
let conn = self.conn.lock().unwrap();
|
let conn = self.conn.lock().unwrap();
|
||||||
|
// Row shape: (agent, kind, commit_ref, requested_at, status,
|
||||||
|
// fetched_sha, description).
|
||||||
let current: Option<(
|
let current: Option<(
|
||||||
String,
|
String,
|
||||||
String,
|
String,
|
||||||
|
|
|
||||||
|
|
@ -63,7 +63,7 @@ pub async fn rebuild_agent(coord: &Arc<Coordinator>, name: &str, current_rev: &s
|
||||||
// lifecycle::rebuild. Dashboard rebuilds already do this via
|
// lifecycle::rebuild. Dashboard rebuilds already do this via
|
||||||
// lifecycle_action; this catches the auto-update scan + any
|
// lifecycle_action; this catches the auto-update scan + any
|
||||||
// other direct caller.
|
// other direct caller.
|
||||||
let _guard = coord.transient_guard(name, crate::coordinator::TransientKind::Rebuilding);
|
let guard = coord.transient_guard(name, crate::coordinator::TransientKind::Rebuilding);
|
||||||
let result = lifecycle::rebuild(
|
let result = lifecycle::rebuild(
|
||||||
name,
|
name,
|
||||||
&coord.hyperhive_flake,
|
&coord.hyperhive_flake,
|
||||||
|
|
@ -75,7 +75,7 @@ pub async fn rebuild_agent(coord: &Arc<Coordinator>, name: &str, current_rev: &s
|
||||||
&coord.operator_pronouns,
|
&coord.operator_pronouns,
|
||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
drop(_guard);
|
drop(guard);
|
||||||
match &result {
|
match &result {
|
||||||
Ok(()) => {
|
Ok(()) => {
|
||||||
if let Err(e) = std::fs::write(rev_marker_path(name), current_rev) {
|
if let Err(e) = std::fs::write(rev_marker_path(name), current_rev) {
|
||||||
|
|
|
||||||
|
|
@ -512,7 +512,7 @@ impl Broker {
|
||||||
|
|
||||||
/// Clear the failure state on a pending reminder so the
|
/// Clear the failure state on a pending reminder so the
|
||||||
/// scheduler picks it up again. No-op when the row is already
|
/// scheduler picks it up again. No-op when the row is already
|
||||||
/// fresh (attempt_count == 0). Returns the number of rows
|
/// fresh (`attempt_count == 0`). Returns the number of rows
|
||||||
/// affected so callers can distinguish "retried" from "no
|
/// affected so callers can distinguish "retried" from "no
|
||||||
/// such pending reminder" (already delivered, or wrong id).
|
/// such pending reminder" (already delivered, or wrong id).
|
||||||
pub fn reset_reminder_failure(&self, id: i64) -> Result<usize> {
|
pub fn reset_reminder_failure(&self, id: i64) -> Result<usize> {
|
||||||
|
|
|
||||||
|
|
@ -215,6 +215,7 @@ impl Coordinator {
|
||||||
/// already have an authoritative timestamp from the db update,
|
/// already have an authoritative timestamp from the db update,
|
||||||
/// the tiny skew between "row updated" and "event emitted" is
|
/// the tiny skew between "row updated" and "event emitted" is
|
||||||
/// presentation-only and doesn't matter to clients.
|
/// presentation-only and doesn't matter to clients.
|
||||||
|
#[allow(clippy::too_many_arguments)]
|
||||||
pub fn emit_approval_resolved(
|
pub fn emit_approval_resolved(
|
||||||
&self,
|
&self,
|
||||||
id: i64,
|
id: i64,
|
||||||
|
|
@ -247,6 +248,7 @@ impl Coordinator {
|
||||||
/// both operator-targeted (`target = None`) and peer-to-peer
|
/// both operator-targeted (`target = None`) and peer-to-peer
|
||||||
/// (`target = Some(agent)`) threads — the dashboard surfaces
|
/// (`target = Some(agent)`) threads — the dashboard surfaces
|
||||||
/// both, distinguishing visually + offering operator override.
|
/// both, distinguishing visually + offering operator override.
|
||||||
|
#[allow(clippy::too_many_arguments)]
|
||||||
pub fn emit_question_added(
|
pub fn emit_question_added(
|
||||||
&self,
|
&self,
|
||||||
id: i64,
|
id: i64,
|
||||||
|
|
@ -318,7 +320,7 @@ impl Coordinator {
|
||||||
/// resolves to — lifecycle ops, destroy, approve (post-spawn),
|
/// resolves to — lifecycle ops, destroy, approve (post-spawn),
|
||||||
/// rebuild, meta-update, and the crash-watcher's periodic poll.
|
/// rebuild, meta-update, and the crash-watcher's periodic poll.
|
||||||
/// Cheap when nothing changed (one `nixos-container list` + a
|
/// Cheap when nothing changed (one `nixos-container list` + a
|
||||||
/// HashMap diff + zero emits).
|
/// `HashMap` diff + zero emits).
|
||||||
pub async fn rescan_containers_and_emit(self: &Arc<Self>) {
|
pub async fn rescan_containers_and_emit(self: &Arc<Self>) {
|
||||||
let fresh = container_view::build_all(self).await;
|
let fresh = container_view::build_all(self).await;
|
||||||
let mut last = self.last_containers.lock().await;
|
let mut last = self.last_containers.lock().await;
|
||||||
|
|
|
||||||
|
|
@ -187,7 +187,7 @@ struct StateSnapshot {
|
||||||
meta_inputs: Vec<MetaInputView>,
|
meta_inputs: Vec<MetaInputView>,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// OpQuestion + computed `question_refs` / `answer_refs`. Built
|
/// `OpQuestion` + computed `question_refs` / `answer_refs`. Built
|
||||||
/// from the snapshot read; the live channel attaches the same
|
/// from the snapshot read; the live channel attaches the same
|
||||||
/// fields directly on `QuestionAdded` / `QuestionResolved`.
|
/// fields directly on `QuestionAdded` / `QuestionResolved`.
|
||||||
#[derive(Serialize)]
|
#[derive(Serialize)]
|
||||||
|
|
@ -1207,7 +1207,7 @@ pub(crate) fn emit_meta_inputs_snapshot(coord: &Coordinator) {
|
||||||
/// natural-language text but aren't part of the path itself. Any
|
/// natural-language text but aren't part of the path itself. Any
|
||||||
/// token starting with `/agents/`, `/shared/`, or
|
/// token starting with `/agents/`, `/shared/`, or
|
||||||
/// `/var/lib/hyperhive/{agents,shared}/` is a candidate. The
|
/// `/var/lib/hyperhive/{agents,shared}/` is a candidate. The
|
||||||
/// allow-list + is_file check happens via the same
|
/// allow-list + `is_file` check happens via the same
|
||||||
/// `resolve_state_path` helper the read endpoint uses, so the
|
/// `resolve_state_path` helper the read endpoint uses, so the
|
||||||
/// security rules can't drift.
|
/// security rules can't drift.
|
||||||
pub(crate) fn scan_validated_paths(body: &str) -> Vec<String> {
|
pub(crate) fn scan_validated_paths(body: &str) -> Vec<String> {
|
||||||
|
|
@ -1222,7 +1222,7 @@ pub(crate) fn scan_validated_paths(body: &str) -> Vec<String> {
|
||||||
// Trim trailing natural-language punctuation that wouldn't
|
// Trim trailing natural-language punctuation that wouldn't
|
||||||
// be part of any real path. Inline rather than via a regex
|
// be part of any real path. Inline rather than via a regex
|
||||||
// dep — the set is small and the call is hot.
|
// dep — the set is small and the call is hot.
|
||||||
let token = raw.trim_end_matches(|c: char| matches!(c, ',' | ';' | ':' | ')' | ']' | '}' | '.' | '\'' | '"'));
|
let token = raw.trim_end_matches([',', ';', ':', ')', ']', '}', '.', '\'', '"']);
|
||||||
if token.is_empty() {
|
if token.is_empty() {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
@ -1265,9 +1265,8 @@ async fn get_state_file(
|
||||||
let body_bytes = if truncated { &bytes[..MAX_BYTES] } else { &bytes[..] };
|
let body_bytes = if truncated { &bytes[..MAX_BYTES] } else { &bytes[..] };
|
||||||
let mut body = String::from_utf8_lossy(body_bytes).into_owned();
|
let mut body = String::from_utf8_lossy(body_bytes).into_owned();
|
||||||
if truncated {
|
if truncated {
|
||||||
body.push_str(&format!(
|
use std::fmt::Write as _;
|
||||||
"\n\n--- truncated at {MAX_BYTES} of {size} bytes ---\n"
|
let _ = write!(body, "\n\n--- truncated at {MAX_BYTES} of {size} bytes ---\n");
|
||||||
));
|
|
||||||
}
|
}
|
||||||
([("content-type", "text/plain; charset=utf-8")], body).into_response()
|
([("content-type", "text/plain; charset=utf-8")], body).into_response()
|
||||||
}
|
}
|
||||||
|
|
@ -1575,9 +1574,9 @@ where
|
||||||
Fut: std::future::Future<Output = anyhow::Result<()>>,
|
Fut: std::future::Future<Output = anyhow::Result<()>>,
|
||||||
{
|
{
|
||||||
let logical = strip_container_prefix(name);
|
let logical = strip_container_prefix(name);
|
||||||
let _guard = state.coord.transient_guard(&logical, kind);
|
let guard = state.coord.transient_guard(&logical, kind);
|
||||||
let result = body(logical.clone()).await;
|
let result = body(logical.clone()).await;
|
||||||
drop(_guard);
|
drop(guard);
|
||||||
match result {
|
match result {
|
||||||
Ok(()) => {
|
Ok(()) => {
|
||||||
extra(state, &logical);
|
extra(state, &logical);
|
||||||
|
|
|
||||||
|
|
@ -146,15 +146,15 @@ pub enum DashboardEvent {
|
||||||
/// Clients drop the spinner row.
|
/// Clients drop the spinner row.
|
||||||
TransientCleared { seq: u64, name: String },
|
TransientCleared { seq: u64, name: String },
|
||||||
/// One container row changed — new container appeared (post-spawn
|
/// One container row changed — new container appeared (post-spawn
|
||||||
/// finalise), an existing one flipped running/needs_update/sha,
|
/// finalise), an existing one flipped `running` / `needs_update` /
|
||||||
/// etc. Clients upsert by `container.name`. Payload carries the
|
/// `sha`, etc. Clients upsert by `container.name`. Payload carries
|
||||||
/// full row so cold-loaded clients and event-driven clients
|
/// the full row so cold-loaded clients and event-driven clients
|
||||||
/// converge on the same render.
|
/// converge on the same render.
|
||||||
///
|
///
|
||||||
/// Fired by `Coordinator::rescan_containers_and_emit`, which diffs
|
/// Fired by `Coordinator::rescan_containers_and_emit`, which diffs
|
||||||
/// a fresh `nixos-container list`–derived snapshot against the
|
/// a fresh `nixos-container list`–derived snapshot against the
|
||||||
/// last one cached on the coordinator. Mutation sites (lifecycle
|
/// last one cached on the coordinator. Mutation sites (lifecycle
|
||||||
/// endpoints, actions::destroy / approve, crash_watch's poll loop)
|
/// endpoints, `actions::destroy` / approve, `crash_watch`'s poll loop)
|
||||||
/// call the rescan after their work lands.
|
/// call the rescan after their work lands.
|
||||||
ContainerStateChanged {
|
ContainerStateChanged {
|
||||||
seq: u64,
|
seq: u64,
|
||||||
|
|
|
||||||
|
|
@ -165,6 +165,7 @@ async fn ensure_user_exists(name: &str, admin: bool) -> Result<()> {
|
||||||
/// monotonic clock so re-issuing doesn't collide with an existing
|
/// monotonic clock so re-issuing doesn't collide with an existing
|
||||||
/// token of the same name in the DB.
|
/// token of the same name in the DB.
|
||||||
async fn mint_and_persist_token(name: &str, path: &Path) -> Result<()> {
|
async fn mint_and_persist_token(name: &str, path: &Path) -> Result<()> {
|
||||||
|
use std::os::unix::fs::PermissionsExt;
|
||||||
let token_name = format!(
|
let token_name = format!(
|
||||||
"{TOKEN_NAME_PREFIX}-{}",
|
"{TOKEN_NAME_PREFIX}-{}",
|
||||||
std::time::SystemTime::now()
|
std::time::SystemTime::now()
|
||||||
|
|
@ -190,7 +191,6 @@ async fn mint_and_persist_token(name: &str, path: &Path) -> Result<()> {
|
||||||
}
|
}
|
||||||
std::fs::write(path, format!("{token}\n"))
|
std::fs::write(path, format!("{token}\n"))
|
||||||
.with_context(|| format!("write token to {}", path.display()))?;
|
.with_context(|| format!("write token to {}", path.display()))?;
|
||||||
use std::os::unix::fs::PermissionsExt;
|
|
||||||
let _ = std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600));
|
let _ = std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600));
|
||||||
tracing::info!(%name, path = %path.display(), %token_name, "forge: persisted access token");
|
tracing::info!(%name, path = %path.display(), %token_name, "forge: persisted access token");
|
||||||
Ok(())
|
Ok(())
|
||||||
|
|
|
||||||
|
|
@ -234,9 +234,9 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
|
||||||
message: "update: hyperhive_flake has no canonical path".into(),
|
message: "update: hyperhive_flake has no canonical path".into(),
|
||||||
};
|
};
|
||||||
};
|
};
|
||||||
let _guard = coord.transient_guard(name, crate::coordinator::TransientKind::Rebuilding);
|
let guard = coord.transient_guard(name, crate::coordinator::TransientKind::Rebuilding);
|
||||||
let result = crate::auto_update::rebuild_agent(coord, name, ¤t_rev).await;
|
let result = crate::auto_update::rebuild_agent(coord, name, ¤t_rev).await;
|
||||||
drop(_guard);
|
drop(guard);
|
||||||
match result {
|
match result {
|
||||||
Ok(()) => {
|
Ok(()) => {
|
||||||
coord.kick_agent(name, "container rebuilt");
|
coord.kick_agent(name, "container rebuilt");
|
||||||
|
|
|
||||||
|
|
@ -102,9 +102,9 @@ pub async fn run(coord: &Arc<Coordinator>) -> Result<()> {
|
||||||
// update activation triggers. Without this, crash_watch
|
// update activation triggers. Without this, crash_watch
|
||||||
// would fire ContainerCrash for every agent here and the
|
// would fire ContainerCrash for every agent here and the
|
||||||
// manager would spuriously try to recover them.
|
// manager would spuriously try to recover them.
|
||||||
let _guard = coord.transient_guard(name, crate::coordinator::TransientKind::Rebuilding);
|
let guard = coord.transient_guard(name, crate::coordinator::TransientKind::Rebuilding);
|
||||||
let result = repoint_container(name).await;
|
let result = repoint_container(name).await;
|
||||||
drop(_guard);
|
drop(guard);
|
||||||
if let Err(e) = result {
|
if let Err(e) = result {
|
||||||
tracing::warn!(%name, error = ?e, "migration: container repoint failed");
|
tracing::warn!(%name, error = ?e, "migration: container repoint failed");
|
||||||
all_ok = false;
|
all_ok = false;
|
||||||
|
|
|
||||||
|
|
@ -284,7 +284,8 @@ impl OperatorQuestions {
|
||||||
ORDER BY answered_at DESC
|
ORDER BY answered_at DESC
|
||||||
LIMIT ?1",
|
LIMIT ?1",
|
||||||
)?;
|
)?;
|
||||||
let rows = stmt.query_map(params![limit as i64], row_to_question)?;
|
let limit_i = i64::try_from(limit).unwrap_or(i64::MAX);
|
||||||
|
let rows = stmt.query_map(params![limit_i], row_to_question)?;
|
||||||
rows.collect::<rusqlite::Result<Vec<_>>>()
|
rows.collect::<rusqlite::Result<Vec<_>>>()
|
||||||
.map_err(Into::into)
|
.map_err(Into::into)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue