P3 backend: files, skills, routines, claw chat with LIVE taint plumbing

- 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]>
2026-06-10 05:28:50 -05:00
parent ea5162ac65
commit 67f918439c
69 changed files with 3651 additions and 183 deletions
@@ -0,0 +1,204 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use serde_json::json;
+use tc_domain::{
+    AccessPolicy, Agent, AgentId, AgentStatus, MessageRole, Role, User, UserId, Workspace,
+    WorkspaceId,
+};
+use tc_llm::ScriptedProvider;
+use tc_runtime::{Runtime, RuntimeConfig};
+use tc_scheduler::{next_occurrence, Scheduler};
+use time::macros::datetime;
+
+#[test]
+fn next_occurrence_follows_the_cron_pattern() {
+    let after = datetime!(2026-06-10 08:30:00 UTC);
+    // Daily at 09:00.
+    assert_eq!(
+        next_occurrence("0 9 * * *", after).unwrap(),
+        datetime!(2026-06-10 09:00:00 UTC)
+    );
+    // Already past 09:00 today → tomorrow.
+    let late = datetime!(2026-06-10 09:30:00 UTC);
+    assert_eq!(
+        next_occurrence("0 9 * * *", late).unwrap(),
+        datetime!(2026-06-11 09:00:00 UTC)
+    );
+    // Every minute.
+    assert_eq!(
+        next_occurrence("* * * * *", after).unwrap(),
+        datetime!(2026-06-10 08:31:00 UTC)
+    );
+    // Mondays only (2026-06-10 is a Wednesday).
+    assert_eq!(
+        next_occurrence("0 9 * * MON", after).unwrap(),
+        datetime!(2026-06-15 09:00:00 UTC)
+    );
+}
+
+#[test]
+fn invalid_cron_patterns_are_errors() {
+    let after = datetime!(2026-06-10 08:30:00 UTC);
+    assert!(next_occurrence("not a cron", after).is_err());
+    assert!(next_occurrence("99 99 * * *", after).is_err());
+}
+
+async fn seeded(pool: &sqlx::PgPool) -> Agent {
+    let workspace = Workspace {
+        id: WorkspaceId::new(),
+        name: "Acme".into(),
+        plan: "team".into(),
+    };
+    tc_db::repo::workspaces::insert(pool, &workspace)
+        .await
+        .unwrap();
+    let owner = User {
+        id: UserId::new(),
+        workspace_id: workspace.id,
+        email: format!("{}@acme.test", UserId::new()),
+        role: Role::Owner,
+        display_name: "Owner".into(),
+        created_at: time::OffsetDateTime::UNIX_EPOCH,
+    };
+    tc_db::repo::users::insert(pool, &owner).await.unwrap();
+    let agent = Agent {
+        id: AgentId::new(),
+        workspace_id: workspace.id,
+        name: "Scout".into(),
+        job_title: "Analyst".into(),
+        system_prompt: String::new(),
+        avatar: String::new(),
+        accent: String::new(),
+        wallpaper: String::new(),
+        managed_by: owner.id,
+        status: AgentStatus::Online,
+    };
+    tc_db::repo::agents::insert(pool, &agent, &AccessPolicy::default())
+        .await
+        .unwrap();
+    agent
+}
+
+#[tokio::test]
+async fn due_routines_fire_real_runs_exactly_once() {
+    let pool = tc_testkit::test_pool().await;
+    let agent = seeded(&pool).await;
+    let runtime = Runtime::new(
+        pool.clone(),
+        Arc::new(ScriptedProvider::from_toml("").unwrap()),
+        RuntimeConfig {
+            model: "scripted".into(),
+            max_tokens: 1024,
+        },
+    );
+    let scheduler = Scheduler::new(pool.clone(), runtime);
+
+    // A routine that became due a minute ago.
+    let now = time::OffsetDateTime::now_utc();
+    tc_db::repo::routines::create(
+        &pool,
+        agent.id,
+        "Morning digest",
+        "* * * * *",
+        json!({"message": "compile the digest"}),
+        now - time::Duration::minutes(1),
+    )
+    .await
+    .unwrap();
+
+    let fired = scheduler.tick(now).await.unwrap();
+    assert_eq!(fired, 1);
+    // Claimed: an immediate second tick fires nothing.
+    assert_eq!(scheduler.tick(now).await.unwrap(), 0);
+
+    // The routine's clock advanced beyond now.
+    let routines = tc_db::repo::routines::list_by_agent(&pool, agent.id)
+        .await
+        .unwrap();
+    assert!(routines[0].next_run_at.unwrap() > now);
+    assert!(routines[0].last_run_at.is_some());
+
+    // The firing produced a REAL run in the routine's dedicated session.
+    let mut found = false;
+    for _ in 0..100 {
+        let sessions = tc_db::repo::sessions::list_by_agent(&pool, agent.id)
+            .await
+            .unwrap();
+        if let Some(session) = sessions.iter().find(|s| s.title == "⏰ Morning digest") {
+            let history = tc_db::repo::messages::history(&pool, session.id)
+                .await
+                .unwrap();
+            if history.len() == 2
+                && history[0].message.role == MessageRole::User
+                && history[1].message.content["text"]
+                    .as_str()
+                    .unwrap_or_default()
+                    .contains("compile the digest")
+            {
+                found = true;
+                break;
+            }
+        }
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+    assert!(found, "routine run never landed in its session");
+
+    // Re-firing later reuses the same session instead of spamming new ones.
+    tc_db::repo::routines::set_next_run(&pool, routines[0].id, Some(now))
+        .await
+        .unwrap();
+    scheduler.tick(now).await.unwrap();
+    for _ in 0..100 {
+        let sessions = tc_db::repo::sessions::list_by_agent(&pool, agent.id)
+            .await
+            .unwrap();
+        let routine_sessions: Vec<_> = sessions
+            .iter()
+            .filter(|s| s.title == "⏰ Morning digest")
+            .collect();
+        assert_eq!(routine_sessions.len(), 1);
+        let history = tc_db::repo::messages::history(&pool, routine_sessions[0].id)
+            .await
+            .unwrap();
+        if history.len() == 4 {
+            return;
+        }
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+    panic!("second firing never landed");
+}
+
+#[tokio::test]
+async fn paused_routines_do_not_fire() {
+    let pool = tc_testkit::test_pool().await;
+    let agent = seeded(&pool).await;
+    let runtime = Runtime::new(
+        pool.clone(),
+        Arc::new(ScriptedProvider::from_toml("").unwrap()),
+        RuntimeConfig {
+            model: "scripted".into(),
+            max_tokens: 1024,
+        },
+    );
+    let scheduler = Scheduler::new(pool.clone(), runtime);
+
+    let now = time::OffsetDateTime::now_utc();
+    let routine = tc_db::repo::routines::create(
+        &pool,
+        agent.id,
+        "Paused digest",
+        "* * * * *",
+        json!({"message": "nope"}),
+        now - time::Duration::minutes(1),
+    )
+    .await
+    .unwrap();
+    sqlx::query("UPDATE routines SET status = 'paused' WHERE id = $1")
+        .bind(routine.id)
+        .execute(&pool)
+        .await
+        .unwrap();
+
+    assert_eq!(scheduler.tick(now).await.unwrap(), 0);
+}