diff --git a/Cargo.lock b/Cargo.lock index 0b69f7f0d..bf7ac8006 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1562,7 +1562,6 @@ version = "0.1.0" dependencies = [ "anyhow", "axum", - "chrono", "clap", "futures-util", "hive-agent-sock", @@ -1623,7 +1622,6 @@ version = "0.1.0" dependencies = [ "anyhow", "axum", - "chrono", "clap", "hive-agent-sock", "hive-sh4re", diff --git a/Cargo.toml b/Cargo.toml index db61bb2d5..9b509513c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -45,7 +45,6 @@ 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-agent/Cargo.toml b/hive-agent/Cargo.toml index a4029b5c2..3dd41c259 100644 --- a/hive-agent/Cargo.toml +++ b/hive-agent/Cargo.toml @@ -10,7 +10,6 @@ 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/db_migrate.rs b/hive-agent/src/db_migrate.rs index 9feca95b5..503a001b9 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.timestamp(), 1000); + assert_eq!(pending[0].due_at, 1000); } #[test] diff --git a/hive-agent/src/events.rs b/hive-agent/src/events.rs index cf585f5f9..8cf1f6c3d 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, Utc::now().timestamp()))), + state: Arc::new(Mutex::new((TurnState::Idle, now_unix()))), 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(Utc::now().timestamp(), Ordering::Relaxed); + .store(now_unix(), Ordering::Relaxed); self.emit(LiveEvent::TokenUsageChanged { ctx, cost }); } @@ -778,7 +778,7 @@ impl Bus { if guard.0 == next { return; } - *guard = (next, Utc::now().timestamp()); + *guard = (next, now_unix()); since = guard.1; } self.emit(LiveEvent::TurnStateChanged { @@ -861,7 +861,7 @@ impl Bus { } let envelope = BusEvent { seq: self.next_seq(), - ts: Utc::now().timestamp(), + ts: now_unix(), 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 a22f27226..6d578de80 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 = chrono::Utc::now().timestamp(); + let started_at = serve_common::now_unix(); 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 = chrono::Utc::now().timestamp(); + let ended_at = serve_common::now_unix(); 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 62a3fd956..c2c5f8ca7 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 chrono::{DateTime, Utc}; +use hive_sh4re::wire_time::now_unix; 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: DateTime, + pub asked_at: i64, } /// 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, Utc::now().timestamp()], + params![id, role.as_str(), peer, question, now_unix()], )?; Ok(()) } @@ -170,14 +170,12 @@ 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: chrono::DateTime::from_timestamp(asked_at_secs, 0) - .unwrap_or_else(Utc::now), + asked_at: row.get(4)?, }); } Ok(out) diff --git a/hive-agent/src/reminder_timer.rs b/hive-agent/src/reminder_timer.rs index 8d8573437..326f8ef50 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 = chrono::Utc::now().timestamp(); + let now = hive_sh4re::wire_time::now_unix(); 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 = chrono::Utc::now().timestamp(); + let now = hive_sh4re::wire_time::now_unix(); 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 f5837d919..4f4936f3b 100644 --- a/hive-agent/src/reminders.rs +++ b/hive-agent/src/reminders.rs @@ -17,9 +17,8 @@ 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; +use hive_sh4re::wire_time::now_unix; use rusqlite::{Connection, params}; const SCHEMA: &str = r" @@ -43,8 +42,8 @@ pub struct Reminder { pub id: i64, pub message: String, pub file_path: Option, - pub due_at: DateTime, - pub created_at: DateTime, + pub due_at: i64, + pub created_at: i64, } /// The harness-local reminder store. Cheap to share behind an `Arc`; the @@ -82,7 +81,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 = Utc::now().timestamp(); + let now = now_unix(); conn.execute( "INSERT INTO reminders (message, file_path, due_at, created_at, sent_at) \ VALUES (?1, ?2, ?3, ?4, NULL)", @@ -171,7 +170,7 @@ impl Reminders { let conn = self.conn.lock().unwrap(); conn.execute( "UPDATE reminders SET sent_at = ?1 WHERE id = ?2", - params![Utc::now().timestamp(), id], + params![now_unix(), id], )?; Ok(()) } @@ -211,7 +210,7 @@ impl Reminders { pub fn rollup(&self, since_secs: i64) -> Result { let conn = self.conn.lock().unwrap(); let cutoff = if since_secs > 0 { - Utc::now().timestamp().saturating_sub(since_secs) + now_unix().saturating_sub(since_secs) } else { i64::MIN }; @@ -259,14 +258,12 @@ 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: wire_time::from_secs(due_at_secs), - created_at: wire_time::from_secs(created_at_secs), + due_at: row.get(3)?, + created_at: row.get(4)?, }) } @@ -290,7 +287,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.timestamp(), 1000); + assert_eq!(pending[0].due_at, 1000); assert_eq!(s.count_pending().unwrap(), 1); } @@ -351,7 +348,7 @@ mod tests { ) .unwrap(); } - let cutoff = Utc::now().timestamp() - 10; + let cutoff = now_unix() - 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 778aee51c..c3fdcf6fe 100644 --- a/hive-agent/src/serve_common.rs +++ b/hive-agent/src/serve_common.rs @@ -7,6 +7,7 @@ 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 ff018018d..91563fdbd 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 = Utc::now().timestamp(); + let now = now_unix(); 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 = Utc::now().timestamp(); + let now = now_unix(); // 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 = Utc::now().timestamp(); + let now = now_unix(); seed_db( &db, &[ @@ -766,17 +766,7 @@ mod tests { fn bash_breakdown_empty_without_table() { let db = tmp_db(); let _ = std::fs::remove_file(&db); - seed_db( - &db, - &[( - Utc::now().timestamp() - 100, - 1000, - "opus", - "recv", - "ok", - "{}", - )], - ); + seed_db(&db, &[(now_unix() - 100, 1000, "opus", "recv", "ok", "{}")]); let s = snapshot(&db, Window::Day).unwrap(); assert!(s.bash_breakdown.is_empty()); } @@ -788,7 +778,7 @@ mod tests { let db = tmp_db(); let _ = std::fs::remove_file(&db); seed_db(&db, &[]); - let now = Utc::now().timestamp(); + let now = now_unix(); 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 c307e0270..f9a811674 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 = chrono::Utc::now().timestamp(); - let age = u64::try_from(now.saturating_sub(q.asked_at.timestamp())).unwrap_or(0); + let now = hive_sh4re::wire_time::now_unix(); + let age = u64::try_from(now.saturating_sub(q.asked_at)).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 = chrono::Utc::now().timestamp(); - let age = u64::try_from(now.saturating_sub(r.created_at.timestamp())).unwrap_or(0); + let now = hive_sh4re::wire_time::now_unix(); + let age = u64::try_from(now.saturating_sub(r.created_at)).unwrap_or(0); LooseEnd::Reminder { id: r.id, owner: crate::identity::label(), message: r.message, - due_at: r.due_at, + due_at: hive_sh4re::wire_time::from_secs(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 = chrono::Utc::now().timestamp(); - let age = u64::try_from(now.saturating_sub(t.updated_at.timestamp())).unwrap_or(0); + let now = hive_sh4re::wire_time::now_unix(); + let age = u64::try_from(now.saturating_sub(t.updated_at)).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 4fc084657..63deb11da 100644 --- a/hive-agent/src/todos.rs +++ b/hive-agent/src/todos.rs @@ -32,8 +32,7 @@ use std::path::Path; use std::sync::Mutex; use anyhow::{Context, Result}; -use chrono::{DateTime, Utc}; -use hive_sh4re::wire_time; +use hive_sh4re::wire_time::now_unix; use rusqlite::{Connection, params}; /// SQL bootstrap. `CREATE TABLE IF NOT EXISTS` so first-boot agents and @@ -67,8 +66,8 @@ const MIGRATIONS: &[&str] = &[ "ALTER TABLE todos ADD COLUMN acked_at INTEGER", ]; -/// One dynamic, subsystem-pushed todo. The consumer derives `age_seconds` -/// from `updated_at`. +/// One dynamic, subsystem-pushed todo. Timestamps are unix seconds; the +/// consumer derives `age_seconds` from `updated_at`. #[derive(Debug, Clone)] pub struct Todo { pub id: i64, @@ -81,7 +80,7 @@ pub struct Todo { pub summary: String, /// Optional free-text provenance (e.g. the room name / task label). pub source: Option, - pub updated_at: DateTime, + pub updated_at: i64, } /// The harness-local todo store. Cheap to share behind an `Arc`; the inner @@ -145,7 +144,7 @@ impl Todos { source: Option<&str>, ) -> Result<(i64, bool)> { let conn = self.conn.lock().unwrap(); - let now = Utc::now().timestamp(); + let now = now_unix(); let existing: Option<(i64, String, Option)> = if key.is_some() { conn.query_row( "SELECT id, summary, source FROM todos \ @@ -240,7 +239,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![Utc::now().timestamp(), id], + params![now_unix(), id], )?; Ok(n) } @@ -267,14 +266,13 @@ 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: wire_time::from_secs(updated_at_secs), + updated_at: row.get(5)?, }) })? .collect::>>()?; @@ -493,12 +491,12 @@ mod tests { let conn = s.conn.lock().unwrap(); conn.execute( "UPDATE todos SET acked_at = ?1 WHERE id = ?2", - params![Utc::now().timestamp() - 1000, old_id], + params![now_unix() - 1000, old_id], ) .unwrap(); } - let removed = s.reap_acked(Utc::now().timestamp() - 500).unwrap(); + let removed = s.reap_acked(now_unix() - 500).unwrap(); assert_eq!(removed, 1, "only the backdated row is past the cutoff"); let conn = Connection::open(dir.path().join("todos.sqlite")).unwrap(); diff --git a/hive-agent/src/turn.rs b/hive-agent/src/turn.rs index e6ea396c3..f099d7aba 100644 --- a/hive-agent/src/turn.rs +++ b/hive-agent/src/turn.rs @@ -416,8 +416,7 @@ fn maybe_auto_reset(bus: &Bus) { if last_ended == 0 { return; // no completed turn yet } - // Compute idle seconds using the same clock semantics as elsewhere - // in this crate (unix epoch, i64 seconds via `Utc::now().timestamp()`). + // Compute idle seconds using the same clock as now_unix (unix epoch, i64). let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map_or(0, |d| d.as_secs()); diff --git a/hive-agent/src/vacuum.rs b/hive-agent/src/vacuum.rs index fa37184fb..1c77d86d9 100644 --- a/hive-agent/src/vacuum.rs +++ b/hive-agent/src/vacuum.rs @@ -17,6 +17,7 @@ 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. @@ -55,8 +56,7 @@ fn sweep_once() { let tasks_dir = harness.join("bash-tasks"); if tasks_dir.is_dir() { - let removed = - vacuum_bash_tasks(&tasks_dir, chrono::Utc::now().timestamp() - BASH_KEEP_SECS); + let removed = vacuum_bash_tasks(&tasks_dir, now_unix() - 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(chrono::Utc::now().timestamp() - REMINDER_KEEP_SECS) + store.prune_delivered_older_than(now_unix() - 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(chrono::Utc::now().timestamp() - TODO_ACKED_KEEP_SECS) + store.reap_acked(now_unix() - TODO_ACKED_KEEP_SECS) } /// Delete eligible bash-task trios in `dir`. Returns the count of `.json` @@ -140,15 +140,10 @@ 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_str) - .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok()) - .map_or(i64::MAX, |dt| dt.timestamp()); + .and_then(serde_json::Value::as_i64) + .unwrap_or(i64::MAX); completed_at < cutoff } @@ -169,7 +164,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 = chrono::Utc::now().timestamp() - STREAM_KEEP_SECS; + let cutoff = now_unix() - 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 0c4f36ebf..a11592d8d 100644 --- a/hive-bash-mcp/Cargo.toml +++ b/hive-bash-mcp/Cargo.toml @@ -10,7 +10,6 @@ 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 59545a0ad..2653f39c9 100644 --- a/hive-bash-mcp/src/mcp.rs +++ b/hive-bash-mcp/src/mcp.rs @@ -38,12 +38,11 @@ fn format_task(task: &TaskFile) -> String { let _ = write!(out, ", exit={code}"); } if let (Some(started), None) = (task.started_at, task.completed_at) { - let elapsed = (chrono::Utc::now() - started).num_seconds(); - let _ = write!(out, ", running for {elapsed}s"); + let now = hive_sh4re::wire_time::now_unix(); + let _ = write!(out, ", running for {}s", now - started); } if let (Some(completed), Some(started)) = (task.completed_at, task.started_at) { - let took = (completed - started).num_seconds(); - let _ = write!(out, ", took {took}s"); + let _ = write!(out, ", took {}s", completed - started); } let out_file = crate::paths::task_out(&task.id); @@ -296,8 +295,8 @@ mod status_hint_tests { cmd: "echo hi".to_owned(), timeout_secs: None, status, - created_at: chrono::DateTime::from_timestamp(1, 0).unwrap_or_default(), - started_at: Some(chrono::DateTime::from_timestamp(1, 0).unwrap_or_default()), + created_at: 1, + started_at: Some(1), 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 438bf4876..0ddb8874a 100644 --- a/hive-bash-mcp/src/runner.rs +++ b/hive-bash-mcp/src/runner.rs @@ -133,6 +133,8 @@ 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 { @@ -269,7 +271,7 @@ pub fn submit_task(cmd: String, timeout_secs: Option, name: Option) cmd, timeout_secs, status: TaskStatus::Pending, - created_at: chrono::Utc::now(), + created_at: now_unix(), started_at: None, completed_at: None, exit_code: None, @@ -375,7 +377,7 @@ pub fn kill_task(id: &str, force: bool) -> (bool, bool) { && task.status == TaskStatus::Pending { task.status = TaskStatus::Killed; - task.completed_at = Some(chrono::Utc::now()); + task.completed_at = Some(now_unix()); 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 — @@ -433,7 +435,7 @@ async fn mark_interrupted(socket: &Path) { } tracing::warn!(id = %id, "bash_runner: marking interrupted task"); task.status = TaskStatus::Interrupted; - task.completed_at = Some(chrono::Utc::now()); + task.completed_at = Some(now_unix()); if let Err(e) = write_task(&task) { tracing::warn!(id = %id, error = ?e, "bash_runner: write interrupted state failed"); } @@ -488,7 +490,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(chrono::Utc::now()); + task.started_at = Some(now_unix()); if let Err(e) = write_task(&task) { tracing::warn!(id = %id, error = ?e, "bash_runner: write running state failed"); } @@ -557,7 +559,7 @@ async fn run_task(mut task: TaskFile, socket: &Path) { }; task.status = status; - task.completed_at = Some(chrono::Utc::now()); + task.completed_at = Some(now_unix()); 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-c0re/src/actions.rs b/hive-c0re/src/actions.rs index 207ecf959..d3eb90c46 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 `Utc::now()`; + // approval's logged resolved_at indirectly via `now_unix()`; // failures already wrote it via mark_failed above. let approval_kind = approval.kind.as_str(); let sha_short = approval diff --git a/hive-c0re/src/dashboard/state_snapshot.rs b/hive-c0re/src/dashboard/state_snapshot.rs index 27309cc0c..04c439357 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: (chrono::Utc::now() - st.since) + secs: (hive_sh4re::wire_time::from_secs(hive_sh4re::wire_time::now_unix()) - 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 bbe909250..ada7faad6 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -44,6 +44,7 @@ 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; @@ -106,7 +107,7 @@ struct NodeRuntime { struct DagMeta { source: Source, reason: String, - created_at: DateTime, + created_at: i64, } /// The mutable queue state behind the mutex: the crate scheduler plus the @@ -229,7 +230,7 @@ impl JobQueue { NodeKind::Dag { source: spec.source, reason: spec.reason, - created_at: Utc::now(), + created_at: now_unix(), }, Vec::new(), None, @@ -468,7 +469,9 @@ 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(Utc::now), + since: n + .started_at + .unwrap_or_else(|| hive_sh4re::wire_time::from_secs(now_unix())), }) }) .collect() @@ -625,7 +628,7 @@ impl QueueInner { id: container.get(), source: meta.source, reason: meta.reason.clone(), - created_at: meta.created_at, + created_at: hive_sh4re::wire_time::from_secs(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 77ba0757e..96fba3c38 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -12,7 +12,6 @@ //! 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; @@ -284,7 +283,7 @@ pub enum NodeKind { Dag { source: Source, reason: String, - created_at: DateTime, + created_at: i64, }, } diff --git a/hive-c0re/src/loose_ends.rs b/hive-c0re/src/loose_ends.rs index b20ef567a..621419fb8 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 crate::coordinator::Coordinator; /// Propagates errors from `count_pending` and the pending-approval /// sqlite query. pub fn for_agent(coord: &Coordinator, agent: &str) -> Result> { - let now = Utc::now().timestamp(); + let now = now_unix(); 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 = Utc::now().timestamp(); + let now = now_unix(); 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 aa43e99ca..bebe855d3 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: s.next_fire_at_unix, - created_at_unix: s.created_at_unix, + 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), 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, - paused_at_unix: s.paused_at_unix, + 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), description: s.description, targets: s .targets .into_iter() .map(|t| hive_sh4re::WireScheduleTarget { target: t.target, - cancelled_at_unix: t.cancelled_at_unix, - last_fired_at_unix: t.last_fired_at_unix, + 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), 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 94b1e8c98..7c4fea232 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 chrono::Utc; +use hive_sh4re::wire_time::now_unix; /// 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 = Utc::now().timestamp(); + let now = now_unix(); // 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 = Utc::now().timestamp(); + let now = now_unix(); 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 651af2654..313f04e44 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 chrono::Utc; +use hive_sh4re::wire_time::now_unix; use hive_sh4re::{Approval, ApprovalKind, ApprovalStatus}; use rusqlite::{Connection, OptionalExtension, params}; @@ -95,7 +95,7 @@ impl Approvals { agent, kind.as_str(), commit_ref, - Utc::now().timestamp(), + now_unix(), 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 = Utc::now().timestamp(); + let resolved_at = now_unix(); 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![Utc::now().timestamp(), note, id], + params![now_unix(), 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![Utc::now().timestamp(), note, id], + params![now_unix(), note, id], )?; Ok(()) } @@ -267,7 +267,7 @@ impl Approvals { if row.status != "pending" { bail!("approval {id} is {}, not pending", row.status); } - let resolved_at = Utc::now().timestamp(); + let resolved_at = now_unix(); 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![Utc::now().timestamp(), note, agent], + params![now_unix(), note, agent], )?; Ok(n) } diff --git a/hive-c0re/src/stores/audit_log.rs b/hive-c0re/src/stores/audit_log.rs index eb1cd377d..8e8bc1608 100644 --- a/hive-c0re/src/stores/audit_log.rs +++ b/hive-c0re/src/stores/audit_log.rs @@ -25,6 +25,7 @@ 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; @@ -140,7 +141,7 @@ impl AuditLog { outcome: AuditOutcome, detail: Option<&str>, ) -> Option { - let now = Utc::now().timestamp(); + let now = now_unix(); let conn = self.conn.lock().unwrap(); match conn.execute( "INSERT INTO audit_log (ts_unix, agent, action, target, outcome, detail) @@ -207,7 +208,7 @@ impl AuditLog { /// # Errors /// Returns an error if the `DELETE` query fails. pub fn vacuum(&self) -> Result { - let cutoff = Utc::now().timestamp() - KEEP_SECS; + let cutoff = now_unix() - 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)) @@ -310,7 +311,7 @@ mod tests { let conn = db.conn.lock().unwrap(); conn.execute( "UPDATE audit_log SET ts_unix = ?1", - params![Utc::now().timestamp() - KEEP_SECS - 60], + params![now_unix() - KEEP_SECS - 60], ) .unwrap(); } diff --git a/hive-c0re/src/stores/broker.rs b/hive-c0re/src/stores/broker.rs index 76e7dd516..729bbeed6 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 = Utc::now().timestamp(); + let now = now_unix(); // 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 = Utc::now().timestamp(); + let now = now_unix(); 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 = Utc::now().timestamp() - older_than_secs; + let cutoff = now_unix() - 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 = Utc::now().timestamp(); + let now = now_unix(); 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 = Utc::now().timestamp(); + let now = now_unix(); 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![Utc::now().timestamp(), recipient, up_to], + params![now_unix(), 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 = Utc::now().timestamp(); + let now = now_unix(); // 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 24b302bdc..678a479ac 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 chrono::Utc; +use hive_sh4re::wire_time::now_unix; 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 = Utc::now().timestamp(); + let now = now_unix(); 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 = Utc::now().timestamp(); + let now = now_unix(); 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 = Utc::now().timestamp(); + let now = now_unix(); 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 = Utc::now().timestamp(); + let now = now_unix(); { 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 089eef903..64f7b3509 100644 --- a/hive-c0re/src/stores/operator_questions.rs +++ b/hive-c0re/src/stores/operator_questions.rs @@ -15,6 +15,7 @@ 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; @@ -118,7 +119,7 @@ impl OperatorQuestions { i64::from(multi), deadline_at, target, - Utc::now().timestamp(), + now_unix(), ], )?; Ok(conn.last_insert_rowid()) @@ -176,7 +177,7 @@ impl OperatorQuestions { } conn.execute( "UPDATE operator_questions SET answer = ?1, answered_at = ?2 WHERE id = ?3", - params![answer, Utc::now().timestamp(), id], + params![answer, now_unix(), id], )?; Ok((question, asker, target)) } @@ -224,7 +225,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, Utc::now().timestamp(), id], + params![sentinel, now_unix(), id], )?; Ok((question, asker, target)) } diff --git a/hive-c0re/src/stores/power.rs b/hive-c0re/src/stores/power.rs index 6aa6e3844..a5900adbb 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 chrono::Utc; +use hive_sh4re::wire_time::now_unix; 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(), Utc::now().timestamp()], + params![agent, wanted.as_str(), now_unix()], ) .context("upsert agent_power")?; Ok(()) diff --git a/hive-c0re/src/stores/scheduled_prompts.rs b/hive-c0re/src/stores/scheduled_prompts.rs index 75bff2a30..f0877a5c1 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 chrono::{DateTime, Utc}; +use hive_sh4re::wire_time::now_unix; 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: DateTime, - pub created_at_unix: DateTime, + pub next_fire_at_unix: i64, + pub created_at_unix: i64, 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, - Utc::now().timestamp(), + now_unix(), 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 = Utc::now().timestamp(); + let now = now_unix(); 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 = Utc::now().timestamp(); + let now = now_unix(); 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 = Utc::now().timestamp(); + let now = now_unix(); 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 = Utc::now().timestamp(); + let now = now_unix(); let n = conn.execute( "UPDATE scheduled_prompts SET paused_at_unix = COALESCE(paused_at_unix, ?1) @@ -653,16 +653,12 @@ 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: hive_sh4re::wire_time::from_secs(row.get(4)?), - created_at_unix: hive_sh4re::wire_time::from_secs(row.get(5)?), + next_fire_at_unix: row.get(4)?, + created_at_unix: row.get(5)?, source: ScheduleSource::from_db_string(&source_str), - cancelled_at_unix: row - .get::<_, Option>(7)? - .map(hive_sh4re::wire_time::from_secs), + cancelled_at_unix: row.get(7)?, description: row.get(8)?, - paused_at_unix: row - .get::<_, Option>(9)? - .map(hive_sh4re::wire_time::from_secs), + paused_at_unix: row.get(9)?, targets: Vec::new(), }) } @@ -677,12 +673,8 @@ 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), + cancelled_at_unix: row.get(1)?, + last_fired_at_unix: row.get(2)?, last_result: row.get(3)?, }) })?; @@ -779,7 +771,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.timestamp(), 460); + assert_eq!(s.next_fire_at_unix, 460); } #[test] @@ -801,7 +793,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.timestamp(), 1_234); + assert_eq!(s.next_fire_at_unix, 1_234); } #[test] @@ -822,7 +814,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.timestamp(), 160); + assert_eq!(s.next_fire_at_unix, 160); } #[test] @@ -833,7 +825,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.timestamp(), 100); + assert_eq!(s.next_fire_at_unix, 100); } #[test] @@ -886,7 +878,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.map(|dt| dt.timestamp()), Some(200)); + assert_eq!(bob.last_fired_at_unix, Some(200)); assert_eq!(bob.last_result.as_deref(), Some("ok")); } @@ -917,7 +909,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.timestamp(), 100); + assert_eq!(s.next_fire_at_unix, 100); } #[test] diff --git a/hive-c0re/src/workers/scheduled_prompts_worker.rs b/hive-c0re/src/workers/scheduled_prompts_worker.rs index b7dac0816..4adb0743a 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 = Utc::now().timestamp(); + let now = now_unix(); 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 = Utc::now().timestamp(); + let now = now_unix(); let schedule = coord .scheduled_prompts .get(schedule_id)? diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index d2226c78b..8ca7570c0 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: DateTime, + pub created_at: i64, #[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). diff --git a/hive-sh4re/src/wire_time.rs b/hive-sh4re/src/wire_time.rs index 661cd78af..2d9529013 100644 --- a/hive-sh4re/src/wire_time.rs +++ b/hive-sh4re/src/wire_time.rs @@ -15,13 +15,18 @@ pub fn from_secs(secs: i64) -> DateTime { DateTime::::from_timestamp(secs, 0).unwrap_or_default() } -// `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. +/// 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. +#[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) +} #[cfg(test)] mod tests {