fix(ecc2): prevent recursive auto-dispatch of assignment handoffs

ecc2/src/session/manager.rs

@@ -7658,6 +7658,82 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test(flavor = "current_thread")]
+    async fn auto_dispatch_does_not_recursively_route_assigned_handoffs() -> Result<()> {
+        let tempdir = TestDir::new("manager-auto-dispatch-no-recursion")?;
+        let repo_root = tempdir.path().join("repo");
+        init_git_repo(&repo_root)?;
+
+        let mut cfg = build_config(tempdir.path());
+        cfg.auto_dispatch_limit_per_session = 5;
+        let db = StateStore::open(&cfg.db_path)?;
+        let now = Utc::now();
+        db.insert_session(&Session {
+            id: "parent".to_string(),
+            task: "parent task".to_string(),
+            project: "workspace".to_string(),
+            task_group: "general".to_string(),
+            agent_type: "claude".to_string(),
+            working_dir: repo_root,
+            state: SessionState::Running,
+            pid: Some(42),
+            worktree: None,
+            created_at: now - Duration::minutes(3),
+            updated_at: now - Duration::minutes(3),
+            last_heartbeat_at: now - Duration::minutes(3),
+            metrics: SessionMetrics::default(),
+        })?;
+        db.send_message(
+            "operator",
+            "parent",
+            "{\"task\":\"Handle incoming request\",\"context\":\"External handoff\"}",
+            "task_handoff",
+        )?;
+
+        let first_tick = auto_dispatch_backlog(&db, &cfg, "claude", true, 10).await?;
+        assert_eq!(first_tick.len(), 1);
+        assert_eq!(first_tick[0].routed.len(), 1);
+        let child_id = first_tick[0].routed[0].session_id.clone();
+        assert_eq!(
+            db.latest_task_handoff_source(&child_id)?.as_deref(),
+            Some("parent")
+        );
+        assert_eq!(db.unread_task_handoff_count(&child_id)?, 0);
+        assert!(db.unread_task_handoff_targets(10)?.is_empty());
+
+        let second_tick = auto_dispatch_backlog(&db, &cfg, "claude", true, 10).await?;
+        assert!(second_tick.is_empty());
+        assert_eq!(db.delegated_children("parent", 10)?, vec![child_id.clone()]);
+        assert!(db.delegated_children(&child_id, 10)?.is_empty());
+        assert_eq!(
+            db.latest_task_handoff_source(&child_id)?.as_deref(),
+            Some("parent")
+        );
+
+        db.send_message(
+            "operator",
+            &child_id,
+            "{\"task\":\"Explain the phrase spawned new delegate\",\"context\":\"Intentional delegation\"}",
+            "task_handoff",
+        )?;
+        assert_eq!(db.unread_task_handoff_count(&child_id)?, 1);
+        let explicit_tick = auto_dispatch_backlog(&db, &cfg, "claude", true, 10).await?;
+        assert!(explicit_tick
+            .iter()
+            .any(|outcome| { outcome.lead_session_id == child_id && outcome.routed.len() == 1 }));
+        let grandchildren = db.delegated_children(&child_id, 10)?;
+        assert_eq!(grandchildren.len(), 1);
+        let grandchild_id = grandchildren[0].clone();
+        assert_eq!(
+            db.latest_task_handoff_source(&grandchild_id)?.as_deref(),
+            Some(child_id.as_str())
+        );
+        assert_eq!(db.unread_task_handoff_count(&grandchild_id)?, 0);
+        assert!(db.unread_task_handoff_targets(10)?.is_empty());
+
+        Ok(())
+    }
+
     #[tokio::test(flavor = "current_thread")]
     async fn coordinate_backlog_reports_remaining_backlog_after_limited_pass() -> Result<()> {
         let tempdir = TestDir::new("manager-coordinate-backlog")?;

ecc2/src/session/store.rs

@@ -2801,7 +2801,11 @@ impl StateStore {
             })
         })?;
 
