From d6ea92085a23990f8a1c2da2ea66cbeca55071cf Mon Sep 17 00:00:00 2001 From: damocles Date: Sat, 1 Aug 2026 17:51:49 +0200 Subject: [PATCH 1/5] hive-sh4re: use chrono's clock feature for now_unix instead of hand-rolled SystemTime math --- Cargo.toml | 1 + hive-sh4re/src/wire_time.rs | 11 +++++------ 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 9b509513c..db61bb2d5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -45,6 +45,7 @@ axum = { version = "0.8", features = ["ws"] } base64 = "0.22" bcrypt = "0.19" chrono = { version = "0.4", default-features = false, features = [ + "clock", "serde", "std", ] } diff --git a/hive-sh4re/src/wire_time.rs b/hive-sh4re/src/wire_time.rs index 2d9529013..508507475 100644 --- a/hive-sh4re/src/wire_time.rs +++ b/hive-sh4re/src/wire_time.rs @@ -18,14 +18,13 @@ pub fn from_secs(secs: i64) -> DateTime { /// Current unix timestamp in seconds — the single definition behind /// every store's `created_at` / `sent_at` / … stamp (this module owns /// the epoch-seconds convention; a dozen local copies of this fn used -/// to float around both binaries). Clamps to 0 on a pre-epoch clock. +/// to float around both binaries). Uses chrono's `clock` feature +/// (already pulled in workspace-wide) instead of hand-rolled +/// `SystemTime` math, consistent with every other timestamp in this +/// module being a `chrono` type. #[must_use] pub fn now_unix() -> i64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .ok() - .and_then(|d| i64::try_from(d.as_secs()).ok()) - .unwrap_or(0) + Utc::now().timestamp() } #[cfg(test)] From 2122e23d8111acb1e10779eb8da1cdf3d345f0bd Mon Sep 17 00:00:00 2001 From: damocles Date: Sat, 1 Aug 2026 21:34:29 +0200 Subject: [PATCH 2/5] hive-sh4re/hive-bash-mcp/hive-agent: retype TaskFile timestamps to DateTime, drop now_unix from these crates --- Cargo.lock | 2 ++ hive-agent/Cargo.toml | 1 + hive-agent/src/vacuum.rs | 19 ++++++++++++------- hive-bash-mcp/Cargo.toml | 1 + hive-bash-mcp/src/mcp.rs | 11 ++++++----- hive-bash-mcp/src/runner.rs | 12 +++++------- hive-sh4re/src/lib.rs | 6 +++--- 7 files changed, 30 insertions(+), 22 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index bf7ac8006..0b69f7f0d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1562,6 +1562,7 @@ version = "0.1.0" dependencies = [ "anyhow", "axum", + "chrono", "clap", "futures-util", "hive-agent-sock", @@ -1622,6 +1623,7 @@ version = "0.1.0" dependencies = [ "anyhow", "axum", + "chrono", "clap", "hive-agent-sock", "hive-sh4re", diff --git a/hive-agent/Cargo.toml b/hive-agent/Cargo.toml index 3dd41c259..a4029b5c2 100644 --- a/hive-agent/Cargo.toml +++ b/hive-agent/Cargo.toml @@ -10,6 +10,7 @@ workspace = true [dependencies] anyhow.workspace = true axum.workspace = true +chrono.workspace = true reqwest.workspace = true hyper.workspace = true hyper-util.workspace = true diff --git a/hive-agent/src/vacuum.rs b/hive-agent/src/vacuum.rs index 1c77d86d9..fa37184fb 100644 --- a/hive-agent/src/vacuum.rs +++ b/hive-agent/src/vacuum.rs @@ -17,7 +17,6 @@ use std::path::Path; use std::time::Duration; -use hive_sh4re::wire_time::now_unix; use rusqlite::{Connection, Result, params}; /// How often the sweep runs. @@ -56,7 +55,8 @@ fn sweep_once() { let tasks_dir = harness.join("bash-tasks"); if tasks_dir.is_dir() { - let removed = vacuum_bash_tasks(&tasks_dir, now_unix() - BASH_KEEP_SECS); + let removed = + vacuum_bash_tasks(&tasks_dir, chrono::Utc::now().timestamp() - BASH_KEEP_SECS); if removed > 0 { tracing::info!(removed, "bash-tasks vacuum"); } @@ -92,14 +92,14 @@ fn sweep_once() { /// `Reminders` handle). fn vacuum_reminders(path: &Path) -> anyhow::Result { let store = crate::reminders::Reminders::open(path)?; - store.prune_delivered_older_than(now_unix() - REMINDER_KEEP_SECS) + store.prune_delivered_older_than(chrono::Utc::now().timestamp() - REMINDER_KEEP_SECS) } /// Reap acked todo rows older than [`TODO_ACKED_KEEP_SECS`] via the typed /// store API — same own-short-lived-connection shape as `vacuum_reminders`. fn vacuum_todos(path: &Path) -> anyhow::Result { let store = crate::todos::Todos::open(path)?; - store.reap_acked(now_unix() - TODO_ACKED_KEEP_SECS) + store.reap_acked(chrono::Utc::now().timestamp() - TODO_ACKED_KEEP_SECS) } /// Delete eligible bash-task trios in `dir`. Returns the count of `.json` @@ -140,10 +140,15 @@ fn should_delete(json_path: &Path, cutoff: i64) -> bool { if !TERMINAL_STATUSES.contains(&status) { return false; } + // `completed_at` serializes as an RFC3339 string (`TaskFile.completed_at` + // is `Option>`), not a bare epoch-seconds integer — parse + // it the same way, falling back to "never expired" on anything + // unparseable so a corrupt/legacy field never causes a premature delete. let completed_at = v .get("completed_at") - .and_then(serde_json::Value::as_i64) - .unwrap_or(i64::MAX); + .and_then(serde_json::Value::as_str) + .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok()) + .map_or(i64::MAX, |dt| dt.timestamp()); completed_at < cutoff } @@ -164,7 +169,7 @@ fn delete_trio(dir: &Path, stem: &str) { /// agent's `events.sqlite`. Returns the number of rows deleted. fn vacuum_events(path: &Path) -> Result { let conn = Connection::open(path)?; - let cutoff = now_unix() - STREAM_KEEP_SECS; + let cutoff = chrono::Utc::now().timestamp() - STREAM_KEEP_SECS; let removed = conn.execute( "DELETE FROM events WHERE kind = 'stream' AND ts < ?1", params![cutoff], diff --git a/hive-bash-mcp/Cargo.toml b/hive-bash-mcp/Cargo.toml index a11592d8d..0c4f36ebf 100644 --- a/hive-bash-mcp/Cargo.toml +++ b/hive-bash-mcp/Cargo.toml @@ -10,6 +10,7 @@ workspace = true [dependencies] anyhow.workspace = true axum.workspace = true +chrono.workspace = true clap.workspace = true hive-agent-sock.workspace = true hive-sh4re.workspace = true diff --git a/hive-bash-mcp/src/mcp.rs b/hive-bash-mcp/src/mcp.rs index 2653f39c9..59545a0ad 100644 --- a/hive-bash-mcp/src/mcp.rs +++ b/hive-bash-mcp/src/mcp.rs @@ -38,11 +38,12 @@ fn format_task(task: &TaskFile) -> String { let _ = write!(out, ", exit={code}"); } if let (Some(started), None) = (task.started_at, task.completed_at) { - let now = hive_sh4re::wire_time::now_unix(); - let _ = write!(out, ", running for {}s", now - started); + let elapsed = (chrono::Utc::now() - started).num_seconds(); + let _ = write!(out, ", running for {elapsed}s"); } if let (Some(completed), Some(started)) = (task.completed_at, task.started_at) { - let _ = write!(out, ", took {}s", completed - started); + let took = (completed - started).num_seconds(); + let _ = write!(out, ", took {took}s"); } let out_file = crate::paths::task_out(&task.id); @@ -295,8 +296,8 @@ mod status_hint_tests { cmd: "echo hi".to_owned(), timeout_secs: None, status, - created_at: 1, - started_at: Some(1), + created_at: chrono::DateTime::from_timestamp(1, 0).unwrap_or_default(), + started_at: Some(chrono::DateTime::from_timestamp(1, 0).unwrap_or_default()), completed_at: None, exit_code: None, stdout_tail: None, diff --git a/hive-bash-mcp/src/runner.rs b/hive-bash-mcp/src/runner.rs index 0ddb8874a..438bf4876 100644 --- a/hive-bash-mcp/src/runner.rs +++ b/hive-bash-mcp/src/runner.rs @@ -133,8 +133,6 @@ fn signal_group(pgid: Option, sig: i32) { // Helpers // --------------------------------------------------------------------------- -use hive_sh4re::wire_time::now_unix; - /// Generate a task ID: ``. #[must_use] pub fn new_task_id() -> String { @@ -271,7 +269,7 @@ pub fn submit_task(cmd: String, timeout_secs: Option, name: Option) cmd, timeout_secs, status: TaskStatus::Pending, - created_at: now_unix(), + created_at: chrono::Utc::now(), started_at: None, completed_at: None, exit_code: None, @@ -377,7 +375,7 @@ pub fn kill_task(id: &str, force: bool) -> (bool, bool) { && task.status == TaskStatus::Pending { task.status = TaskStatus::Killed; - task.completed_at = Some(now_unix()); + task.completed_at = Some(chrono::Utc::now()); let _ = write_task(&task); // A Pending task never reached Running, so it has no keyed "active" // todo to clear and (like the old code) fires no completion signal — @@ -435,7 +433,7 @@ async fn mark_interrupted(socket: &Path) { } tracing::warn!(id = %id, "bash_runner: marking interrupted task"); task.status = TaskStatus::Interrupted; - task.completed_at = Some(now_unix()); + task.completed_at = Some(chrono::Utc::now()); if let Err(e) = write_task(&task) { tracing::warn!(id = %id, error = ?e, "bash_runner: write interrupted state failed"); } @@ -490,7 +488,7 @@ async fn run_task(mut task: TaskFile, socket: &Path) { tracing::info!(id = %id, cmd = %task.cmd, "bash_runner: starting task"); task.status = TaskStatus::Running; - task.started_at = Some(now_unix()); + task.started_at = Some(chrono::Utc::now()); if let Err(e) = write_task(&task) { tracing::warn!(id = %id, error = ?e, "bash_runner: write running state failed"); } @@ -559,7 +557,7 @@ async fn run_task(mut task: TaskFile, socket: &Path) { }; task.status = status; - task.completed_at = Some(now_unix()); + task.completed_at = Some(chrono::Utc::now()); task.exit_code = exit_code; task.stdout_tail = stdout_tail.clone().filter(|s| !s.is_empty()); task.stderr_tail = stderr_tail.clone().filter(|s| !s.is_empty()); diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 8ca7570c0..d2226c78b 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -337,11 +337,11 @@ pub struct TaskFile { #[serde(default, skip_serializing_if = "Option::is_none")] pub timeout_secs: Option, pub status: TaskStatus, - pub created_at: i64, + pub created_at: DateTime, #[serde(default, skip_serializing_if = "Option::is_none")] - pub started_at: Option, + pub started_at: Option>, #[serde(default, skip_serializing_if = "Option::is_none")] - pub completed_at: Option, + pub completed_at: Option>, #[serde(default, skip_serializing_if = "Option::is_none")] pub exit_code: Option, /// Last `SUMMARY_BYTES` of stdout (see `hive-bash-mcp` runner). From f4a35786b18298bfd26de90960e2362b466aaa3f Mon Sep 17 00:00:00 2001 From: damocles Date: Sat, 1 Aug 2026 23:49:03 +0200 Subject: [PATCH 3/5] hive-agent: finish chrono-clock migration on remaining call sites --- hive-agent/src/db_migrate.rs | 2 +- hive-agent/src/events.rs | 10 +++++----- hive-agent/src/main.rs | 4 ++-- hive-agent/src/questions.rs | 10 ++++++---- hive-agent/src/reminder_timer.rs | 4 ++-- hive-agent/src/reminders.rs | 23 +++++++++++++---------- hive-agent/src/serve_common.rs | 1 - hive-agent/src/stats.rs | 22 ++++++++++++++++------ hive-agent/src/todo_server.rs | 14 +++++++------- hive-agent/src/todos.rs | 20 +++++++++++--------- 10 files changed, 63 insertions(+), 47 deletions(-) diff --git a/hive-agent/src/db_migrate.rs b/hive-agent/src/db_migrate.rs index 503a001b9..9feca95b5 100644 --- a/hive-agent/src/db_migrate.rs +++ b/hive-agent/src/db_migrate.rs @@ -172,7 +172,7 @@ mod tests { let pending = migrated.list_pending().unwrap(); assert_eq!(pending.len(), 1); assert_eq!(pending[0].id, orig_id, "id preserved across the copy"); - assert_eq!(pending[0].due_at, 1000); + assert_eq!(pending[0].due_at.timestamp(), 1000); } #[test] diff --git a/hive-agent/src/events.rs b/hive-agent/src/events.rs index 8cf1f6c3d..cf585f5f9 100644 --- a/hive-agent/src/events.rs +++ b/hive-agent/src/events.rs @@ -12,8 +12,8 @@ use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; +use chrono::Utc; use hive_claude::TokenUsage; -use hive_sh4re::wire_time::now_unix; use rusqlite::{Connection, params}; use serde::{Deserialize, Serialize}; use tokio::sync::broadcast; @@ -390,7 +390,7 @@ impl Bus { tx: Arc::new(tx), event_seq: Arc::new(AtomicU64::new(0)), store, - state: Arc::new(Mutex::new((TurnState::Idle, now_unix()))), + state: Arc::new(Mutex::new((TurnState::Idle, Utc::now().timestamp()))), model: Arc::new(Mutex::new(initial_model)), effort: Arc::new(Mutex::new(initial_effort)), last_ctx_usage: Arc::new(Mutex::new(None)), @@ -575,7 +575,7 @@ impl Bus { *self.last_ctx_usage.lock().unwrap() = Some(ctx); *self.last_cost_usage.lock().unwrap() = Some(cost); self.last_turn_ended_unix - .store(now_unix(), Ordering::Relaxed); + .store(Utc::now().timestamp(), Ordering::Relaxed); self.emit(LiveEvent::TokenUsageChanged { ctx, cost }); } @@ -778,7 +778,7 @@ impl Bus { if guard.0 == next { return; } - *guard = (next, now_unix()); + *guard = (next, Utc::now().timestamp()); since = guard.1; } self.emit(LiveEvent::TurnStateChanged { @@ -861,7 +861,7 @@ impl Bus { } let envelope = BusEvent { seq: self.next_seq(), - ts: now_unix(), + ts: Utc::now().timestamp(), event, }; // Lagged subscribers drop events — fine; the UI is a tail, not a log. diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 6d578de80..a22f27226 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -770,7 +770,7 @@ async fn handle_turn( unread, }); bus.set_state(TurnState::Thinking); - let started_at = serve_common::now_unix(); + let started_at = chrono::Utc::now().timestamp(); let started_instant = std::time::Instant::now(); let model_at_start = bus.model(); let prompt = serve_common::format_wake_prompt(msg_id, &from, &body, unread, redelivered); @@ -842,7 +842,7 @@ async fn handle_turn( let sid = stats.start_session(started_at, &model_at_start); bus.set_session_id(sid); } - let ended_at = serve_common::now_unix(); + let ended_at = chrono::Utc::now().timestamp(); let duration_ms = i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX); let (open_threads, open_reminders) = S::post_turn_counts(socket).await; let row = serve_common::build_row(serve_common::TurnRowArgs { diff --git a/hive-agent/src/questions.rs b/hive-agent/src/questions.rs index c2c5f8ca7..62a3fd956 100644 --- a/hive-agent/src/questions.rs +++ b/hive-agent/src/questions.rs @@ -32,7 +32,7 @@ use std::path::Path; use std::sync::Mutex; use anyhow::{Context, Result}; -use hive_sh4re::wire_time::now_unix; +use chrono::{DateTime, Utc}; use rusqlite::{Connection, params}; const SCHEMA: &str = r" @@ -78,7 +78,7 @@ pub struct QuestionMirror { pub role: Role, pub peer: String, pub question: String, - pub asked_at: i64, + pub asked_at: DateTime, } /// The harness-local questions mirror. Same sharing/locking shape as @@ -120,7 +120,7 @@ impl Questions { conn.execute( "INSERT OR REPLACE INTO questions (id, role, peer, question, asked_at) \ VALUES (?1, ?2, ?3, ?4, ?5)", - params![id, role.as_str(), peer, question, now_unix()], + params![id, role.as_str(), peer, question, Utc::now().timestamp()], )?; Ok(()) } @@ -170,12 +170,14 @@ impl Questions { tracing::warn!(%id, %role_str, "questions mirror: unknown role, skipping row"); continue; }; + let asked_at_secs: i64 = row.get(4)?; out.push(QuestionMirror { id, role, peer: row.get(2)?, question: row.get(3)?, - asked_at: row.get(4)?, + asked_at: chrono::DateTime::from_timestamp(asked_at_secs, 0) + .unwrap_or_else(Utc::now), }); } Ok(out) diff --git a/hive-agent/src/reminder_timer.rs b/hive-agent/src/reminder_timer.rs index 326f8ef50..8d8573437 100644 --- a/hive-agent/src/reminder_timer.rs +++ b/hive-agent/src/reminder_timer.rs @@ -74,7 +74,7 @@ pub async fn run( } fn tick(store: &Reminders, tx: &mpsc::UnboundedSender) { - let now = hive_sh4re::wire_time::now_unix(); + let now = chrono::Utc::now().timestamp(); let due = match store.due(now, REMINDER_BATCH_LIMIT) { Ok(rows) => rows, Err(e) => { @@ -269,7 +269,7 @@ mod tests { #[test] fn resolve_due_at_in_seconds_is_close_to_now_plus_n() { let due = resolve_due_at(&hive_sh4re::ReminderTiming::InSeconds { seconds: 60 }).unwrap(); - let now = hive_sh4re::wire_time::now_unix(); + let now = chrono::Utc::now().timestamp(); assert!((due - now - 60).abs() <= 2, "due={due} now={now}"); } diff --git a/hive-agent/src/reminders.rs b/hive-agent/src/reminders.rs index 4f4936f3b..f5837d919 100644 --- a/hive-agent/src/reminders.rs +++ b/hive-agent/src/reminders.rs @@ -17,8 +17,9 @@ use std::path::Path; use std::sync::Mutex; use anyhow::{Context, Result}; +use chrono::{DateTime, Utc}; use hive_sh4re::ReminderStats; -use hive_sh4re::wire_time::now_unix; +use hive_sh4re::wire_time; use rusqlite::{Connection, params}; const SCHEMA: &str = r" @@ -42,8 +43,8 @@ pub struct Reminder { pub id: i64, pub message: String, pub file_path: Option, - pub due_at: i64, - pub created_at: i64, + pub due_at: DateTime, + pub created_at: DateTime, } /// The harness-local reminder store. Cheap to share behind an `Arc`; the @@ -81,7 +82,7 @@ impl Reminders { /// Panics if the connection mutex is poisoned. pub fn store(&self, message: &str, file_path: Option<&str>, due_at: i64) -> Result { let conn = self.conn.lock().unwrap(); - let now = now_unix(); + let now = Utc::now().timestamp(); conn.execute( "INSERT INTO reminders (message, file_path, due_at, created_at, sent_at) \ VALUES (?1, ?2, ?3, ?4, NULL)", @@ -170,7 +171,7 @@ impl Reminders { let conn = self.conn.lock().unwrap(); conn.execute( "UPDATE reminders SET sent_at = ?1 WHERE id = ?2", - params![now_unix(), id], + params![Utc::now().timestamp(), id], )?; Ok(()) } @@ -210,7 +211,7 @@ impl Reminders { pub fn rollup(&self, since_secs: i64) -> Result { let conn = self.conn.lock().unwrap(); let cutoff = if since_secs > 0 { - now_unix().saturating_sub(since_secs) + Utc::now().timestamp().saturating_sub(since_secs) } else { i64::MIN }; @@ -258,12 +259,14 @@ impl Reminders { } fn row_to_reminder(row: &rusqlite::Row) -> rusqlite::Result { + let due_at_secs: i64 = row.get(3)?; + let created_at_secs: i64 = row.get(4)?; Ok(Reminder { id: row.get(0)?, message: row.get(1)?, file_path: row.get(2)?, - due_at: row.get(3)?, - created_at: row.get(4)?, + due_at: wire_time::from_secs(due_at_secs), + created_at: wire_time::from_secs(created_at_secs), }) } @@ -287,7 +290,7 @@ mod tests { let pending = s.list_pending().unwrap(); assert_eq!(pending.len(), 1); assert_eq!(pending[0].id, id); - assert_eq!(pending[0].due_at, 1000); + assert_eq!(pending[0].due_at.timestamp(), 1000); assert_eq!(s.count_pending().unwrap(), 1); } @@ -348,7 +351,7 @@ mod tests { ) .unwrap(); } - let cutoff = now_unix() - 10; + let cutoff = Utc::now().timestamp() - 10; let n = s.prune_delivered_older_than(cutoff).unwrap(); assert_eq!(n, 1, "only the backdated row is older than cutoff"); let remaining_ids: Vec = { diff --git a/hive-agent/src/serve_common.rs b/hive-agent/src/serve_common.rs index c3fdcf6fe..778aee51c 100644 --- a/hive-agent/src/serve_common.rs +++ b/hive-agent/src/serve_common.rs @@ -7,7 +7,6 @@ use crate::events::Bus; use crate::turn::{TurnError, TurnOutcome}; use crate::turn_stats::TurnStatRow; -pub use hive_sh4re::wire_time::now_unix; /// Assemble the per-turn wake prompt string. The role/tools/etc. live in the /// system prompt; this is just the wake signal body. `id` is the broker row diff --git a/hive-agent/src/stats.rs b/hive-agent/src/stats.rs index 91563fdbd..ff018018d 100644 --- a/hive-agent/src/stats.rs +++ b/hive-agent/src/stats.rs @@ -11,11 +11,11 @@ use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; use anyhow::{Context, Result}; +use chrono::Utc; use rusqlite::{Connection, OpenFlags}; use serde::Serialize; use hive_sh4re::ReminderStats; -use hive_sh4re::wire_time::now_unix; /// Window param accepted by `/api/stats?window=`. Each maps to a /// total span + the bucket width used to roll up trend series. @@ -225,7 +225,7 @@ fn default_path() -> PathBuf { } fn empty_snapshot(window: Window) -> Snapshot { - let now = now_unix(); + let now = Utc::now().timestamp(); let from = now - window.span_secs(); let buckets = fill_buckets(from, now, window.bucket_secs(), &HashMap::new()); Snapshot { @@ -305,7 +305,7 @@ fn snapshot(path: &Path, window: Window) -> Result { // and blanks the whole stats page. Wait out the brief write instead. conn.busy_timeout(std::time::Duration::from_millis(500)) .with_context(|| format!("set busy_timeout on {}", path.display()))?; - let now = now_unix(); + let now = Utc::now().timestamp(); // Fixed windows look back a constant span; `all` starts at the earliest // recorded turn and sizes its buckets adaptively from that span. let (from, bucket_secs) = match window { @@ -677,7 +677,7 @@ mod tests { fn snapshot_aggregates_rows() { let db = tmp_db(); let _ = std::fs::remove_file(&db); - let now = now_unix(); + let now = Utc::now().timestamp(); seed_db( &db, &[ @@ -766,7 +766,17 @@ mod tests { fn bash_breakdown_empty_without_table() { let db = tmp_db(); let _ = std::fs::remove_file(&db); - seed_db(&db, &[(now_unix() - 100, 1000, "opus", "recv", "ok", "{}")]); + seed_db( + &db, + &[( + Utc::now().timestamp() - 100, + 1000, + "opus", + "recv", + "ok", + "{}", + )], + ); let s = snapshot(&db, Window::Day).unwrap(); assert!(s.bash_breakdown.is_empty()); } @@ -778,7 +788,7 @@ mod tests { let db = tmp_db(); let _ = std::fs::remove_file(&db); seed_db(&db, &[]); - let now = now_unix(); + let now = Utc::now().timestamp(); let conn = Connection::open(&db).unwrap(); conn.execute_batch("CREATE TABLE bash_commands (ts INTEGER NOT NULL, head TEXT NOT NULL);") .unwrap(); diff --git a/hive-agent/src/todo_server.rs b/hive-agent/src/todo_server.rs index f9a811674..c307e0270 100644 --- a/hive-agent/src/todo_server.rs +++ b/hive-agent/src/todo_server.rs @@ -479,8 +479,8 @@ fn no_questions_store() -> Response { /// `target` are derived from `role` — this agent's own label fills whichever /// side `role` says is us, `peer` fills the other. fn question_to_loose_end(q: QuestionMirror) -> LooseEnd { - let now = hive_sh4re::wire_time::now_unix(); - let age = u64::try_from(now.saturating_sub(q.asked_at)).unwrap_or(0); + let now = chrono::Utc::now().timestamp(); + let age = u64::try_from(now.saturating_sub(q.asked_at.timestamp())).unwrap_or(0); let me = crate::identity::label(); let (asker, target) = match q.role { Role::Asked => (me, Some(q.peer)), @@ -500,13 +500,13 @@ fn question_to_loose_end(q: QuestionMirror) -> LooseEnd { /// "age" is how long the reminder has been *scheduled*, not how soon /// it's due). fn reminder_to_loose_end(r: Reminder) -> LooseEnd { - let now = hive_sh4re::wire_time::now_unix(); - let age = u64::try_from(now.saturating_sub(r.created_at)).unwrap_or(0); + let now = chrono::Utc::now().timestamp(); + let age = u64::try_from(now.saturating_sub(r.created_at.timestamp())).unwrap_or(0); LooseEnd::Reminder { id: r.id, owner: crate::identity::label(), message: r.message, - due_at: hive_sh4re::wire_time::from_secs(r.due_at), + due_at: r.due_at, age_seconds: age, } } @@ -521,8 +521,8 @@ fn err(e: &anyhow::Error) -> Response { /// Map a stored [`Todo`] to a [`LooseEnd::Todo`], deriving `age_seconds` /// from `updated_at` (saturating so a backwards clock step reads 0). fn to_loose_end(t: Todo) -> LooseEnd { - let now = hive_sh4re::wire_time::now_unix(); - let age = u64::try_from(now.saturating_sub(t.updated_at)).unwrap_or(0); + let now = chrono::Utc::now().timestamp(); + let age = u64::try_from(now.saturating_sub(t.updated_at.timestamp())).unwrap_or(0); LooseEnd::Todo { id: t.id, subsystem: t.subsystem, diff --git a/hive-agent/src/todos.rs b/hive-agent/src/todos.rs index 63deb11da..4fc084657 100644 --- a/hive-agent/src/todos.rs +++ b/hive-agent/src/todos.rs @@ -32,7 +32,8 @@ use std::path::Path; use std::sync::Mutex; use anyhow::{Context, Result}; -use hive_sh4re::wire_time::now_unix; +use chrono::{DateTime, Utc}; +use hive_sh4re::wire_time; use rusqlite::{Connection, params}; /// SQL bootstrap. `CREATE TABLE IF NOT EXISTS` so first-boot agents and @@ -66,8 +67,8 @@ const MIGRATIONS: &[&str] = &[ "ALTER TABLE todos ADD COLUMN acked_at INTEGER", ]; -/// One dynamic, subsystem-pushed todo. Timestamps are unix seconds; the -/// consumer derives `age_seconds` from `updated_at`. +/// One dynamic, subsystem-pushed todo. The consumer derives `age_seconds` +/// from `updated_at`. #[derive(Debug, Clone)] pub struct Todo { pub id: i64, @@ -80,7 +81,7 @@ pub struct Todo { pub summary: String, /// Optional free-text provenance (e.g. the room name / task label). pub source: Option, - pub updated_at: i64, + pub updated_at: DateTime, } /// The harness-local todo store. Cheap to share behind an `Arc`; the inner @@ -144,7 +145,7 @@ impl Todos { source: Option<&str>, ) -> Result<(i64, bool)> { let conn = self.conn.lock().unwrap(); - let now = now_unix(); + let now = Utc::now().timestamp(); let existing: Option<(i64, String, Option)> = if key.is_some() { conn.query_row( "SELECT id, summary, source FROM todos \ @@ -239,7 +240,7 @@ impl Todos { let conn = self.conn.lock().unwrap(); let n = conn.execute( "UPDATE todos SET acked = 1, acked_at = ?1 WHERE id = ?2 AND acked = 0", - params![now_unix(), id], + params![Utc::now().timestamp(), id], )?; Ok(n) } @@ -266,13 +267,14 @@ impl Todos { )?; let rows = stmt .query_map(params![subsystem], |row| { + let updated_at_secs: i64 = row.get(5)?; Ok(Todo { id: row.get(0)?, subsystem: row.get(1)?, subsystem_key: row.get(2)?, summary: row.get(3)?, source: row.get(4)?, - updated_at: row.get(5)?, + updated_at: wire_time::from_secs(updated_at_secs), }) })? .collect::>>()?; @@ -491,12 +493,12 @@ mod tests { let conn = s.conn.lock().unwrap(); conn.execute( "UPDATE todos SET acked_at = ?1 WHERE id = ?2", - params![now_unix() - 1000, old_id], + params![Utc::now().timestamp() - 1000, old_id], ) .unwrap(); } - let removed = s.reap_acked(now_unix() - 500).unwrap(); + let removed = s.reap_acked(Utc::now().timestamp() - 500).unwrap(); assert_eq!(removed, 1, "only the backdated row is past the cutoff"); let conn = Connection::open(dir.path().join("todos.sqlite")).unwrap(); From 9293c5d580f9a244e52d7bc927b16c9c8c504fc3 Mon Sep 17 00:00:00 2001 From: damocles Date: Sat, 1 Aug 2026 23:55:00 +0200 Subject: [PATCH 4/5] hive-c0re: finish chrono-clock migration --- hive-c0re/src/dashboard/state_snapshot.rs | 2 +- hive-c0re/src/job_queue/mod.rs | 11 ++-- hive-c0re/src/job_queue/model.rs | 3 +- hive-c0re/src/loose_ends.rs | 6 +- hive-c0re/src/socket_server/schedules.rs | 12 ++-- hive-c0re/src/stats/hive_stats.rs | 6 +- hive-c0re/src/stores/approvals.rs | 14 ++--- hive-c0re/src/stores/audit_log.rs | 7 +-- hive-c0re/src/stores/broker.rs | 16 +++--- hive-c0re/src/stores/build_logs.rs | 10 ++-- hive-c0re/src/stores/operator_questions.rs | 7 +-- hive-c0re/src/stores/power.rs | 4 +- hive-c0re/src/stores/scheduled_prompts.rs | 56 +++++++++++-------- .../src/workers/scheduled_prompts_worker.rs | 6 +- 14 files changed, 82 insertions(+), 78 deletions(-) diff --git a/hive-c0re/src/dashboard/state_snapshot.rs b/hive-c0re/src/dashboard/state_snapshot.rs index 04c439357..27309cc0c 100644 --- a/hive-c0re/src/dashboard/state_snapshot.rs +++ b/hive-c0re/src/dashboard/state_snapshot.rs @@ -538,7 +538,7 @@ fn build_transient_views( // Clamped at 0: `since` is wall-clock now (the node's own // `started_at`), so a backwards clock adjustment could otherwise // render a negative age. - secs: (hive_sh4re::wire_time::from_secs(hive_sh4re::wire_time::now_unix()) - st.since) + secs: (chrono::Utc::now() - st.since) .num_seconds() .max(0) .cast_unsigned(), diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index ada7faad6..bbe909250 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -44,7 +44,6 @@ use hive_host_sock::jobs::NodeView; use hive_jobq::resources::ResourceTable; use hive_jobq::scheduler::{Outcome, Scheduler}; use hive_jobq::{Dep, Graph, NodeId}; -use hive_sh4re::wire_time::now_unix; use tokio::sync::Notify; pub use hive_jobq::TerminalState; @@ -107,7 +106,7 @@ struct NodeRuntime { struct DagMeta { source: Source, reason: String, - created_at: i64, + created_at: DateTime, } /// The mutable queue state behind the mutex: the crate scheduler plus the @@ -230,7 +229,7 @@ impl JobQueue { NodeKind::Dag { source: spec.source, reason: spec.reason, - created_at: now_unix(), + created_at: Utc::now(), }, Vec::new(), None, @@ -469,9 +468,7 @@ impl JobQueue { // `started_at` is set when a node enters `Running`, and this // only sees `Running` nodes — the fallback is unreachable in // practice, and "just now" is the honest answer if it isn't. - since: n - .started_at - .unwrap_or_else(|| hive_sh4re::wire_time::from_secs(now_unix())), + since: n.started_at.unwrap_or_else(Utc::now), }) }) .collect() @@ -628,7 +625,7 @@ impl QueueInner { id: container.get(), source: meta.source, reason: meta.reason.clone(), - created_at: hive_sh4re::wire_time::from_secs(meta.created_at), + created_at: meta.created_at, started_at: started.into_iter().min(), finished_at: is_terminal.then(|| finished.into_iter().max()).flatten(), nodes, diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 96fba3c38..77ba0757e 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -12,6 +12,7 @@ //! DAG can span agents). See `docs/coordinator.md::Job queue` for the //! full design. +use chrono::{DateTime, Utc}; pub use hive_host_sock::jobs::{DagView, NodeId, PermPayload, Source, State}; use serde::Serialize; @@ -283,7 +284,7 @@ pub enum NodeKind { Dag { source: Source, reason: String, - created_at: i64, + created_at: DateTime, }, } diff --git a/hive-c0re/src/loose_ends.rs b/hive-c0re/src/loose_ends.rs index 621419fb8..b20ef567a 100644 --- a/hive-c0re/src/loose_ends.rs +++ b/hive-c0re/src/loose_ends.rs @@ -20,10 +20,10 @@ //! (de)serialisation, not the read. use anyhow::Result; +use chrono::Utc; use hive_sh4re::LooseEnd; use crate::coordinator::Coordinator; -use hive_sh4re::wire_time::now_unix; /// Open threads pending against `agent`: /// - undelivered inbox messages this agent still owes itself a `recv` @@ -42,7 +42,7 @@ use hive_sh4re::wire_time::now_unix; /// Propagates errors from `count_pending` and the pending-approval /// sqlite query. pub fn for_agent(coord: &Coordinator, agent: &str) -> Result> { - let now = now_unix(); + let now = Utc::now().timestamp(); let mut out = Vec::new(); // Undelivered inbox messages this agent still owes itself a `recv` // for. Surfaced first (most actionable) and only when non-zero so a @@ -81,7 +81,7 @@ pub fn for_agent(coord: &Coordinator, agent: &str) -> Result> { /// only; sub-agents can't see each other's threads via the agent surface /// (`for_agent` filters by name). pub fn hive_wide(coord: &Coordinator) -> Result> { - let now = now_unix(); + let now = Utc::now().timestamp(); let mut out = Vec::new(); for a in coord.approvals.pending()? { out.push(LooseEnd::Approval { diff --git a/hive-c0re/src/socket_server/schedules.rs b/hive-c0re/src/socket_server/schedules.rs index bebe855d3..aa43e99ca 100644 --- a/hive-c0re/src/socket_server/schedules.rs +++ b/hive-c0re/src/socket_server/schedules.rs @@ -321,8 +321,8 @@ fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSc owner: s.owner, body: s.body, interval_seconds: s.interval_seconds, - next_fire_at_unix: hive_sh4re::wire_time::from_secs(s.next_fire_at_unix), - created_at_unix: hive_sh4re::wire_time::from_secs(s.created_at_unix), + next_fire_at_unix: s.next_fire_at_unix, + created_at_unix: s.created_at_unix, source: match s.source { crate::scheduled_prompts::ScheduleSource::Operator => { hive_sh4re::WireScheduleSource::Operator @@ -331,16 +331,16 @@ fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSc hive_sh4re::WireScheduleSource::Approval { id } } }, - cancelled_at_unix: s.cancelled_at_unix.map(hive_sh4re::wire_time::from_secs), - paused_at_unix: s.paused_at_unix.map(hive_sh4re::wire_time::from_secs), + cancelled_at_unix: s.cancelled_at_unix, + paused_at_unix: s.paused_at_unix, description: s.description, targets: s .targets .into_iter() .map(|t| hive_sh4re::WireScheduleTarget { target: t.target, - cancelled_at_unix: t.cancelled_at_unix.map(hive_sh4re::wire_time::from_secs), - last_fired_at_unix: t.last_fired_at_unix.map(hive_sh4re::wire_time::from_secs), + cancelled_at_unix: t.cancelled_at_unix, + last_fired_at_unix: t.last_fired_at_unix, last_result: t.last_result, }) .collect(), diff --git a/hive-c0re/src/stats/hive_stats.rs b/hive-c0re/src/stats/hive_stats.rs index 7c4fea232..94b1e8c98 100644 --- a/hive-c0re/src/stats/hive_stats.rs +++ b/hive-c0re/src/stats/hive_stats.rs @@ -25,7 +25,7 @@ use serde::{Deserialize, Serialize}; use utoipa::ToSchema; use crate::coordinator::Coordinator; -use hive_sh4re::wire_time::now_unix; +use chrono::Utc; /// Window accepted by `/api/stats-hive?window=`. Maps to a lookback /// span; the hive view is a flat rollup (no per-bucket trend — the @@ -349,7 +349,7 @@ fn read_bash_heads(conn: &Connection, from: i64) -> HashMap { /// per-agent db is skipped (logged), never fatal. #[must_use] pub fn hive_snapshot(window: Window, prices: &PriceTable) -> HiveStats { - let now = now_unix(); + let now = Utc::now().timestamp(); // Fixed windows look back a constant span; `all` aggregates every // recorded turn (`from == 0`). The hive rollup isn't time-bucketed, so // unlike the per-agent snapshot it needs no adaptive bucket sizing. @@ -466,7 +466,7 @@ mod tests { let conn = Connection::open_in_memory().unwrap(); conn.execute_batch("CREATE TABLE bash_commands (ts INTEGER NOT NULL, head TEXT NOT NULL);") .unwrap(); - let now = now_unix(); + let now = Utc::now().timestamp(); for (ts, head) in [ (now - 100, "cargo"), (now - 200, "cargo"), diff --git a/hive-c0re/src/stores/approvals.rs b/hive-c0re/src/stores/approvals.rs index 313f04e44..651af2654 100644 --- a/hive-c0re/src/stores/approvals.rs +++ b/hive-c0re/src/stores/approvals.rs @@ -7,7 +7,7 @@ use std::path::Path; use std::sync::Mutex; use anyhow::{Context, Result, bail}; -use hive_sh4re::wire_time::now_unix; +use chrono::Utc; use hive_sh4re::{Approval, ApprovalKind, ApprovalStatus}; use rusqlite::{Connection, OptionalExtension, params}; @@ -95,7 +95,7 @@ impl Approvals { agent, kind.as_str(), commit_ref, - now_unix(), + Utc::now().timestamp(), description, submitter, fetched_sha, @@ -206,7 +206,7 @@ impl Approvals { if row.status != "pending" { bail!("approval {id} is {}, not pending", row.status); } - let resolved_at = now_unix(); + let resolved_at = Utc::now().timestamp(); conn.execute( "UPDATE approvals SET status = 'approved', resolved_at = ?1 WHERE id = ?2", params![resolved_at, id], @@ -230,7 +230,7 @@ impl Approvals { let affected = conn.execute( "UPDATE approvals SET status = 'denied', resolved_at = ?1, note = ?2 WHERE id = ?3 AND status = 'pending'", - params![now_unix(), note, id], + params![Utc::now().timestamp(), note, id], )?; if affected == 0 { bail!("approval {id} not pending"); @@ -242,7 +242,7 @@ impl Approvals { let conn = self.conn.lock().unwrap(); conn.execute( "UPDATE approvals SET status = 'failed', resolved_at = ?1, note = ?2 WHERE id = ?3", - params![now_unix(), note, id], + params![Utc::now().timestamp(), note, id], )?; Ok(()) } @@ -267,7 +267,7 @@ impl Approvals { if row.status != "pending" { bail!("approval {id} is {}, not pending", row.status); } - let resolved_at = now_unix(); + 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", @@ -295,7 +295,7 @@ impl Approvals { let n = conn.execute( "UPDATE approvals SET status = 'failed', resolved_at = ?1, note = ?2 WHERE agent = ?3 AND status = 'pending'", - params![now_unix(), note, agent], + params![Utc::now().timestamp(), note, agent], )?; Ok(n) } diff --git a/hive-c0re/src/stores/audit_log.rs b/hive-c0re/src/stores/audit_log.rs index 8e8bc1608..eb1cd377d 100644 --- a/hive-c0re/src/stores/audit_log.rs +++ b/hive-c0re/src/stores/audit_log.rs @@ -25,7 +25,6 @@ use std::sync::{Arc, Mutex, OnceLock}; use anyhow::{Context, Result}; use chrono::{DateTime, Utc}; -use hive_sh4re::wire_time::now_unix; use rusqlite::{Connection, params}; use serde::Serialize; @@ -141,7 +140,7 @@ impl AuditLog { outcome: AuditOutcome, detail: Option<&str>, ) -> Option { - let now = now_unix(); + let now = Utc::now().timestamp(); let conn = self.conn.lock().unwrap(); match conn.execute( "INSERT INTO audit_log (ts_unix, agent, action, target, outcome, detail) @@ -208,7 +207,7 @@ impl AuditLog { /// # Errors /// Returns an error if the `DELETE` query fails. pub fn vacuum(&self) -> Result { - let cutoff = now_unix() - KEEP_SECS; + let cutoff = Utc::now().timestamp() - KEEP_SECS; let conn = self.conn.lock().unwrap(); let removed = conn.execute("DELETE FROM audit_log WHERE ts_unix < ?1", params![cutoff])?; Ok(u64::try_from(removed).unwrap_or(0)) @@ -311,7 +310,7 @@ mod tests { let conn = db.conn.lock().unwrap(); conn.execute( "UPDATE audit_log SET ts_unix = ?1", - params![now_unix() - KEEP_SECS - 60], + params![Utc::now().timestamp() - KEEP_SECS - 60], ) .unwrap(); } diff --git a/hive-c0re/src/stores/broker.rs b/hive-c0re/src/stores/broker.rs index 729bbeed6..76e7dd516 100644 --- a/hive-c0re/src/stores/broker.rs +++ b/hive-c0re/src/stores/broker.rs @@ -6,8 +6,8 @@ use std::path::Path; use std::sync::Mutex; use anyhow::{Context, Result}; +use chrono::Utc; -use hive_sh4re::wire_time::now_unix; use hive_sh4re::{InboxRow, Message}; use crate::db::Migration; @@ -211,7 +211,7 @@ impl Broker { pub fn send(&self, message: &Message) -> Result<()> { let conn = self.conn.lock().unwrap(); - let now = now_unix(); + let now = Utc::now().timestamp(); // Operator messages get elevated priority so they surface before // queued wakes (bash completions, forge events, etc.) when the // harness pops the next turn driver. All other senders stay at 0. @@ -391,7 +391,7 @@ impl Broker { const PREFIX: &str = "your parent changed from "; const SEPARATOR: &str = " to "; let conn = self.conn.lock().unwrap(); - let now = now_unix(); + let now = Utc::now().timestamp(); let existing: Option<(i64, String)> = conn .query_row( "SELECT id, body FROM messages @@ -513,7 +513,7 @@ impl Broker { /// `requeue_inflight`, the latter because they're still in flight /// from the broker's POV. Returns the number of rows removed. pub fn vacuum_delivered(&self, older_than_secs: i64) -> Result { - let cutoff = now_unix() - older_than_secs; + let cutoff = Utc::now().timestamp() - older_than_secs; let conn = self.conn.lock().unwrap(); let n = conn.execute( "DELETE FROM messages @@ -578,7 +578,7 @@ impl Broker { } // Stamp all popped rows in a single UPDATE — under the broker // mutex, well within sqlite's 999-param default. - let now = now_unix(); + let now = Utc::now().timestamp(); let ids: Vec = rows.iter().map(|(id, _, _, _, _)| *id).collect(); let placeholders = std::iter::repeat_n("?", ids.len()) .collect::>() @@ -642,7 +642,7 @@ impl Broker { if ids.is_empty() { return Ok(0); } - let now = now_unix(); + let now = Utc::now().timestamp(); let conn = self.conn.lock().unwrap(); // Bind every id explicitly. Caps in the hundreds in the worst // case (a single very chatty turn); well under sqlite's 999 @@ -688,7 +688,7 @@ impl Broker { let n = conn.execute( "UPDATE messages SET acked_at = ?1 WHERE recipient = ?2 AND id <= ?3 AND acked_at IS NULL", - params![now_unix(), recipient, up_to], + params![Utc::now().timestamp(), recipient, up_to], )?; Ok(u64::try_from(n).unwrap_or(0)) } @@ -762,7 +762,7 @@ impl Broker { pub fn mark_all_read(&self, recipient: &str) -> Result { let mut inflight = self.inflight.lock().unwrap(); let conn = self.conn.lock().unwrap(); - let now = now_unix(); + let now = Utc::now().timestamp(); // Two-axis update in one statement: set acked_at on every // row for the recipient that doesn't have it yet, AND backfill // delivered_at if it was NULL so the row is fully consumed diff --git a/hive-c0re/src/stores/build_logs.rs b/hive-c0re/src/stores/build_logs.rs index 678a479ac..24b302bdc 100644 --- a/hive-c0re/src/stores/build_logs.rs +++ b/hive-c0re/src/stores/build_logs.rs @@ -7,7 +7,7 @@ use std::path::Path; use std::sync::{Arc, Mutex, OnceLock}; use anyhow::{Context, Result}; -use hive_sh4re::wire_time::now_unix; +use chrono::Utc; use rusqlite::{Connection, OptionalExtension, params}; use serde::Serialize; use tokio::sync::broadcast; @@ -170,7 +170,7 @@ impl BuildLogs { /// — the caller threads it through `append_stdout` / `append_stderr` /// while the child runs and into `finish` once it exits. pub fn start(&self, agent: &str, kind: &str, cmdline: &str) -> Result { - let now = now_unix(); + let now = Utc::now().timestamp(); let conn = self.conn.lock().unwrap(); conn.execute( "INSERT INTO build_logs (agent, kind, cmdline, started_at) VALUES (?1, ?2, ?3, ?4)", @@ -220,7 +220,7 @@ impl BuildLogs { /// Finalize a build attempt. Sets `finished_at` to now and /// `status` to the terminal state. Best-effort. pub fn finish(&self, id: i64, status: BuildStatus) { - let now = now_unix(); + let now = Utc::now().timestamp(); let conn = self.conn.lock().unwrap(); if let Err(e) = conn.execute( "UPDATE build_logs SET finished_at = ?1, status = ?2 WHERE id = ?3", @@ -375,7 +375,7 @@ impl BuildLogs { /// a long-running build shouldn't disappear from its own log /// viewer mid-stream. pub fn vacuum(&self) -> Result { - let now = now_unix(); + let now = Utc::now().timestamp(); let conn = self.conn.lock().unwrap(); let fail_cutoff = now - KEEP_FAIL_SECS; let ok_cutoff = now - KEEP_OK_SECS; @@ -526,7 +526,7 @@ mod tests { // stays within KEEP_FAIL_SECS so it survives; old_fail goes // beyond; old_ok goes past KEEP_OK_SECS but inside // KEEP_FAIL_SECS — proves the per-status rule. - let now = now_unix(); + let now = Utc::now().timestamp(); { let conn = db.conn.lock().unwrap(); conn.execute( diff --git a/hive-c0re/src/stores/operator_questions.rs b/hive-c0re/src/stores/operator_questions.rs index 64f7b3509..089eef903 100644 --- a/hive-c0re/src/stores/operator_questions.rs +++ b/hive-c0re/src/stores/operator_questions.rs @@ -15,7 +15,6 @@ use std::sync::Mutex; use anyhow::{Context, Result, bail}; use chrono::{DateTime, Utc}; -use hive_sh4re::wire_time::now_unix; use rusqlite::{Connection, OptionalExtension, params}; use serde::Serialize; @@ -119,7 +118,7 @@ impl OperatorQuestions { i64::from(multi), deadline_at, target, - now_unix(), + Utc::now().timestamp(), ], )?; Ok(conn.last_insert_rowid()) @@ -177,7 +176,7 @@ impl OperatorQuestions { } conn.execute( "UPDATE operator_questions SET answer = ?1, answered_at = ?2 WHERE id = ?3", - params![answer, now_unix(), id], + params![answer, Utc::now().timestamp(), id], )?; Ok((question, asker, target)) } @@ -225,7 +224,7 @@ impl OperatorQuestions { let sentinel = format!("[cancelled by {canceller}]"); conn.execute( "UPDATE operator_questions SET answer = ?1, answered_at = ?2 WHERE id = ?3", - params![sentinel, now_unix(), id], + params![sentinel, Utc::now().timestamp(), id], )?; Ok((question, asker, target)) } diff --git a/hive-c0re/src/stores/power.rs b/hive-c0re/src/stores/power.rs index a5900adbb..6aa6e3844 100644 --- a/hive-c0re/src/stores/power.rs +++ b/hive-c0re/src/stores/power.rs @@ -18,7 +18,7 @@ use std::path::Path; use std::sync::Mutex; use anyhow::{Context, Result}; -use hive_sh4re::wire_time::now_unix; +use chrono::Utc; use rusqlite::{Connection, OptionalExtension, params}; const SCHEMA: &str = " @@ -129,7 +129,7 @@ impl PowerStore { conn.execute( "INSERT INTO agent_power (agent, wanted, updated_at) VALUES (?1, ?2, ?3) ON CONFLICT(agent) DO UPDATE SET wanted = ?2, updated_at = ?3", - params![agent, wanted.as_str(), now_unix()], + params![agent, wanted.as_str(), Utc::now().timestamp()], ) .context("upsert agent_power")?; Ok(()) diff --git a/hive-c0re/src/stores/scheduled_prompts.rs b/hive-c0re/src/stores/scheduled_prompts.rs index f0877a5c1..75bff2a30 100644 --- a/hive-c0re/src/stores/scheduled_prompts.rs +++ b/hive-c0re/src/stores/scheduled_prompts.rs @@ -16,7 +16,7 @@ use std::path::Path; use std::sync::Mutex; use anyhow::{Context, Result, bail}; -use hive_sh4re::wire_time::now_unix; +use chrono::{DateTime, Utc}; use rusqlite::{Connection, OptionalExtension, params}; use serde::{Deserialize, Serialize}; @@ -78,25 +78,25 @@ pub struct Schedule { /// `None` = one-shot, deleted after first fire. /// `Some(n)` = recurring every `n` seconds. pub interval_seconds: Option, - pub next_fire_at_unix: i64, - pub created_at_unix: i64, + pub next_fire_at_unix: DateTime, + pub created_at_unix: DateTime, pub source: ScheduleSource, /// Set when the *entire* schedule was cancelled (all targets /// flipped, or operator cancel-all). Worker reaps these on the /// next pass. - pub cancelled_at_unix: Option, + pub cancelled_at_unix: Option>, pub description: Option, /// Set while the schedule is paused. Worker skips rows where /// `paused_at_unix IS NOT NULL`. Cleared by `resume()`. - pub paused_at_unix: Option, + pub paused_at_unix: Option>, pub targets: Vec, } #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ScheduleTarget { pub target: String, - pub cancelled_at_unix: Option, - pub last_fired_at_unix: Option, + pub cancelled_at_unix: Option>, + pub last_fired_at_unix: Option>, pub last_result: Option, } @@ -237,7 +237,7 @@ impl ScheduledPrompts { &new.body, new.interval_seconds.map(i64::try_from).and_then(Result::ok), new.first_fire_at_unix, - now_unix(), + Utc::now().timestamp(), new.source.to_db_string(), &new.description, ], @@ -472,7 +472,7 @@ impl ScheduledPrompts { // orphaned tombstone. Plus removing-then-adding is the // natural way to "restart history" on one target without // a separate flow. - let now = now_unix(); + let now = Utc::now().timestamp(); if let Some(remove) = patch.targets_remove.as_deref() { for target in remove { tx.execute( @@ -536,7 +536,7 @@ impl ScheduledPrompts { /// already-cancelled row. pub fn cancel_all(&self, id: i64) -> Result<()> { let mut conn = self.conn.lock().unwrap(); - let now = now_unix(); + let now = Utc::now().timestamp(); let tx = conn.transaction()?; tx.execute( "UPDATE scheduled_prompts @@ -560,7 +560,7 @@ impl ScheduledPrompts { /// are silently skipped. pub fn cancel_targets(&self, id: i64, targets: &[String]) -> Result<()> { let mut conn = self.conn.lock().unwrap(); - let now = now_unix(); + let now = Utc::now().timestamp(); let tx = conn.transaction()?; for target in targets { tx.execute( @@ -599,7 +599,7 @@ impl ScheduledPrompts { /// cancelled or does not exist (handlers downcast to emit 404). pub fn pause(&self, id: i64) -> Result<()> { let conn = self.conn.lock().unwrap(); - let now = now_unix(); + let now = Utc::now().timestamp(); let n = conn.execute( "UPDATE scheduled_prompts SET paused_at_unix = COALESCE(paused_at_unix, ?1) @@ -653,12 +653,16 @@ fn row_to_schedule_header(row: &rusqlite::Row) -> rusqlite::Result { owner: row.get(1)?, body: row.get(2)?, interval_seconds: interval.and_then(|i| u64::try_from(i).ok()), - next_fire_at_unix: row.get(4)?, - created_at_unix: row.get(5)?, + next_fire_at_unix: hive_sh4re::wire_time::from_secs(row.get(4)?), + created_at_unix: hive_sh4re::wire_time::from_secs(row.get(5)?), source: ScheduleSource::from_db_string(&source_str), - cancelled_at_unix: row.get(7)?, + cancelled_at_unix: row + .get::<_, Option>(7)? + .map(hive_sh4re::wire_time::from_secs), description: row.get(8)?, - paused_at_unix: row.get(9)?, + paused_at_unix: row + .get::<_, Option>(9)? + .map(hive_sh4re::wire_time::from_secs), targets: Vec::new(), }) } @@ -673,8 +677,12 @@ fn load_targets(conn: &Connection, schedule_id: i64) -> Result>(1)? + .map(hive_sh4re::wire_time::from_secs), + last_fired_at_unix: row + .get::<_, Option>(2)? + .map(hive_sh4re::wire_time::from_secs), last_result: row.get(3)?, }) })?; @@ -771,7 +779,7 @@ mod tests { let skipped = db.rearm(id, 400).expect("rearm"); assert_eq!(skipped, 5); let s = db.get(id).expect("get").expect("present"); - assert_eq!(s.next_fire_at_unix, 460); + assert_eq!(s.next_fire_at_unix.timestamp(), 460); } #[test] @@ -793,7 +801,7 @@ mod tests { // the manual fire-now "reset timer" path uses now + interval. db.set_next_fire(id, 1_234).expect("set_next_fire"); let s = db.get(id).expect("get").expect("present"); - assert_eq!(s.next_fire_at_unix, 1_234); + assert_eq!(s.next_fire_at_unix.timestamp(), 1_234); } #[test] @@ -814,7 +822,7 @@ mod tests { let skipped = db.rearm(id, 100).expect("rearm"); assert_eq!(skipped, 0); let s = db.get(id).expect("get").expect("present"); - assert_eq!(s.next_fire_at_unix, 160); + assert_eq!(s.next_fire_at_unix.timestamp(), 160); } #[test] @@ -825,7 +833,7 @@ mod tests { assert_eq!(skipped, 0); // next_fire_at unchanged — one-shots are reaped via delete(). let s = db.get(id).expect("get").expect("present"); - assert_eq!(s.next_fire_at_unix, 100); + assert_eq!(s.next_fire_at_unix.timestamp(), 100); } #[test] @@ -878,7 +886,7 @@ mod tests { // Cancelled targets do NOT get last-result writes. assert!(alice.last_fired_at_unix.is_none()); assert!(alice.last_result.is_none()); - assert_eq!(bob.last_fired_at_unix, Some(200)); + assert_eq!(bob.last_fired_at_unix.map(|dt| dt.timestamp()), Some(200)); assert_eq!(bob.last_result.as_deref(), Some("ok")); } @@ -909,7 +917,7 @@ mod tests { assert_eq!(s.body, "new body"); assert_eq!(s.description.as_deref(), Some("old desc")); assert_eq!(s.interval_seconds, Some(60)); - assert_eq!(s.next_fire_at_unix, 100); + assert_eq!(s.next_fire_at_unix.timestamp(), 100); } #[test] diff --git a/hive-c0re/src/workers/scheduled_prompts_worker.rs b/hive-c0re/src/workers/scheduled_prompts_worker.rs index 4adb0743a..b7dac0816 100644 --- a/hive-c0re/src/workers/scheduled_prompts_worker.rs +++ b/hive-c0re/src/workers/scheduled_prompts_worker.rs @@ -6,11 +6,11 @@ use std::sync::Arc; use std::time::Duration; +use chrono::Utc; use hive_sh4re::Message; use crate::coordinator::Coordinator; use crate::scheduled_prompts::Schedule; -use hive_sh4re::wire_time::now_unix; /// Per-tick cap. Each schedule fires once per tick at most; /// 100/tick × 5s tick = sustained throughput cap of ~20/sec, @@ -47,7 +47,7 @@ pub fn spawn(coord: Arc) { } fn tick(coord: &Arc) { - let now = now_unix(); + let now = Utc::now().timestamp(); let due = match coord.scheduled_prompts.due(now, SCHEDULE_BATCH_LIMIT) { Ok(rows) => rows, Err(e) => { @@ -285,7 +285,7 @@ pub async fn fire_now( schedule_id: i64, reset_timer: bool, ) -> anyhow::Result { - let now = now_unix(); + let now = Utc::now().timestamp(); let schedule = coord .scheduled_prompts .get(schedule_id)? From 9f5ce4d941bb96b79b2bbc951e80718477f73b0f Mon Sep 17 00:00:00 2001 From: damocles Date: Sun, 2 Aug 2026 02:04:00 +0200 Subject: [PATCH 5/5] hive-sh4re: delete now_unix(), the migration's last call sites are gone --- hive-agent/src/turn.rs | 3 ++- hive-c0re/src/actions.rs | 2 +- hive-sh4re/src/wire_time.rs | 18 +++++++----------- 3 files changed, 10 insertions(+), 13 deletions(-) diff --git a/hive-agent/src/turn.rs b/hive-agent/src/turn.rs index f099d7aba..e6ea396c3 100644 --- a/hive-agent/src/turn.rs +++ b/hive-agent/src/turn.rs @@ -416,7 +416,8 @@ fn maybe_auto_reset(bus: &Bus) { if last_ended == 0 { return; // no completed turn yet } - // Compute idle seconds using the same clock as now_unix (unix epoch, i64). + // Compute idle seconds using the same clock semantics as elsewhere + // in this crate (unix epoch, i64 seconds via `Utc::now().timestamp()`). let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map_or(0, |d| d.as_secs()); diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index d3eb90c46..207ecf959 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -714,7 +714,7 @@ fn finish_approval( // Phase 5b: also fire on the dashboard event channel so the // browser moves the row out of pending into history without a // snapshot refetch. `approved` rows that succeed get the - // approval's logged resolved_at indirectly via `now_unix()`; + // approval's logged resolved_at indirectly via `Utc::now()`; // failures already wrote it via mark_failed above. let approval_kind = approval.kind.as_str(); let sha_short = approval diff --git a/hive-sh4re/src/wire_time.rs b/hive-sh4re/src/wire_time.rs index 508507475..661cd78af 100644 --- a/hive-sh4re/src/wire_time.rs +++ b/hive-sh4re/src/wire_time.rs @@ -15,17 +15,13 @@ pub fn from_secs(secs: i64) -> DateTime { DateTime::::from_timestamp(secs, 0).unwrap_or_default() } -/// Current unix timestamp in seconds — the single definition behind -/// every store's `created_at` / `sent_at` / … stamp (this module owns -/// the epoch-seconds convention; a dozen local copies of this fn used -/// to float around both binaries). Uses chrono's `clock` feature -/// (already pulled in workspace-wide) instead of hand-rolled -/// `SystemTime` math, consistent with every other timestamp in this -/// module being a `chrono` type. -#[must_use] -pub fn now_unix() -> i64 { - Utc::now().timestamp() -} +// `now_unix()` (a hand-rolled-`SystemTime`-math-turned-`chrono` helper +// that every store used to call for its `created_at` / `sent_at` / … +// stamps) has been removed — every call site now uses +// `chrono::Utc::now()` directly, either as a `DateTime` struct +// field or via `.timestamp()` for the ephemeral i64-typed (sqlite +// bind / cutoff arithmetic) call sites. hivectl's own separate +// `now_unix()` copy in `dag_progress.rs` is unrelated and untouched. #[cfg(test)] mod tests {