diff --git a/crates/colibri-daemon/src/scheduler.rs b/crates/colibri-daemon/src/scheduler.rs index 7364a2b..6a264c9 100644 --- a/crates/colibri-daemon/src/scheduler.rs +++ b/crates/colibri-daemon/src/scheduler.rs @@ -98,6 +98,8 @@ pub struct ScheduledJob { /// A task submitted from an external intake (Telegram, socket, cron). #[derive(Debug, Clone, Serialize, Deserialize)] pub struct TaskRequest { + /// Pre-created task ID from the caller (task is already in the store). + pub task_id: String, pub title: String, pub description: Option, pub required_capabilities: Vec, @@ -204,32 +206,29 @@ impl Scheduler { } // ── Process intake queue ────────────────────────────────── + // Tasks are already in the store (created by + // cmd_intake_task); tick only routes them to capable agents. let intake: Vec = self.intake_queue.drain(..).collect(); for req in intake { - let create_result = { + let agents = { let store = state.store.lock().unwrap(); - store.create_task(&req.title, req.description.as_deref()) + store.list_agents().unwrap_or_default() }; - match create_result { - Ok(task) => { - let agents = { - let store = state.store.lock().unwrap(); - store.list_agents().unwrap_or_default() - }; - if let Some(agent) = pick_agent(&req.required_capabilities, &agents) { - let _ = { - let store = state.store.lock().unwrap(); - store.claim_task(&task.id, &agent.id) - }; - info!( - task_id = %task.id, - agent_id = %agent.id, - "scheduler: assigned intake task to agent" - ); - } - } - Err(e) => { - warn!(error = %e, "scheduler: failed to create task from intake"); + if let Some(agent) = pick_agent(&req.required_capabilities, &agents) { + match { + let store = state.store.lock().unwrap(); + store.claim_task(&req.task_id, &agent.id) + } { + Ok(_) => info!( + task_id = %req.task_id, + agent_id = %agent.id, + "scheduler: assigned intake task to agent" + ), + Err(e) => warn!( + task_id = %req.task_id, + error = %e, + "scheduler: failed to claim intake task" + ), } } } @@ -509,8 +508,8 @@ mod tests { capabilities: serde_json::json!(["rust"]), status: "offline".into(), created_at: "2026-01-01T00:00:00Z".into(), - host: None, - last_seen: None, + host: None, + last_seen: None, }]; assert!(pick_agent(&required, &agents).is_none()); } @@ -524,8 +523,8 @@ mod tests { capabilities: serde_json::json!(["linux"]), status: "idle".into(), created_at: "2026-01-01T00:00:00Z".into(), - host: None, - last_seen: None, + host: None, + last_seen: None, }]; assert!(pick_agent(&required, &agents).is_none()); } @@ -538,8 +537,8 @@ mod tests { capabilities: serde_json::json!([]), status: "idle".into(), created_at: "2026-01-01T00:00:00Z".into(), - host: None, - last_seen: None, + host: None, + last_seen: None, }]; assert_eq!(pick_agent(&[], &agents).unwrap().name, "generalist"); } @@ -613,8 +612,8 @@ mod tests { capabilities: serde_json::json!(["rust"]), status: "active".into(), created_at: "2026-01-01T00:00:00Z".into(), - host: None, - last_seen: None, + host: None, + last_seen: None, }]; let picked = pick_agent(&required, &agents); assert!( @@ -630,7 +629,16 @@ mod tests { let data_dir = config.data_dir.clone(); let state: SharedState = Arc::new(crate::daemon::DaemonState::new(config)); let mut scheduler = Scheduler::new(); + // Pre-create the task so the tick can claim it (the tick no longer + // creates tasks — that happens in cmd_intake_task now). + let task = state + .store + .lock() + .unwrap() + .create_task("intake check", Some("must become a task")) + .unwrap(); scheduler.submit(TaskRequest { + task_id: task.id, title: "intake check".to_string(), description: Some("must become a task".to_string()), required_capabilities: Vec::new(), diff --git a/crates/colibri-daemon/src/socket.rs b/crates/colibri-daemon/src/socket.rs index 312c2a6..5e39ed3 100644 --- a/crates/colibri-daemon/src/socket.rs +++ b/crates/colibri-daemon/src/socket.rs @@ -264,9 +264,11 @@ async fn dispatch(cmd: ColibriCommand, state: &SharedState) -> ColibriResponse { cmd_claim_task(state, task_id, agent_id).await } ColibriCommand::ListAgents => cmd_list_agents(state).await, - ColibriCommand::RegisterAgent { name, capabilities, host } => { - cmd_register_agent(state, name, capabilities, host.as_deref()).await - } + ColibriCommand::RegisterAgent { + name, + capabilities, + host, + } => cmd_register_agent(state, name, capabilities, host.as_deref()).await, ColibriCommand::Heartbeat { agent_id, host } => { cmd_heartbeat(state, agent_id, host.as_deref()).await } @@ -1083,13 +1085,19 @@ async fn cmd_intake_task( ) -> ColibriResponse { let caps = capabilities.unwrap_or_default(); // Create the task immediately so the caller gets the ID back. - let task = match state.store.lock().unwrap().create_task(&title, description.as_deref()) { + let task = match state + .store + .lock() + .unwrap() + .create_task(&title, description.as_deref()) + { Ok(t) => t, Err(e) => return ColibriResponse::err(format!("create task failed: {e}")), }; // Also queue for the scheduler tick to route to a capable agent. let mut scheduler = state.scheduler.lock().await; scheduler.submit(crate::scheduler::TaskRequest { + task_id: task.id.clone(), title, description, required_capabilities: caps, diff --git a/crates/colibri-daemon/tests/multi_agent_board.rs b/crates/colibri-daemon/tests/multi_agent_board.rs index ef0ea31..98069f5 100644 --- a/crates/colibri-daemon/tests/multi_agent_board.rs +++ b/crates/colibri-daemon/tests/multi_agent_board.rs @@ -199,7 +199,10 @@ async fn scheduler_routes_intake_tasks_by_capability() { ) .await; // Intake now returns the full task (id + status), not just {"status":"queued"}. - assert!(fs_intake["data"]["id"].is_string(), "intake must return task id"); + assert!( + fs_intake["data"]["id"].is_string(), + "intake must return task id" + ); send_command( &socket_path, diff --git a/crates/colibri-store/src/lib.rs b/crates/colibri-store/src/lib.rs index ea0fd87..12afb5d 100644 --- a/crates/colibri-store/src/lib.rs +++ b/crates/colibri-store/src/lib.rs @@ -375,9 +375,7 @@ impl Store { params![now, host, agent_id], )?; if rows == 0 { - return Err(StoreError::NotFound(format!( - "agent not found: {agent_id}" - ))); + return Err(StoreError::NotFound(format!("agent not found: {agent_id}"))); } Ok(()) }