fix(task-sources): sync and prune stale tasks (#3418)

src/openhuman/task_sources/ops.rs

@@ -18,7 +18,7 @@ use crate::rpc::RpcOutcome;
 use super::types::{
     FetchReason, FilterSpec, ProviderSlug, SourceTarget, TaskSource, TaskSourcePatch,
 };
-use super::{filter, pipeline, store};
+use super::{filter, pipeline, route, store};
 
 /// List all configured task sources.
 pub async fn list(config: &Config) -> Result<RpcOutcome<Vec<TaskSource>>, String> {
@@ -104,10 +104,20 @@ pub async fn update(
 
 /// Remove a source by id.
 pub async fn remove(config: &Config, id: &str) -> Result<RpcOutcome<Value>, String> {
+    let ingested = store::list_ingested_refs(config, id).map_err(|e| e.to_string())?;
+    let mut pruned = 0usize;
+    for item in ingested {
+        if let Some(card_id) = item.card_id.as_deref().filter(|id| !id.trim().is_empty()) {
+            route::remove_card(config, card_id)?;
+        }
+        if store::remove_ingested(config, id, &item.external_id).map_err(|e| e.to_string())? {
+            pruned += 1;
+        }
+    }
     store::remove_source(config, id).map_err(|e| e.to_string())?;
-    tracing::debug!(source_id = %id, "[task_sources:ops] removed");
+    tracing::debug!(source_id = %id, pruned, "[task_sources:ops] removed");
     Ok(RpcOutcome::new(
-        json!({ "id": id, "removed": true }),
+        json!({ "id": id, "removed": true, "pruned": pruned }),
         vec![],
     ))
 }
@@ -119,6 +129,26 @@ pub async fn fetch(config: &Config, id: &str) -> Result<RpcOutcome<super::FetchO
     Ok(RpcOutcome::new(outcome, vec![]))
 }
 
+/// Manually sync all enabled task sources now.
+pub async fn sync(config: &Config) -> Result<RpcOutcome<Vec<super::FetchOutcome>>, String> {
+    let sources = store::list_sources(config).map_err(|e| e.to_string())?;
+    let mut outcomes = Vec::new();
+    for source in sources.into_iter().filter(|source| source.enabled) {
+        outcomes.push(pipeline::run_source_once(config, &source, FetchReason::Manual).await);
+    }
+    tracing::info!(
+        source_count = outcomes.len(),
+        fetched = outcomes
+            .iter()
+            .map(|outcome| outcome.fetched)
+            .sum::<usize>(),
+        routed = outcomes.iter().map(|outcome| outcome.routed).sum::<usize>(),
+        pruned = outcomes.iter().map(|outcome| outcome.pruned).sum::<usize>(),
+        "[task_sources:ops] sync completed"
+    );
+    Ok(RpcOutcome::new(outcomes, vec![]))
+}
+
 /// Recently ingested tasks for a source (newest first).
 pub async fn list_tasks(
     config: &Config,

src/openhuman/task_sources/pipeline.rs

@@ -9,6 +9,7 @@
 use std::sync::Arc;
 
 use chrono::Utc;
+use std::collections::HashSet;
 
 use crate::core::event_bus::{publish_global, DomainEvent};
 use crate::openhuman::config::Config;
@@ -40,8 +41,8 @@ pub async fn run_source_once(
     match run_inner(config, source, reason, &mut outcome).await {
         Ok(()) => {
             let status = format!(
-                "fetched {} routed {} dupes {}",
-                outcome.fetched, outcome.routed, outcome.skipped_dupe
+                "fetched {} routed {} dupes {} pruned {}",
+                outcome.fetched, outcome.routed, outcome.skipped_dupe, outcome.pruned
             );
             let _ = store::record_fetch(config, &source.id, Utc::now(), reason, &status);
             publish_global(DomainEvent::TaskSourceFetched {
@@ -56,6 +57,7 @@ pub async fn run_source_once(
                 fetched = outcome.fetched,
                 routed = outcome.routed,
                 skipped_dupe = outcome.skipped_dupe,
+                pruned = outcome.pruned,
                 "[task_sources:pipeline] fetch pass complete"
             );
         }
@@ -109,6 +111,8 @@ async fn run_inner(
     let fetch_filter = filter::to_fetch_filter(&source.filter, source.max_tasks_per_fetch);
     let tasks = provider.fetch_tasks(&ctx, &fetch_filter).await?;
     outcome.fetched = tasks.len();
+    let current_external_ids: HashSet<String> =
+        tasks.iter().map(|task| task.external_id.clone()).collect();
 
     for mut task in tasks {
         // Stamp the originating source before dedup / enrichment.
@@ -188,9 +192,50 @@ async fn run_inner(
         outcome.routed += 1;
     }
 
+    outcome.pruned = reconcile_missing_tasks(config, source, &current_external_ids)?;
+
     Ok(())
 }
 
+fn reconcile_missing_tasks(
+    config: &Config,
+    source: &TaskSource,
+    current_external_ids: &HashSet<String>,
+) -> Result<usize, String> {
+    let ingested = store::list_ingested_refs(config, &source.id)
+        .map_err(|e| format!("list_ingested_refs failed: {e}"))?;
+    let mut pruned = 0usize;
+
+    for item in ingested {
+        if current_external_ids.contains(&item.external_id) {
+            continue;
+        }
+
+        if let Some(card_id) = item.card_id.as_deref().filter(|id| !id.trim().is_empty()) {
+            route::remove_card(config, card_id).map_err(|e| {
+                format!(
+                    "remove stale card '{}' for source '{}' external task '{}': {e}",
+                    card_id, source.id, item.external_id
+                )
+            })?;
+        }
+
+        if store::remove_ingested(config, &source.id, &item.external_id)
+            .map_err(|e| format!("remove_ingested failed: {e}"))?
+        {
+            pruned += 1;
+            tracing::debug!(
+                source_id = %source.id,
+                provider = %source.provider.as_str(),
+                external_id = %item.external_id,
+                "[task_sources:pipeline] pruned task absent from latest source fetch"
+            );
+        }
+    }
+
+    Ok(pruned)
+}
+
 #[cfg(test)]
 #[path = "pipeline_tests.rs"]
 mod tests;

src/openhuman/task_sources/pipeline_tests.rs

@@ -174,6 +174,46 @@ async fn edited_task_reroutes_as_new_card() {
     assert_eq!(ingested[0].title, "Edited title");
 }
 
+#[tokio::test]
+async fn task_missing_from_latest_fetch_is_pruned() {
+    let _guard = registry_lock();
+    register_provider(Arc::new(StubProvider {
+        tasks: vec![
+            canned_task("1", "Keep task", "2025-01-01T00:00:00Z"),
+            canned_task("2", "Closed task", "2025-01-02T00:00:00Z"),
+        ],
+    }));
+
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    let source = add_github_source(&config);
+
+    let first = run_source_once(&config, &source, FetchReason::Manual).await;
+    assert_eq!(first.routed, 2, "error={:?}", first.error);
+    assert_eq!(route::board_cards(&config).unwrap().len(), 2);
+
+    // Simulate the provider returning only currently-open/matching tasks:
+    // task 2 was closed or no longer matches the source filter.
+    register_provider(Arc::new(StubProvider {
+        tasks: vec![canned_task("1", "Keep task", "2025-01-01T00:00:00Z")],
+    }));
+
+    let second = run_source_once(&config, &source, FetchReason::Manual).await;
+    assert_eq!(second.fetched, 1);
+    assert_eq!(second.routed, 0);
+    assert_eq!(second.skipped_dupe, 1);
+    assert_eq!(second.pruned, 1);
+    assert!(second.error.is_none());
+
+    let cards = route::board_cards(&config).unwrap();
+    assert_eq!(cards.len(), 1);
+    assert!(cards[0].title.contains("Keep task"));
+
+    let ingested = store::list_ingested(&config, &source.id, 10).unwrap();
+    assert_eq!(ingested.len(), 1);
+    assert_eq!(ingested[0].external_id, "1");
+}
+
 #[tokio::test]
 async fn missing_provider_surfaces_error_in_outcome() {
     let _guard = registry_lock();

src/openhuman/task_sources/route.rs

@@ -25,6 +25,13 @@ use super::TaskKind;
 /// Stable thread id whose board collects every ingested task.
 pub const TASK_SOURCES_THREAD_ID: &str = "task-sources";
 
+fn task_sources_location(config: &Config) -> BoardLocation {
+    BoardLocation::Thread {
+        workspace_dir: config.workspace_dir.clone(),
+        thread_id: TASK_SOURCES_THREAD_ID.to_string(),
+    }
+}
+
 /// Route an enriched task: append a todo card, then (for proactive
 /// sources) dispatch a triage turn. Returns the new card id on success.
 pub async fn route_enriched(
@@ -65,16 +72,13 @@ fn add_card(
     enriched: &EnrichedTask,
     stale_card_id: Option<&str>,
 ) -> Result<String, String> {
-    let location = BoardLocation::Thread {
-        workspace_dir: config.workspace_dir.clone(),
-        thread_id: TASK_SOURCES_THREAD_ID.to_string(),
-    };
+    let location = task_sources_location(config);
 
     // Remove stale card from the previous ingestion of this task (if any)
     // before creating the replacement, so the board never accumulates
     // duplicate cards for the same upstream item.
     if let Some(old_id) = stale_card_id {
-        match todo_remove(&location, old_id) {
+        match remove_card(config, old_id) {
             Ok(_) => {
                 tracing::debug!(
                     source_id = %source.id,
@@ -230,10 +234,7 @@ async fn dispatch_triage(
     // Link the envelope to the board card so triage's escalation arm routes
     // it through the deterministic dispatcher (claim → autonomous run →
     // write-back) instead of the one-shot triage sub-agent.
-    let location = BoardLocation::Thread {
-        workspace_dir: config.workspace_dir.clone(),
-        thread_id: TASK_SOURCES_THREAD_ID.to_string(),
-    };
+    let location = task_sources_location(config);
     let envelope = TriggerEnvelope::from_external(
         &format!("task_sources:{}", source.id),
         "external task ingested",
@@ -289,13 +290,28 @@ fn provider_label(provider: &str) -> String {
 pub fn board_cards(
     config: &Config,
 ) -> Result<Vec<crate::openhuman::agent::task_board::TaskBoardCard>, String> {
-    let location = BoardLocation::Thread {
-        workspace_dir: config.workspace_dir.clone(),
-        thread_id: TASK_SOURCES_THREAD_ID.to_string(),
-    };
+    let location = task_sources_location(config);
     todos::ops::list(&location).map(|snap| snap.cards)
 }
 
+/// Remove a task-source board card. Missing cards are treated as already
+/// reconciled so ledger cleanup can still proceed.
+pub fn remove_card(config: &Config, card_id: &str) -> Result<bool, String> {
+    let location = task_sources_location(config);
+    match todo_remove(&location, card_id) {
+        Ok(_) => Ok(true),
+        Err(e) if e.contains("not found") => {
+            tracing::debug!(
+                card_id,
+                error = %e,
+                "[task_sources:route] card already absent during reconciliation"
+            );
+            Ok(false)
+        }
+        Err(e) => Err(e),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/openhuman/task_sources/schemas.rs

@@ -53,6 +53,7 @@ pub fn all_controller_schemas() -> Vec<ControllerSchema> {
         schemas("update"),
         schemas("remove"),
         schemas("fetch"),
+        schemas("sync"),
         schemas("list_tasks"),
         schemas("preview_filter"),
         schemas("list_databases"),
@@ -86,6 +87,10 @@ pub fn all_registered_controllers() -> Vec<RegisteredController> {
             schema: schemas("fetch"),
             handler: handle_fetch,
         },
+        RegisteredController {
+            schema: schemas("sync"),
+            handler: handle_sync,
+        },
         RegisteredController {
             schema: schemas("list_tasks"),
             handler: handle_list_tasks,
@@ -230,6 +235,12 @@ pub fn schemas(function: &str) -> ControllerSchema {
                             comment: "True when the source was removed.",
                             required: true,
                         },
+                        FieldSchema {
+                            name: "pruned",
+                            ty: TypeSchema::U64,
+                            comment: "Count of ingested refs/cards pruned during removal.",
+                            required: true,
+                        },
                     ],
                 },
                 comment: "Removal result payload.",
@@ -239,7 +250,7 @@ pub fn schemas(function: &str) -> ControllerSchema {
         "fetch" => ControllerSchema {
             namespace: "task_sources",
             function: "fetch",
-            description: "Fetch one source immediately and route any new tasks.",
+            description: "Fetch one source immediately, route new tasks, and prune tasks no longer returned by the source.",
             inputs: vec![source_id_input(
                 "Identifier of the task source to fetch now.",
             )],
@@ -250,6 +261,18 @@ pub fn schemas(function: &str) -> ControllerSchema {
                 required: true,
             }],
         },
+        "sync" => ControllerSchema {
+            namespace: "task_sources",
+            function: "sync",
+            description: "Fetch every enabled source immediately, route new tasks, and prune tasks no longer returned by their source.",
+            inputs: vec![],
+            outputs: vec![FieldSchema {
+                name: "outcomes",
+                ty: TypeSchema::Array(Box::new(TypeSchema::Ref("FetchOutcome"))),
+                comment: "Fetch outcome counts for each enabled source.",
+                required: true,
+            }],
+        },
         "list_tasks" => ControllerSchema {
             namespace: "task_sources",
             function: "list_tasks",
@@ -444,6 +467,13 @@ fn handle_fetch(params: Map<String, Value>) -> ControllerFuture {
     })
 }
 
+fn handle_sync(_params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let config = config_rpc::load_config_with_timeout().await?;
+        to_json(ops::sync(&config).await?)
+    })
+}
+
 fn handle_list_tasks(params: Map<String, Value>) -> ControllerFuture {
     Box::pin(async move {
         let config = config_rpc::load_config_with_timeout().await?;
@@ -550,6 +580,7 @@ mod tests {
                 "update",
                 "remove",
                 "fetch",
+                "sync",
                 "list_tasks",
                 "preview_filter",
                 "list_databases",
@@ -561,7 +592,7 @@ mod tests {
     #[test]
     fn all_registered_controllers_has_handler_per_schema() {
         let controllers = all_registered_controllers();
-        assert_eq!(controllers.len(), 10);
+        assert_eq!(controllers.len(), all_controller_schemas().len());
         assert!(controllers
             .iter()
             .all(|c| c.schema.namespace == "task_sources"));

src/openhuman/task_sources/store.rs

@@ -25,6 +25,12 @@ use super::types::{
     FetchReason, FilterSpec, ProviderSlug, SourceTarget, TaskSource, TaskSourcePatch,
 };
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct IngestedTaskRef {
+    pub external_id: String,
+    pub card_id: Option<String>,
+}
+
 /// Compute an edit-aware content hash for a task. Two fetches of the
 /// same `external_id` whose title/body/status/updated_at/url differ produce
 /// different hashes, so an *edited* upstream item re-ingests.
@@ -309,6 +315,41 @@ pub fn get_card_id(config: &Config, source_id: &str, external_id: &str) -> Resul
     })
 }
 
+/// Return ingested task ids/card ids for one source. Used by reconciliation
+/// to prune board cards that no longer match the upstream source/filter.
+pub fn list_ingested_refs(config: &Config, source_id: &str) -> Result<Vec<IngestedTaskRef>> {
+    with_connection(config, |conn| {
+        let mut stmt = conn.prepare(
+            "SELECT external_id, card_id FROM ingested_tasks
+             WHERE source_id = ?1
+             ORDER BY ingested_at ASC, external_id ASC",
+        )?;
+        let rows = stmt.query_map(params![source_id], |row| {
+            Ok(IngestedTaskRef {
+                external_id: row.get(0)?,
+                card_id: row.get(1)?,
+            })
+        })?;
+        let mut out = Vec::new();
+        for row in rows {
+            out.push(row?);
+        }
+        Ok(out)
+    })
+}
+
+/// Delete one ingested ledger row after its board card has been reconciled.
+pub fn remove_ingested(config: &Config, source_id: &str, external_id: &str) -> Result<bool> {
+    let changed = with_connection(config, |conn| {
+        conn.execute(
+            "DELETE FROM ingested_tasks WHERE source_id = ?1 AND external_id = ?2",
+            params![source_id, external_id],
+        )
+        .context("Failed to delete ingested task")
+    })?;
+    Ok(changed > 0)
+}
+
 /// List the most recently ingested tasks for a source (newest first).
 pub fn list_ingested(
     config: &Config,

src/openhuman/task_sources/store_tests.rs

@@ -264,6 +264,45 @@ async fn add_with_assigned_executor_persists_and_filters_blank() {
     assert_eq!(blank.value.assigned_executor, None);
 }
 
+#[tokio::test]
+async fn ops_remove_prunes_routed_cards_for_source() {
+    use crate::openhuman::task_sources::{ops, route};
+    use crate::openhuman::todos::ops::{add as todo_add, BoardLocation, CardPatch};
+
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    let src = add_source(
+        &config,
+        ProviderSlug::Github,
+        None,
+        None,
+        github_filter(),
+        1800,
+        SourceTarget::TodoOnly,
+        25,
+    )
+    .unwrap();
+    let location = BoardLocation::Thread {
+        workspace_dir: config.workspace_dir.clone(),
+        thread_id: route::TASK_SOURCES_THREAD_ID.to_string(),
+    };
+    let snapshot = todo_add(&location, "[GitHub] A", CardPatch::default()).unwrap();
+    let card_id = snapshot.cards.last().unwrap().id.clone();
+    mark_ingested(
+        &config,
+        &src.id,
+        &sample_task("1", "A", "2025-01-01"),
+        &card_id,
+    )
+    .unwrap();
+
+    let out = ops::remove(&config, &src.id).await.expect("remove source");
+    assert_eq!(out.value["removed"], true);
+    assert_eq!(out.value["pruned"], 1);
+    assert!(route::board_cards(&config).unwrap().is_empty());
+    assert!(list_ingested(&config, &src.id, 10).unwrap().is_empty());
+}
+
 #[test]
 fn content_hash_changes_when_only_url_changes() {
     // `url` is load-bearing downstream (source_metadata / external write-back),

src/openhuman/task_sources/types.rs

@@ -239,6 +239,10 @@ pub struct FetchOutcome {
     pub routed: usize,
     /// Tasks skipped because they were already ingested.
     pub skipped_dupe: usize,
+    /// Previously ingested tasks removed because the upstream source no longer
+    /// returns them for this filter/status.
+    #[serde(default)]
+    pub pruned: usize,
     /// Optional human-readable status line.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub error: Option<String>,