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

hive-agent: finish chrono-clock migration on remaining call sites

Browse source
This commit is contained in:
damocles 2026-08-01 23:49:03 +02:00 • committed by mara
commit f4a35786b1
10 changed files with 63 additions and 47 deletions
Download patch file Download diff file Expand all files Collapse all files

2
hive-agent/src/db_migrate.rs
View file

@ -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]

10
hive-agent/src/events.rs
View file

@ -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.

4
hive-agent/src/main.rs
View file

@ -770,7 +770,7 @@ async fn handle_turn<S: Surface>(
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<S: Surface>(
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 {

10
hive-agent/src/questions.rs
View file

@ -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<Utc>,
}
/// 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)

4
hive-agent/src/reminder_timer.rs
View file

@ -74,7 +74,7 @@ pub async fn run(
}
fn tick(store: &Reminders, tx: &mpsc::UnboundedSender<hive_sh4re::DeliveredMessage>) {
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}");
}

23
hive-agent/src/reminders.rs
View file

@ -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<String>,
pub due_at: i64,
pub created_at: i64,
pub due_at: DateTime<Utc>,
pub created_at: DateTime<Utc>,
}
/// 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<i64> {
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<ReminderStats> {
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<Reminder> {
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<i64> = {

1
hive-agent/src/serve_common.rs
View file

@ -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

22
hive-agent/src/stats.rs
View file

@ -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<Snapshot> {
// 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();

14
hive-agent/src/todo_server.rs
View file

@ -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,

20
hive-agent/src/todos.rs
View file

@ -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<String>,
pub updated_at: i64,
pub updated_at: DateTime<Utc>,
}
/// 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<String>)> = 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::<rusqlite::Result<Vec<_>>>()?;
@ -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();