hive-sh4re/hive-bash-mcp/hive-agent: retype TaskFile timestamps to DateTime<Utc>, drop now_unix from these crates

This commit is contained in:
damocles 2026-08-01 21:34:29 +02:00 committed by mara
commit 2122e23d81
7 changed files with 30 additions and 22 deletions

2
Cargo.lock generated
View file

@ -1562,6 +1562,7 @@ version = "0.1.0"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"axum", "axum",
"chrono",
"clap", "clap",
"futures-util", "futures-util",
"hive-agent-sock", "hive-agent-sock",
@ -1622,6 +1623,7 @@ version = "0.1.0"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"axum", "axum",
"chrono",
"clap", "clap",
"hive-agent-sock", "hive-agent-sock",
"hive-sh4re", "hive-sh4re",

View file

@ -10,6 +10,7 @@ workspace = true
[dependencies] [dependencies]
anyhow.workspace = true anyhow.workspace = true
axum.workspace = true axum.workspace = true
chrono.workspace = true
reqwest.workspace = true reqwest.workspace = true
hyper.workspace = true hyper.workspace = true
hyper-util.workspace = true hyper-util.workspace = true

View file

@ -17,7 +17,6 @@
use std::path::Path; use std::path::Path;
use std::time::Duration; use std::time::Duration;
use hive_sh4re::wire_time::now_unix;
use rusqlite::{Connection, Result, params}; use rusqlite::{Connection, Result, params};
/// How often the sweep runs. /// How often the sweep runs.
@ -56,7 +55,8 @@ fn sweep_once() {
let tasks_dir = harness.join("bash-tasks"); let tasks_dir = harness.join("bash-tasks");
if tasks_dir.is_dir() { 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 { if removed > 0 {
tracing::info!(removed, "bash-tasks vacuum"); tracing::info!(removed, "bash-tasks vacuum");
} }
@ -92,14 +92,14 @@ fn sweep_once() {
/// `Reminders` handle). /// `Reminders` handle).
fn vacuum_reminders(path: &Path) -> anyhow::Result<usize> { fn vacuum_reminders(path: &Path) -> anyhow::Result<usize> {
let store = crate::reminders::Reminders::open(path)?; 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 /// Reap acked todo rows older than [`TODO_ACKED_KEEP_SECS`] via the typed
/// store API — same own-short-lived-connection shape as `vacuum_reminders`. /// store API — same own-short-lived-connection shape as `vacuum_reminders`.
fn vacuum_todos(path: &Path) -> anyhow::Result<usize> { fn vacuum_todos(path: &Path) -> anyhow::Result<usize> {
let store = crate::todos::Todos::open(path)?; 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` /// 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) { if !TERMINAL_STATUSES.contains(&status) {
return false; return false;
} }
// `completed_at` serializes as an RFC3339 string (`TaskFile.completed_at`
// is `Option<DateTime<Utc>>`), 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 let completed_at = v
.get("completed_at") .get("completed_at")
.and_then(serde_json::Value::as_i64) .and_then(serde_json::Value::as_str)
.unwrap_or(i64::MAX); .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok())
.map_or(i64::MAX, |dt| dt.timestamp());
completed_at < cutoff completed_at < cutoff
} }
@ -164,7 +169,7 @@ fn delete_trio(dir: &Path, stem: &str) {
/// agent's `events.sqlite`. Returns the number of rows deleted. /// agent's `events.sqlite`. Returns the number of rows deleted.
fn vacuum_events(path: &Path) -> Result<u64> { fn vacuum_events(path: &Path) -> Result<u64> {
let conn = Connection::open(path)?; 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( let removed = conn.execute(
"DELETE FROM events WHERE kind = 'stream' AND ts < ?1", "DELETE FROM events WHERE kind = 'stream' AND ts < ?1",
params![cutoff], params![cutoff],

View file

@ -10,6 +10,7 @@ workspace = true
[dependencies] [dependencies]
anyhow.workspace = true anyhow.workspace = true
axum.workspace = true axum.workspace = true
chrono.workspace = true
clap.workspace = true clap.workspace = true
hive-agent-sock.workspace = true hive-agent-sock.workspace = true
hive-sh4re.workspace = true hive-sh4re.workspace = true

View file

@ -38,11 +38,12 @@ fn format_task(task: &TaskFile) -> String {
let _ = write!(out, ", exit={code}"); let _ = write!(out, ", exit={code}");
} }
if let (Some(started), None) = (task.started_at, task.completed_at) { if let (Some(started), None) = (task.started_at, task.completed_at) {
let now = hive_sh4re::wire_time::now_unix(); let elapsed = (chrono::Utc::now() - started).num_seconds();
let _ = write!(out, ", running for {}s", now - started); let _ = write!(out, ", running for {elapsed}s");
} }
if let (Some(completed), Some(started)) = (task.completed_at, task.started_at) { 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); let out_file = crate::paths::task_out(&task.id);
@ -295,8 +296,8 @@ mod status_hint_tests {
cmd: "echo hi".to_owned(), cmd: "echo hi".to_owned(),
timeout_secs: None, timeout_secs: None,
status, status,
created_at: 1, created_at: chrono::DateTime::from_timestamp(1, 0).unwrap_or_default(),
started_at: Some(1), started_at: Some(chrono::DateTime::from_timestamp(1, 0).unwrap_or_default()),
completed_at: None, completed_at: None,
exit_code: None, exit_code: None,
stdout_tail: None, stdout_tail: None,

View file

@ -133,8 +133,6 @@ fn signal_group(pgid: Option<i32>, sig: i32) {
// Helpers // Helpers
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
use hive_sh4re::wire_time::now_unix;
/// Generate a task ID: `<timestamp_hex><seq_hex>`. /// Generate a task ID: `<timestamp_hex><seq_hex>`.
#[must_use] #[must_use]
pub fn new_task_id() -> String { pub fn new_task_id() -> String {
@ -271,7 +269,7 @@ pub fn submit_task(cmd: String, timeout_secs: Option<u64>, name: Option<String>)
cmd, cmd,
timeout_secs, timeout_secs,
status: TaskStatus::Pending, status: TaskStatus::Pending,
created_at: now_unix(), created_at: chrono::Utc::now(),
started_at: None, started_at: None,
completed_at: None, completed_at: None,
exit_code: 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::Pending
{ {
task.status = TaskStatus::Killed; task.status = TaskStatus::Killed;
task.completed_at = Some(now_unix()); task.completed_at = Some(chrono::Utc::now());
let _ = write_task(&task); let _ = write_task(&task);
// A Pending task never reached Running, so it has no keyed "active" // A Pending task never reached Running, so it has no keyed "active"
// todo to clear and (like the old code) fires no completion signal — // 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"); tracing::warn!(id = %id, "bash_runner: marking interrupted task");
task.status = TaskStatus::Interrupted; task.status = TaskStatus::Interrupted;
task.completed_at = Some(now_unix()); task.completed_at = Some(chrono::Utc::now());
if let Err(e) = write_task(&task) { if let Err(e) = write_task(&task) {
tracing::warn!(id = %id, error = ?e, "bash_runner: write interrupted state failed"); 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"); tracing::info!(id = %id, cmd = %task.cmd, "bash_runner: starting task");
task.status = TaskStatus::Running; task.status = TaskStatus::Running;
task.started_at = Some(now_unix()); task.started_at = Some(chrono::Utc::now());
if let Err(e) = write_task(&task) { if let Err(e) = write_task(&task) {
tracing::warn!(id = %id, error = ?e, "bash_runner: write running state failed"); 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.status = status;
task.completed_at = Some(now_unix()); task.completed_at = Some(chrono::Utc::now());
task.exit_code = exit_code; task.exit_code = exit_code;
task.stdout_tail = stdout_tail.clone().filter(|s| !s.is_empty()); task.stdout_tail = stdout_tail.clone().filter(|s| !s.is_empty());
task.stderr_tail = stderr_tail.clone().filter(|s| !s.is_empty()); task.stderr_tail = stderr_tail.clone().filter(|s| !s.is_empty());

View file

@ -337,11 +337,11 @@ pub struct TaskFile {
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub timeout_secs: Option<u64>, pub timeout_secs: Option<u64>,
pub status: TaskStatus, pub status: TaskStatus,
pub created_at: i64, pub created_at: DateTime<Utc>,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub started_at: Option<i64>, pub started_at: Option<DateTime<Utc>>,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub completed_at: Option<i64>, pub completed_at: Option<DateTime<Utc>>,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub exit_code: Option<i32>, pub exit_code: Option<i32>,
/// Last `SUMMARY_BYTES` of stdout (see `hive-bash-mcp` runner). /// Last `SUMMARY_BYTES` of stdout (see `hive-bash-mcp` runner).