feat: add supervisor auto-dispatch loop (item #408)

crates/agentflare-backend/src/item.rs

@@ -233,6 +233,18 @@ pub fn list_by_project(conn: &Connection, project_id: &str) -> Result<Vec<Item>>
     Ok(rows.collect::<std::result::Result<_, _>>()?)
 }
 
+pub fn list_by_label(conn: &Connection, project_id: &str, label_id: &str) -> Result<Vec<Item>> {
+    let mut stmt = conn.prepare(
+        "SELECT items.id, items.project_id, items.state_id, items.name, items.description, items.priority, items.parent_id, items.assignee_agent, items.sequence_id, items.sort_order, items.started_at, items.completed_at, items.archived_at, items.external_source, items.external_id, items.metadata, items.created_at, items.updated_at, items.deleted_at
+         FROM items
+         INNER JOIN item_labels ON item_labels.item_id = items.id
+         WHERE item_labels.label_id = ?1 AND items.project_id = ?2 AND items.deleted_at IS NULL
+         ORDER BY items.sort_order",
+    )?;
+    let rows = stmt.query_map(rusqlite::params![label_id, project_id], row_to_item)?;
+    Ok(rows.collect::<std::result::Result<_, _>>()?)
+}
+
 /// List non-deleted items assigned to an agent (excludes completed/cancelled).
 pub fn list_by_assignee_agent(
     conn: &Connection,
@@ -1900,4 +1912,71 @@ mod tests {
         .unwrap();
         assert_eq!(updated.assignee_agent.as_deref(), Some("claude-code"));
     }
+
+    #[test]
+    fn list_by_label_returns_only_items_carrying_that_label() {
+        let conn = db::open_in_memory().unwrap();
+        let (pid, sid) = seed_project(&conn, "label");
+        let ws_id = crate::project::get(&conn, &pid).unwrap().workspace_id;
+        let label = crate::label::create(
+            &conn,
+            crate::label::CreateLabel {
+                project_id: Some(pid.clone()),
+                workspace_id: ws_id,
+                name: "ready-for-work".into(),
+                color: None,
+                parent_id: None,
+                sort_order: None,
+                external_source: None,
+                external_id: None,
+            },
+        )
+        .unwrap();
+
+        let labeled = create(
+            &conn,
+            CreateItem {
+                project_id: pid.clone(),
+                state_id: sid.clone(),
+                name: "Labeled".into(),
+                description: None,
+                priority: None,
+                parent_id: None,
+                assignee_agent: None,
+                sort_order: None,
+                external_source: None,
+                external_id: None,
+                metadata: None,
+                label_ids: vec![],
+                assignee_ids: vec![],
+                dependency_ids: vec![],
+            },
+        )
+        .unwrap();
+        create(
+            &conn,
+            CreateItem {
+                project_id: pid.clone(),
+                state_id: sid,
+                name: "Unlabeled".into(),
+                description: None,
+                priority: None,
+                parent_id: None,
+                assignee_agent: None,
+                sort_order: None,
+                external_source: None,
+                external_id: None,
+                metadata: None,
+                label_ids: vec![],
+                assignee_ids: vec![],
+                dependency_ids: vec![],
+            },
+        )
+        .unwrap();
+        add_label(&conn, &labeled.id, &label.id).unwrap();
+
+        let found = list_by_label(&conn, &pid, &label.id).unwrap();
+        assert_eq!(found.len(), 1);
+        assert_eq!(found[0].id, labeled.id);
+    }
 }

src/dashboard/server.rs

@@ -142,6 +142,7 @@ const COST_REFRESH: std::time::Duration = std::time::Duration::from_secs(30);
 /// nothing else ever calls `Queue::cleanup`.
 const JOB_CLEANUP_INTERVAL: std::time::Duration = std::time::Duration::from_secs(3600);
 const JOB_RETENTION_SECS: i64 = 7 * 24 * 3600;
+const SUPERVISOR_DISCOVERY_INTERVAL: std::time::Duration = std::time::Duration::from_secs(12);
 
 /// Runs for the lifetime of the process (like `snapshot_broadcaster`'s
 /// producer task): wakes on `JOB_CLEANUP_INTERVAL`, deletes finished jobs
@@ -165,6 +166,39 @@ fn spawn_job_cleanup(queue: agentflare_jobs::Queue) {
     });
 }
 
