Omar SobhandClaude Fable 5
67f918439c
- tc-files: BlobStore trait + LocalBlobStore (traversal-proof keys); wired through Runtime (config storage.data_dir in deployments) - File tools: files.write/files.list (workspace-internal) + files.delete (gated FileDeletion — tested: file survives pending, gone after approve); GET /api/openclaw/files + /api/shared-drive/files (drive/agent scoped) - Skills: catalog/library + idempotent install with counter, uninstall; GET /api/skills[?clawId=], POST install/uninstall - tc-scheduler: croner cron math (clock-controlled tests), SKIP LOCKED claim-and-advance firing REAL runs into dedicated '⏰ name' sessions (reused, exactly-once), paused routines skipped; routines API + agent tool routine.schedule; loop spawned in server - Claw chat: 1:1 threads, chat.send enforcing the target's Other-Claws policy, chat.inbox whose output carries inter_agent taint; the run loop now ACCUMULATES taint from tool outputs into LoopState, classifies with it, and stamps steps + approvals — a poisoned inbox followed by email.send produces an approval whose taint_sources says inter_agent - ScriptedProvider scenario selection now keys on the most recent marker (session history kept earlier markers alive) 132 Rust tests green. Co-Authored-By: Claude Fable 5 <[email protected]>
75 lines
2.7 KiB
Rust
75 lines
2.7 KiB
Rust
//! Routines (spec §7.6): agent-created scheduled tasks. Due routines are
|
|
//! claimed atomically (one firing even with replicas) and each firing
|
|
//! drives a REAL run through the runtime in the routine's dedicated
|
|
//! session — gated tools inside a routine still hit the approval queue.
|
|
|
|
use sqlx::PgPool;
|
|
use tc_db::repo::{agents, routines, sessions};
|
|
use tc_runtime::Runtime;
|
|
use time::OffsetDateTime;
|
|
|
|
pub use tc_runtime::scheduling::next_occurrence;
|
|
|
|
#[derive(Debug, thiserror::Error)]
|
|
pub enum ScheduleError {
|
|
#[error(transparent)]
|
|
Pattern(#[from] tc_runtime::scheduling::ScheduleError),
|
|
#[error(transparent)]
|
|
Db(#[from] tc_db::DbError),
|
|
}
|
|
|
|
pub struct Scheduler {
|
|
pool: PgPool,
|
|
runtime: Runtime,
|
|
}
|
|
|
|
impl Scheduler {
|
|
pub fn new(pool: PgPool, runtime: Runtime) -> Scheduler {
|
|
Scheduler { pool, runtime }
|
|
}
|
|
|
|
/// Fires every due routine once and reschedules it. Returns how many
|
|
/// fired. Time is a parameter so tests control the clock.
|
|
pub async fn tick(&self, now: OffsetDateTime) -> Result<usize, ScheduleError> {
|
|
let due = routines::claim_due(&self.pool, now).await?;
|
|
for routine in &due {
|
|
// Reschedule first: a firing failure must not stall the clock.
|
|
let next = next_occurrence(&routine.schedule_cron, now).ok();
|
|
routines::set_next_run(&self.pool, routine.id, next).await?;
|
|
|
|
let message = routine.action["message"].as_str().unwrap_or_default();
|
|
if message.is_empty() {
|
|
continue;
|
|
}
|
|
let agent_id = tc_domain::AgentId::from(routine.agent_id);
|
|
let Ok(agent) = agents::get(&self.pool, agent_id).await else {
|
|
continue; // deleted agent: routine is orphaned
|
|
};
|
|
|
|
// Each routine runs in one dedicated, recognizable session.
|
|
let title = format!("⏰ {}", routine.name);
|
|
let session = match sessions::list_by_agent(&self.pool, agent_id)
|
|
.await?
|
|
.into_iter()
|
|
.find(|s| s.title == title)
|
|
{
|
|
Some(existing) => existing,
|
|
None => sessions::create(&self.pool, agent_id, agent.workspace_id, &title).await?,
|
|
};
|
|
let _ = self.runtime.send_message(session.id, message).await;
|
|
}
|
|
Ok(due.len())
|
|
}
|
|
|
|
/// The production loop: ticks on an interval with the real clock.
|
|
pub fn spawn(self, interval: std::time::Duration) {
|
|
tokio::spawn(async move {
|
|
let mut tick = tokio::time::interval(interval);
|
|
loop {
|
|
tick.tick().await;
|
|
let _ = self.tick(OffsetDateTime::now_utc()).await;
|
|
}
|
|
});
|
|
}
|
|
}
|