From 54e66695b157ccecc2d1d0ba54cb144fc3c4a0e1 Mon Sep 17 00:00:00 2001 From: harryert <4210852+harryert@users.noreply.github.com> Date: Wed, 28 Oct 2026 11:13:24 +0800 Subject: [PATCH] fix(ecc2): prevent recursive auto-dispatch of assignment handoffs --- ecc2/src/session/manager.rs | 76 +++++++++++++++++++++++++ ecc2/src/session/store.rs | 108 +++++++++++++++++++++++++++++++----- 2 files changed, 171 insertions(+), 13 deletions(-) diff --git a/ecc2/src/session/manager.rs b/ecc2/src/session/manager.rs index aca68a676..43417a22c 100644 --- a/ecc2/src/session/manager.rs +++ b/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")?; diff --git a/ecc2/src/session/store.rs b/ecc2/src/session/store.rs index de1af81fc..9fa0c2e58 100644 --- a/ecc2/src/session/store.rs +++ b/ecc2/src/session/store.rs @@ -2801,7 +2801,11 @@ impl StateStore { }) })?; - let mut messages = messages.collect::, _>>()?; + let mut messages = messages + .collect::, _>>()? + .into_iter() + .filter(|message| !is_assignment_handoff(&message.content, &message.from_session)) + .collect::>(); 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 { - 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> { 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 = HashMap::new(); - for (to_session, content, id) in targets.collect::, _>>()? { + for (to_session, content, id, sender) in targets.collect::, _>>()? { + 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 { 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")?;