+/// Runs for the lifetime of the process: wakes on `SUPERVISOR_DISCOVERY_INTERVAL`,
+/// lists items labeled `ready-for-work` and dispatches an `agentflare work` job
+/// for each confirmed-autonomous assignee. The blocking SQLite work happens off
+/// the async worker threads via `spawn_blocking`, same as `spawn_job_cleanup`.
+fn spawn_supervisor_discovery(
+    queue: agentflare_jobs::Queue,
+    mcp: std::sync::Arc<crate::mcp_server::AgentflareMcp>,
+    interval: std::time::Duration,
+) {
+    tokio::spawn(async move {
+        let mut ticker = tokio::time::interval(interval);
+        loop {
+            ticker.tick().await;
+            let queue = queue.clone();
+            let mcp = mcp.clone();
+            let result = tokio::task::spawn_blocking(move || {
+                crate::supervisor::run_discovery_tick(&mcp, &queue)
+            })
+            .await;
+            match result {
+                Ok(summary) if summary.dispatched > 0 || summary.skipped > 0 => {
+                    eprintln!(
+                        "agentflare-supervisor: dispatched {}, skipped {}",
+                        summary.dispatched, summary.skipped
+                    );
+                }
+                Ok(_) => {}
+                Err(e) => eprintln!("agentflare-supervisor: tick task panicked: {e}"),
+            }
+        }
+    });
+}
+
 /// Single shared broadcast of the live `{ claims, cost_today }` snapshot. Every
 /// `/events` client subscribes to this one channel, so there is no per-client
 /// work — the producer below runs at most one refresh cycle per `TICK`,
@@ -557,6 +591,11 @@ pub async fn run(host: &str, port: u16, open: bool, yes_expose: bool) {
     let mut worker_pool = agentflare_jobs::WorkerPool::new(queue.clone());
     worker_pool.start(2);
     spawn_job_cleanup(queue.clone());
+    spawn_supervisor_discovery(
+        queue.clone(),
+        std::sync::Arc::new(crate::mcp_server::AgentflareMcp::default()),
+        SUPERVISOR_DISCOVERY_INTERVAL,
+    );
 
     let listener = tokio::net::TcpListener::bind((host, port))
         .await
@@ -1044,4 +1083,14 @@ mod tests {
             "expected dashboard to serve on a local bind without --yes-expose"
         );
     }
+
+    #[tokio::test]
+    async fn spawn_supervisor_discovery_runs_without_panicking_on_an_empty_project() {
+        let dir = tempfile::tempdir().unwrap().keep();
+        let queue = agentflare_jobs::Queue::open_memory(dir.join("logs")).unwrap();
+        let mcp = std::sync::Arc::new(crate::mcp_server::AgentflareMcp::for_test_memory());
+
+        spawn_supervisor_discovery(queue, mcp, std::time::Duration::from_millis(20));
+        tokio::time::sleep(std::time::Duration::from_millis(80)).await;
+    }
 }

src/main.rs

@@ -57,6 +57,7 @@ mod skill_detect;
 mod skill_proactive;
 mod state;
 mod store;
+mod supervisor;
 mod tool_install;
 mod ui;
 mod uninstall;

src/mcp_server.rs

@@ -82,7 +82,7 @@ pub struct AgentflareMcp {
     /// its own source of truth, so nothing to refresh.
     backend_db: std::sync::Mutex<Option<rusqlite::Connection>>,
     /// Tests inject a temp path here so they never touch the shared backend.db.
-    backend_db_override: Option<std::path::PathBuf>,
+    pub(crate) backend_db_override: Option<std::path::PathBuf>,
     /// Tests inject a temp file path here so project-link resolution never
     /// reads/writes this actual repo's `.agentflare/project.json`.
     backend_project_link_override: Option<std::path::PathBuf>,
@@ -580,6 +580,17 @@ impl AgentflareMcp {
         }
     }
 
+    /// Create an in-memory backend for tests that don't need a real repo/disk.
+    /// The other fields (skills, gateway, store, etc.) stay defaulted so they
+    /// lazily open `:memory:` SQLite connections or no-ops — no I/O to ~/.
+    #[cfg(test)]
+    pub(crate) fn for_test_memory() -> Self {
+        Self {
+            backend_db_override: Some(std::path::PathBuf::from(":memory:")),
+            ..Default::default()
+        }
+    }
+
     /// Pure walk-up so the non-git fallback path is unit-testable without
     /// touching process-global state: neither this process's real cwd nor
     /// `crate::paths::home()` (which itself reads the `AGENTFLARE_HOME_OVERRIDE`

src/mcp_server/item.rs

@@ -610,7 +610,7 @@ impl AgentflareMcp {
         })?
     }
 
-    pub(super) fn item_add_label(&self, req: ItemRequest) -> Result<String, ErrorData> {
+    pub(crate) fn item_add_label(&self, req: ItemRequest) -> Result<String, ErrorData> {
         let raw = req
             .id
             .ok_or_else(|| ErrorData::invalid_params("id is required for add_label", None))?;
@@ -634,7 +634,7 @@ impl AgentflareMcp {
         })?
     }
 
-    pub(super) fn item_remove_label(&self, req: ItemRequest) -> Result<String, ErrorData> {
+    pub(crate) fn item_remove_label(&self, req: ItemRequest) -> Result<String, ErrorData> {
         let raw = req
             .id
             .ok_or_else(|| ErrorData::invalid_params("id is required for remove_label", None))?;