use std::sync::Arc;
use std::time::Duration;
use parking_lot::Mutex;
use tokio::sync::mpsc::UnboundedSender;
use tracing::{debug, info, warn};
use crate::error::{Result, TokenError};
use crate::seed_agent::{
SeedAgentEarmarkManager, SeedAgentStatus, MONTH_MILLIS,
};
use crate::seed_agent_gossip::SeedAgentGossipMessage;
use tenzro_types::primitives::Timestamp;
pub const DEFAULT_DAEMON_POLL_INTERVAL_SECS: u64 = 6 * 60 * 60;
pub const DEFAULT_MIN_REFILL_INTERVAL_MS: i64 = MONTH_MILLIS;
pub const DEFAULT_DAEMON_QUARANTINE_GRACE_MS: i64 =
crate::seed_agent::DEFAULT_QUARANTINE_GRACE_MS;
pub type TickAuthorityFn = Arc<dyn Fn() -> bool + Send + Sync + 'static>;
#[derive(Clone)]
pub struct SeedAgentDaemonConfig {
pub poll_interval_secs: u64,
pub min_refill_interval_ms: i64,
pub quarantine_grace_ms: i64,
}
impl Default for SeedAgentDaemonConfig {
fn default() -> Self {
Self {
poll_interval_secs: DEFAULT_DAEMON_POLL_INTERVAL_SECS,
min_refill_interval_ms: DEFAULT_MIN_REFILL_INTERVAL_MS,
quarantine_grace_ms: DEFAULT_DAEMON_QUARANTINE_GRACE_MS,
}
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct TickOutcome {
pub paused_count: u32,
pub refilled_count: u32,
pub total_granted_wei: u128,
pub master_switch_skipped: bool,
pub authority_skipped: bool,
pub quarantined_count: u32,
pub terminated_count: u32,
pub surplus_disposed_wei: u128,
}
pub type SurplusDispositionFn = Arc<
dyn Fn(&crate::seed_agent::SurplusDisposition) -> Result<()> + Send + Sync + 'static,
>;
pub struct SeedAgentDaemon {
manager: Arc<SeedAgentEarmarkManager>,
gossip_tx: Option<UnboundedSender<SeedAgentGossipMessage>>,
tick_authority: Option<TickAuthorityFn>,
surplus_disposition: Option<SurplusDispositionFn>,
config: SeedAgentDaemonConfig,
last_outcome: Mutex<Option<TickOutcome>>,
}
impl SeedAgentDaemon {
pub fn new(manager: Arc<SeedAgentEarmarkManager>) -> Self {
Self::with_config(manager, SeedAgentDaemonConfig::default())
}
pub fn with_config(
manager: Arc<SeedAgentEarmarkManager>,
config: SeedAgentDaemonConfig,
) -> Self {
Self {
manager,
gossip_tx: None,
tick_authority: None,
surplus_disposition: None,
config,
last_outcome: Mutex::new(None),
}
}
pub fn with_surplus_disposition(
mut self,
cb: SurplusDispositionFn,
) -> Self {
self.surplus_disposition = Some(cb);
self
}
pub fn with_gossip(
mut self,
gossip_tx: UnboundedSender<SeedAgentGossipMessage>,
) -> Self {
self.gossip_tx = Some(gossip_tx);
self
}
pub fn with_tick_authority(mut self, gate: TickAuthorityFn) -> Self {
self.tick_authority = Some(gate);
self
}
pub fn last_outcome(&self) -> Option<TickOutcome> {
self.last_outcome.lock().clone()
}
pub fn spawn(self: Arc<Self>) -> tokio::task::JoinHandle<()> {
let poll = Duration::from_secs(self.config.poll_interval_secs);
tokio::spawn(async move {
let mut ticker = tokio::time::interval(poll);
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
ticker.tick().await;
loop {
ticker.tick().await;
match self.tick_once(Timestamp::now()) {
Ok(outcome) => {
if outcome.refilled_count > 0 || outcome.paused_count > 0 {
info!(
refilled = outcome.refilled_count,
paused = outcome.paused_count,
granted_wei = %outcome.total_granted_wei,
"SeedAgentDaemon tick"
);
}
}
Err(e) => {
warn!(error = %e, "SeedAgentDaemon: tick_once failed");
}
}
}
})
}
pub fn tick_once(&self, now: Timestamp) -> Result<TickOutcome> {
let mut outcome = TickOutcome::default();
let earmark = self.manager.earmark();
if !earmark.enabled {
debug!("SeedAgentDaemon: earmark disabled, tick is a no-op");
outcome.master_switch_skipped = true;
*self.last_outcome.lock() = Some(outcome.clone());
return Ok(outcome);
}
if let Some(gate) = &self.tick_authority {
if !gate() {
debug!("SeedAgentDaemon: not tick authority, skipping");
outcome.authority_skipped = true;
*self.last_outcome.lock() = Some(outcome.clone());
return Ok(outcome);
}
}
for agent in self.manager.list_agents(None) {
if !matches!(agent.status, SeedAgentStatus::Active) {
continue;
}
let Some(charter) = self.manager.get_charter(&agent.charter_id) else {
if let Err(e) = self
.manager
.set_agent_status(&agent.agent_did, SeedAgentStatus::Paused)
{
warn!(
agent = %agent.agent_did,
error = %e,
"SeedAgentDaemon: failed to pause dangling agent"
);
continue;
}
outcome.paused_count = outcome.paused_count.saturating_add(1);
self.broadcast_status_change(&agent.agent_did, SeedAgentStatus::Paused);
continue;
};
if !charter.enabled || is_charter_past_sunset(&charter, now) {
if let Err(e) = self
.manager
.set_agent_status(&agent.agent_did, SeedAgentStatus::Paused)
{
warn!(
agent = %agent.agent_did,
error = %e,
"SeedAgentDaemon: failed to pause sunsetted agent"
);
continue;
}
outcome.paused_count = outcome.paused_count.saturating_add(1);
self.broadcast_status_change(&agent.agent_did, SeedAgentStatus::Paused);
}
}
for agent in self.manager.list_agents(None) {
if !matches!(agent.status, SeedAgentStatus::Active) {
continue;
}
let Some(charter) = self.manager.get_charter(&agent.charter_id) else {
continue;
};
if !charter.enabled || is_charter_past_sunset(&charter, now) {
continue;
}
let last_ms = agent.last_active.as_millis();
let now_ms = now.as_millis();
if now_ms.saturating_sub(last_ms) < self.config.min_refill_interval_ms {
continue;
}
let requested = charter.spend_caps.monthly_cap_wei;
if requested == 0 {
continue;
}
match self.manager.refill_agent_monthly(&agent.agent_did, requested, now) {
Ok(result) if result.granted_wei > 0 => {
outcome.refilled_count = outcome.refilled_count.saturating_add(1);
outcome.total_granted_wei = outcome
.total_granted_wei
.saturating_add(result.granted_wei);
self.broadcast_refill_completed(
&agent.agent_did,
result.granted_wei,
result.month,
);
}
Ok(_) => {
debug!(
agent = %agent.agent_did,
"SeedAgentDaemon: refill returned 0 granted (sunset/exhausted)"
);
}
Err(e) => {
warn!(
agent = %agent.agent_did,
error = %e,
"SeedAgentDaemon: refill failed"
);
}
}
}
match self.manager.sunset_wind_down_sweep(now, self.config.quarantine_grace_ms) {
Ok(report) => {
outcome.quarantined_count = report.quarantined.len() as u32;
outcome.terminated_count = report.terminated.len() as u32;
for agent_did in &report.quarantined {
self.broadcast_status_change(agent_did, SeedAgentStatus::Quarantined);
}
for agent_did in &report.terminated {
self.broadcast_status_change(agent_did, SeedAgentStatus::Terminated);
}
if let Some(disposition) = report.surplus {
outcome.surplus_disposed_wei = disposition.total_wei;
info!(
total = %disposition.total_wei,
burn = %disposition.burn_wei,
treasury = %disposition.treasury_wei,
"SeedAgentDaemon: surplus disposed at sunset"
);
if let Some(tx) = &self.gossip_tx {
let earmark = self.manager.earmark();
let msg = SeedAgentGossipMessage::EarmarkUpdated(earmark);
let _ = tx.send(msg);
}
if let Some(cb) = &self.surplus_disposition {
if let Err(e) = cb(&disposition) {
warn!(
error = %e,
"SeedAgentDaemon: surplus disposition callback failed"
);
}
}
}
}
Err(e) => {
warn!(error = %e, "SeedAgentDaemon: sunset_wind_down_sweep failed");
}
}
*self.last_outcome.lock() = Some(outcome.clone());
Ok(outcome)
}
fn broadcast_refill_completed(
&self,
agent_did: &str,
granted_wei: u128,
month: u8,
) {
let Some(tx) = &self.gossip_tx else { return };
let snapshot = self.manager.earmark();
let msg = SeedAgentGossipMessage::MonthlyRefillCompleted {
agent_did: agent_did.to_string(),
granted_wei,
month,
earmark_snapshot: snapshot,
};
if tx.send(msg).is_err() {
debug!("SeedAgentDaemon: gossip channel closed, dropping refill broadcast");
}
}
fn broadcast_status_change(&self, agent_did: &str, status: SeedAgentStatus) {
let Some(tx) = &self.gossip_tx else { return };
let msg = SeedAgentGossipMessage::AgentStatusChanged {
agent_did: agent_did.to_string(),
status,
};
if tx.send(msg).is_err() {
debug!("SeedAgentDaemon: gossip channel closed, dropping status broadcast");
}
}
}
fn is_charter_past_sunset(
charter: &crate::seed_agent::Charter,
now: Timestamp,
) -> bool {
let sunset_ms = charter.sunset.as_millis();
sunset_ms != 0 && now.as_millis() >= sunset_ms
}
impl SeedAgentDaemonConfig {
pub fn validate(&self) -> Result<()> {
if self.poll_interval_secs == 0 {
return Err(TokenError::InvalidParameter(
"SeedAgentDaemonConfig.poll_interval_secs == 0".into(),
));
}
if self.min_refill_interval_ms <= 0 {
return Err(TokenError::InvalidParameter(
"SeedAgentDaemonConfig.min_refill_interval_ms <= 0".into(),
));
}
if self.quarantine_grace_ms <= 0 {
return Err(TokenError::InvalidParameter(
"SeedAgentDaemonConfig.quarantine_grace_ms <= 0".into(),
));
}
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::seed_agent::{
Charter, CounterpartyFilter, DecaySchedule, OperationKind,
SeedAgentRecord, SpendCaps, TreasuryEarmark, DEFAULT_SURPLUS_BURN_BPS,
};
use tenzro_types::primitives::Hash;
use tokio::sync::mpsc;
fn h(byte: u8) -> Hash {
Hash::new([byte; 32])
}
fn seeded_manager(now: Timestamp) -> Arc<SeedAgentEarmarkManager> {
let mgr = Arc::new(SeedAgentEarmarkManager::new());
let earmark = TreasuryEarmark {
name: "SeedAgent".into(),
initial_allocation_wei: 10_000,
allocation_remaining_wei: 10_000,
total_drawn_wei: 0,
bootstrap_start: now,
bootstrap_end: Timestamp::new(now.as_millis() + MONTH_MILLIS * 12),
decay_schedule: DecaySchedule::default_with_base(100),
seed_agent_count: 0,
charter_ids: Vec::new(),
enabled: true,
surplus_burn_bps: DEFAULT_SURPLUS_BURN_BPS,
current_month: 0,
month_drawn_wei: 0,
};
mgr.apply_earmark(earmark).unwrap();
let charter = Charter {
charter_id: h(0xC1),
name: "InferenceLoad".into(),
purpose: "test".into(),
operations: vec![OperationKind::InferenceConsumer],
spend_caps: SpendCaps {
daily_cap_wei: 50,
monthly_cap_wei: 100,
per_tx_cap_wei: 10,
},
target_throughput: None,
counterparty_filter: CounterpartyFilter::default(),
sunset: Timestamp::new(now.as_millis() + MONTH_MILLIS * 12),
enabled: true,
};
mgr.upsert_charter(charter).unwrap();
mgr.register_agent(SeedAgentRecord::new(
"did:tenzro:machine:seed:a".into(),
"did:tenzro:org:treasury:seedagents".into(),
h(0xC1),
now,
))
.unwrap();
mgr
}
#[test]
fn tick_skips_when_master_switch_off() {
let now = Timestamp::new(0);
let mgr = seeded_manager(now);
let mut earmark = mgr.earmark();
earmark.enabled = false;
mgr.apply_earmark(earmark).unwrap();
let daemon = SeedAgentDaemon::new(mgr.clone());
let later = Timestamp::new(MONTH_MILLIS + 1);
let outcome = daemon.tick_once(later).unwrap();
assert!(outcome.master_switch_skipped);
assert_eq!(outcome.refilled_count, 0);
assert_eq!(mgr.earmark().allocation_remaining_wei, 10_000);
}
#[test]
fn tick_skips_when_authority_returns_false() {
let now = Timestamp::new(0);
let mgr = seeded_manager(now);
let daemon = SeedAgentDaemon::new(mgr.clone())
.with_tick_authority(Arc::new(|| false));
let later = Timestamp::new(MONTH_MILLIS + 1);
let outcome = daemon.tick_once(later).unwrap();
assert!(outcome.authority_skipped);
assert_eq!(outcome.refilled_count, 0);
}
#[test]
fn tick_refills_eligible_agent_and_broadcasts() {
let now = Timestamp::new(0);
let mgr = seeded_manager(now);
let (tx, mut rx) = mpsc::unbounded_channel();
let daemon = SeedAgentDaemon::new(mgr.clone()).with_gossip(tx);
let later = Timestamp::new(MONTH_MILLIS + 1);
let outcome = daemon.tick_once(later).unwrap();
assert_eq!(outcome.refilled_count, 1);
assert_eq!(outcome.total_granted_wei, 100);
let msg = rx.try_recv().expect("missing refill broadcast");
match msg {
SeedAgentGossipMessage::MonthlyRefillCompleted {
granted_wei,
month,
earmark_snapshot,
..
} => {
assert_eq!(granted_wei, 100);
assert_eq!(month, 1);
assert_eq!(earmark_snapshot.allocation_remaining_wei, 9_900);
assert_eq!(earmark_snapshot.total_drawn_wei, 100);
}
other => panic!("unexpected gossip: {:?}", other),
}
}
#[test]
fn tick_skips_recently_active_agent() {
let now = Timestamp::new(0);
let mgr = seeded_manager(now);
let daemon = SeedAgentDaemon::new(mgr.clone());
let too_soon = Timestamp::new(MONTH_MILLIS - 1);
let outcome = daemon.tick_once(too_soon).unwrap();
assert_eq!(outcome.refilled_count, 0);
assert_eq!(mgr.earmark().total_drawn_wei, 0);
}
#[test]
fn tick_pauses_agent_under_disabled_charter() {
let now = Timestamp::new(0);
let mgr = seeded_manager(now);
let mut charter = mgr.get_charter(&h(0xC1)).unwrap();
charter.enabled = false;
mgr.upsert_charter(charter).unwrap();
let (tx, mut rx) = mpsc::unbounded_channel();
let daemon = SeedAgentDaemon::new(mgr.clone()).with_gossip(tx);
let later = Timestamp::new(MONTH_MILLIS + 1);
let outcome = daemon.tick_once(later).unwrap();
assert_eq!(outcome.paused_count, 1);
assert_eq!(outcome.refilled_count, 0);
assert_eq!(outcome.quarantined_count, 1);
let updated = mgr.get_agent("did:tenzro:machine:seed:a").unwrap();
assert_eq!(updated.status, SeedAgentStatus::Quarantined);
let mut saw_paused = false;
let mut saw_quarantined = false;
while let Ok(msg) = rx.try_recv() {
if let SeedAgentGossipMessage::AgentStatusChanged { agent_did, status } = msg {
assert_eq!(agent_did, "did:tenzro:machine:seed:a");
match status {
SeedAgentStatus::Paused => saw_paused = true,
SeedAgentStatus::Quarantined => saw_quarantined = true,
_ => {}
}
}
}
assert!(saw_paused, "missing Paused broadcast");
assert!(saw_quarantined, "missing Quarantined broadcast");
}
#[test]
fn tick_pauses_agent_past_sunset() {
let now = Timestamp::new(0);
let mgr = seeded_manager(now);
let mut charter = mgr.get_charter(&h(0xC1)).unwrap();
charter.sunset = Timestamp::new(MONTH_MILLIS);
mgr.upsert_charter(charter).unwrap();
let daemon = SeedAgentDaemon::new(mgr.clone());
let later = Timestamp::new(MONTH_MILLIS * 2);
let outcome = daemon.tick_once(later).unwrap();
assert_eq!(outcome.paused_count, 1);
assert_eq!(outcome.quarantined_count, 1);
let updated = mgr.get_agent("did:tenzro:machine:seed:a").unwrap();
assert_eq!(updated.status, SeedAgentStatus::Quarantined);
}
#[test]
fn tick_no_op_for_paused_or_terminated_agents() {
let now = Timestamp::new(0);
let mgr = seeded_manager(now);
mgr.set_agent_status("did:tenzro:machine:seed:a", SeedAgentStatus::Paused)
.unwrap();
let daemon = SeedAgentDaemon::new(mgr.clone());
let later = Timestamp::new(MONTH_MILLIS * 2);
let outcome = daemon.tick_once(later).unwrap();
assert_eq!(outcome.refilled_count, 0);
assert_eq!(outcome.paused_count, 0);
}
#[test]
fn config_validate_rejects_zero_poll() {
let cfg = SeedAgentDaemonConfig {
poll_interval_secs: 0,
min_refill_interval_ms: MONTH_MILLIS,
quarantine_grace_ms: DEFAULT_DAEMON_QUARANTINE_GRACE_MS,
};
assert!(cfg.validate().is_err());
}
#[test]
fn config_validate_rejects_non_positive_grace() {
let cfg = SeedAgentDaemonConfig {
poll_interval_secs: DEFAULT_DAEMON_POLL_INTERVAL_SECS,
min_refill_interval_ms: MONTH_MILLIS,
quarantine_grace_ms: 0,
};
assert!(cfg.validate().is_err());
}
#[test]
fn tick_runs_sunset_sweep_and_broadcasts_quarantined() {
let now = Timestamp::new(0);
let mgr = seeded_manager(now);
mgr.set_agent_status("did:tenzro:machine:seed:a", SeedAgentStatus::Paused)
.unwrap();
let mut charter = mgr.get_charter(&h(0xC1)).unwrap();
charter.enabled = false;
mgr.upsert_charter(charter).unwrap();
let (tx, mut rx) = mpsc::unbounded_channel();
let daemon = SeedAgentDaemon::new(mgr.clone()).with_gossip(tx);
let later = Timestamp::new(MONTH_MILLIS + 1);
let outcome = daemon.tick_once(later).unwrap();
assert_eq!(outcome.quarantined_count, 1);
assert_eq!(outcome.terminated_count, 0);
let mut saw_quarantine = false;
while let Ok(msg) = rx.try_recv() {
if let SeedAgentGossipMessage::AgentStatusChanged { status, .. } = msg {
if status == SeedAgentStatus::Quarantined {
saw_quarantine = true;
break;
}
}
}
assert!(saw_quarantine, "expected Quarantined status broadcast");
}
#[test]
fn tick_runs_sunset_sweep_and_invokes_surplus_callback() {
use std::sync::atomic::{AtomicU64, Ordering};
let now = Timestamp::new(0);
let mgr = seeded_manager(now);
let mut charter = mgr.get_charter(&h(0xC1)).unwrap();
charter.sunset = Timestamp::new(1);
mgr.upsert_charter(charter).unwrap();
mgr.set_agent_status("did:tenzro:machine:seed:a", SeedAgentStatus::Terminated)
.unwrap();
let burned = Arc::new(AtomicU64::new(0));
let burned_cb = burned.clone();
let cb: SurplusDispositionFn = Arc::new(move |d| {
burned_cb.store(d.burn_wei as u64, Ordering::SeqCst);
Ok(())
});
let daemon = SeedAgentDaemon::new(mgr.clone()).with_surplus_disposition(cb);
let later = Timestamp::new(MONTH_MILLIS + 1);
let outcome = daemon.tick_once(later).unwrap();
assert_eq!(outcome.surplus_disposed_wei, 10_000);
assert_eq!(burned.load(Ordering::SeqCst), 5_000);
assert_eq!(mgr.earmark().allocation_remaining_wei, 0);
assert!(!mgr.earmark().enabled);
}
#[test]
fn last_outcome_is_stored_after_tick() {
let now = Timestamp::new(0);
let mgr = seeded_manager(now);
let daemon = SeedAgentDaemon::new(mgr.clone());
assert!(daemon.last_outcome().is_none());
let later = Timestamp::new(MONTH_MILLIS + 1);
let _ = daemon.tick_once(later).unwrap();
let snap = daemon.last_outcome().unwrap();
assert_eq!(snap.refilled_count, 1);
}
}