-        let mut messages = messages.collect::<Result<Vec<_>, _>>()?;
+        let mut messages = messages
+            .collect::<Result<Vec<_>, _>>()?
+            .into_iter()
+            .filter(|message| !is_assignment_handoff(&message.content, &message.from_session))
+            .collect::<Vec<_>>();
         messages.sort_by(|left, right| {
             let left_priority = comms::handoff_priority(&left.content);
             let right_priority = comms::handoff_priority(&right.content);
@@ -2814,21 +2818,29 @@ impl StateStore {
     }
 
     pub fn unread_task_handoff_count(&self, session_id: &str) -> Result<usize> {
-        self.conn
-            .query_row(
-                "SELECT COUNT(*)
-                 FROM messages
-                 WHERE to_session = ?1 AND msg_type = 'task_handoff' AND read = 0",
-                rusqlite::params![session_id],
-                |row| row.get::<_, i64>(0),
-            )
-            .map(|count| count as usize)
-            .map_err(Into::into)
+        // Match the dispatchable handoffs used by inbox draining and auto-dispatch.
+        // Internal assignment notices remain readable in normal message history.
+        let mut stmt = self.conn.prepare(
+            "SELECT from_session, content
+             FROM messages
+             WHERE to_session = ?1 AND msg_type = 'task_handoff' AND read = 0",
+        )?;
+        let messages = stmt.query_map(rusqlite::params![session_id], |row| {
+            Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
+        })?;
+        let mut count = 0;
+        for message in messages {
+            let (sender, content) = message?;
+            if !is_assignment_handoff(&content, &sender) {
+                count += 1;
+            }
+        }
+        Ok(count)
     }
 
     pub fn unread_task_handoff_targets(&self, limit: usize) -> Result<Vec<(String, usize)>> {
         let mut stmt = self.conn.prepare(
-            "SELECT to_session, content, id
+            "SELECT to_session, content, id, from_session
              FROM messages
              WHERE msg_type = 'task_handoff' AND read = 0
              ORDER BY id ASC",
@@ -2839,10 +2851,14 @@ impl StateStore {
                 row.get::<_, String>(0)?,
                 row.get::<_, String>(1)?,
                 row.get::<_, i64>(2)?,
+                row.get::<_, String>(3)?,
             ))
         })?;
         let mut aggregated: HashMap<String, (usize, comms::TaskPriority, i64)> = HashMap::new();
-        for (to_session, content, id) in targets.collect::<Result<Vec<_>, _>>()? {
+        for (to_session, content, id, sender) in targets.collect::<Result<Vec<_>, _>>()? {
+            if is_assignment_handoff(&content, &sender) {
+                continue;
+            }
             let priority = comms::handoff_priority(&content);
             aggregated
                 .entry(to_session)
@@ -4938,6 +4954,29 @@ fn routing_activity_suffix(context: &str) -> Option<&'static str> {
     }
 }
 
+/// ECC's own assignment notification is a task handoff for the worker's
+/// inbox/lineage, but must not be interpreted as a *new* delegation request.
+/// Match the generated sender, context delimiter, and routing-reason suffix;
+/// a user task merely mentioning the same words remains dispatchable.
+fn is_assignment_handoff(content: &str, sender: &str) -> bool {
+    let Some(comms::MessageType::TaskHandoff { context, .. }) = comms::parse(content) else {
+        return false;
+    };
+    let expected_prefix = format!("Assigned by {sender} [");
+    if !context.starts_with(&expected_prefix) || !context.contains("] | cwd ") {
+        return false;
+    }
+
+    [
+        "reused idle delegate",
+        "spawned new delegate",
+        "reused active delegate at capacity",
+        "spawned fallback delegate",
+    ]
+    .iter()
+    .any(|reason| context.ends_with(&format!(" | {reason}")))
+}
+
 fn extract_task_handoff_context(content: &str) -> Option<String> {
     if let Some(crate::comms::MessageType::TaskHandoff { context, .. }) =
         crate::comms::parse(content)
@@ -6993,6 +7032,49 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn internal_assignment_notice_is_not_dispatchable_or_counted() -> Result<()> {
+        let tempdir = TestDir::new("store-assignment-notice")?;
+        let db = StateStore::open(&tempdir.path().join("state.db"))?;
+        db.insert_session(&build_session("lead", SessionState::Running))?;
+        db.insert_session(&build_session("worker", SessionState::Running))?;
+
+        let internal = serde_json::to_string(&comms::MessageType::TaskHandoff {
+            task: "Implement the first work item".to_string(),
+            context: "Assigned by lead [claude] | cwd /repo | spawned new delegate".to_string(),
+            priority: comms::TaskPriority::Normal,
+        })?;
+        db.send_message("lead", "worker", &internal, "task_handoff")?;
+        assert_eq!(db.unread_task_handoff_count("worker")?, 0);
+        assert!(db.unread_task_handoff_targets(10)?.is_empty());
+        assert!(db
+            .unread_task_handoffs_for_session("worker", 10)?
+            .is_empty());
+        assert_eq!(
+            db.latest_task_handoff_source("worker")?.as_deref(),
+            Some("lead")
+        );
+
+        // Even a matching routing phrase in a user message is not an internal
+        // assignment when the actual sender differs from the generated context.
+        db.send_message("operator", "worker", &internal, "task_handoff")?;
+        db.send_message(
+            "operator",
+            "worker",
+            r#"{"task":"Explain spawned new delegate","context":"Explicit task"}"#,
+            "task_handoff",
+        )?;
+        assert_eq!(db.unread_task_handoff_count("worker")?, 2);
+        assert_eq!(db.unread_task_handoff_targets(10)?, vec![("worker".to_string(), 2)]);
+        assert_eq!(db.unread_task_handoffs_for_session("worker", 10)?.len(), 2);
+
+        // The original notification remains available to the human/agent inbox,
+        // even though it never spawns a recursive grandchild.
+        let all = db.list_messages_for_session("worker", 10)?;
+        assert_eq!(all.iter().filter(|message| message.msg_type == "task_handoff").count(), 3);
+        Ok(())
+    }
+
     #[test]
     fn scheduled_tasks_round_trip_and_advance_runs() -> Result<()> {
         let tempdir = TestDir::new("store-scheduled-tasks")?;