【Bug已解决】Subagent activity is invisible from the main thread list, spinner only reflects main thread 解决方案

原始报错：Subagent activity is invisible from the main thread list, spinner only reflects main thread 场景：一个任务会派生出多个子智能体（subagent）并行干活。主界面只显示主线程的进度转圈（spinner），子智能体在干什么、是否卡住、是否完成，在主任务列表里完全看不到。用户只能干等主线程转圈，无法判断子任务是不是已经挂了。 关键词：子任务可见性、任务树、状态聚合、事件总线、进度 spinner 聚合。

一、现象长什么样

主任务列表里有一行："任务 A —— 进行中（转圈）"。但任务 A 内部派了 3 个子智能体分别查资料、写代码、跑测试。用户看到的现象是：

1. 列表里只有"任务 A"一行，没有"任务 A / 子任务1 查资料"之类；
2. 转圈只反映主线程是否空闲，子智能体忙不忙它不体现；
3. 如果某个子智能体卡死，主线程还在等它，转圈一直转，用户无从知道"其实卡在子任务3"；
4. 子智能体完成了，界面也不会有任何独立提示，只有等全部结束后主行才变"完成"。

这是典型的监控盲区：系统把"主线程"当成唯一可观测对象，忽略了它派生出的工作树。用户失去了对并行子任务的可观测性，也就失去了判断进度和定位卡点的能力。

二、背景：为什么只看到主线程

任务系统通常这样实现：每个顶层任务有一个状态对象，spinner绑定到这个对象。子智能体往往是"函数内部spawn出来的"，没有注册到任何全局任务登记表（registry），UI 也只订阅顶层任务。于是：

• 子智能体的生命周期对 UI 不可见；
• 子智能体的进度事件发到了别处，或被直接丢弃；
• 主线程的 spinner 由"主线程是否在等"驱动，而"在等子任务"和"真的空闲"在它看来没区别。

要让子任务可见，必须建立任务树 + 事件总线，让所有层级的任务都向同一个可见性系统汇报。

三、根因：没有任务登记表，事件不向上聚合

根因拆解：

1. 子任务未登记：派生子智能体时没有把它加进任务树，UI 根本不知道它的存在。
2. 事件不发总线：子任务的状态变化（开始/进展/完成/失败）只存在局部变量里，没发到全局事件总线。
3. 聚合缺失：主行 spinner 只反映主线程，没把子任务的活跃度聚合进来。
4. 卡点不可定位：因为没树、没事件，子任务卡死时无法在 UI 标红/标注。

下面用最小模型复现"只反映主线程"，再给修复。

四、最小可运行复现

用线程模拟主任务派生子任务。错误写法：只跟踪主线程是否结束，子线程状态没人管。

import threading, time class Task: def __init__(self, name): self.name = name self.done = False def run(self): time.sleep(0.3) self.done = True def main_task(): t = Task("主任务") subs = [] for i in range(3): st = threading.Thread(target=lambda: time.sleep(0.5 + i * 0.2)) st.start() subs.append(st) # 主线程只等自己这点活，不等子任务 t.run() # 错误：UI 只看 t.done，子任务状态完全未知 return t.done if __name__ == "__main__": ok = main_task() print("主任务完成:", ok) # 此时子任务可能还在跑，但没有任何可见状态

这里main_task返回时，子任务还在后台跑，但系统没有任何结构记录它们，UI 自然看不到。

五、方案：建立任务树登记表

第一层：所有任务（含子任务）都注册到一个全局树，每个节点有id / parent / status / progress：

import threading, time, uuid from dataclasses import dataclass, field from typing import Dict, List, Optional @dataclass class TaskNode: id: str name: str parent: Optional[str] status: str = "pending" # pending/running/done/failed progress: float = 0.0 class TaskRegistry: def __init__(self): self._nodes: Dict[str, TaskNode] = {} def spawn(self, name: str, parent: Optional[str] = None) -> str: tid = uuid.uuid4().hex[:8] self._nodes[tid] = TaskNode(id=tid, name=name, parent=parent, status="running") return tid def update(self, tid: str, status: str = None, progress: float = None): n = self._nodes.get(tid) if n is None: return if status: n.status = status if progress is not None: n.progress = progress def children(self, tid: str) -> List[TaskNode]: return [n for n in self._nodes.values() if n.parent == tid] def all(self) -> List[TaskNode]: return list(self._nodes.values()) if __name__ == "__main__": reg = TaskRegistry() root = reg.spawn("主任务") c1 = reg.spawn("查资料", parent=root) c2 = reg.spawn("写代码", parent=root) print("可见任务数:", len(reg.all())) # 3（主+2子，全部可见） print("主任务的子任务:", [n.name for n in reg.children(root)])

现在 UI 只要遍历registry.all()，就能列出任务树，子任务不再隐身。

六、方案：事件总线让状态变化可订阅

第二层：子任务状态变化通过事件总线广播，UI 订阅即可实时刷新，不必轮询：

class EventBus: def __init__(self): self._subs = [] def subscribe(self, fn): self._subs.append(fn) def publish(self, event: dict): for fn in self._subs: fn(event) def make_worker(reg: TaskRegistry, bus: EventBus, tid: str, work: float): def run(): for step in range(1, 6): time.sleep(work / 5) reg.update(tid, progress=step / 5) bus.publish({"task": tid, "progress": step / 5}) reg.update(tid, status="done") bus.publish({"task": tid, "status": "done"}) return threading.Thread(target=run) if __name__ == "__main__": reg = TaskRegistry() bus = EventBus() bus.subscribe(lambda e: print("[事件]", e)) root = reg.spawn("主任务") t1 = reg.spawn("查资料", parent=root) w = make_worker(reg, bus, t1, 0.5) w.start() w.join()

每个进度/状态变化都通过bus.publish发出，UI 订阅后就能实时显示子任务进展，不再只有主线程转圈。

七、方案：聚合 spinner 反映整体活跃度

第三层：主行 spinner 不应只看主线程，而应根据整棵子树是否还有活跃节点来聚合：

def is_active(reg: TaskRegistry, tid: str) -> bool: node = reg._nodes.get(tid) if node is None: return False if node.status in ("running", "pending"): return True # 任一子节点活跃，则整体活跃 return any(is_active(reg, c.id) for c in reg.children(tid)) def aggregate_progress(reg: TaskRegistry, tid: str) -> float: node = reg._nodes.get(tid) if node is None: return 0.0 children = reg.children(tid) if not children: return node.progress # 子树平均进度 return sum(aggregate_progress(reg, c.id) for c in children) / len(children) if __name__ == "__main__": reg = TaskRegistry() bus = EventBus() root = reg.spawn("主任务") t1 = reg.spawn("查资料", parent=root) t2 = reg.spawn("写代码", parent=root) reg.update(t1, status="done", progress=1.0) reg.update(t2, status="running", progress=0.5) print("主任务是否活跃:", is_active(reg, root)) # True（子任务2还在跑） print("聚合进度:", round(aggregate_progress(reg, root), 2)) # 0.75

is_active递归检查子树，只要还有一个子任务在跑，主行 spinner 就继续转；aggregate_progress给出整棵树的真实进度，比"只看主线程"准确得多。

八、验证：把任务树与聚合锁进测试

def test_subtasks_visible_in_registry(): reg = TaskRegistry() root = reg.spawn("主任务") reg.spawn("子1", parent=root) reg.spawn("子2", parent=root) assert len(reg.children(root)) == 2 def test_aggregate_active_while_child_runs(): reg = TaskRegistry() root = reg.spawn("主任务") c = reg.spawn("子", parent=root) reg.update(c, status="running") assert is_active(reg, root) is True reg.update(c, status="done") # 主任务本身还没标 done assert is_active(reg, root) is True if __name__ == "__main__": test_subtasks_visible_in_registry() test_aggregate_active_while_child_runs() print("子任务可见性与聚合测试通过。")

九、排查清单（"子任务活动不可见"按顺序查）

1. 任务登记表：派生子任务时是否注册到全局任务树？UI 能否枚举到它们？
2. 事件总线：子任务状态变化是否发到可订阅的总线？还是只留在局部变量？
3. 树结构：每个任务是否记录parent，能否按树展开？
4. 聚合：主行 spinner / 进度是否聚合子树？还是只看主线程？
5. 卡点定位：子任务失败时 UI 能否标红/标注？还是只有整棵树卡住？
6. 生命周期：子任务结束/失败是否同步更新父节点状态？
7. 订阅清理：UI 关闭时是否退订总线，避免泄漏？

十、小结

"子任务活动不可见、spinner 只反映主线程"是可观测性只建在主线程上导致的监控盲区。修复三层：

• 登记：所有层级的任务都进全局任务树，UI 能枚举整棵工作树；
• 广播：状态变化走事件总线，UI 实时刷新，不靠轮询；
• 聚合：主行 spinner / 进度递归聚合子树，整体活跃度与真实进度一目了然。

核心原则：派生的每一点工作，都应当是可见、可订阅、可聚合的。做到这三点，用户不再被一个"永远在转却不知在等什么"的圈圈吊着，子任务卡死也能立刻在界面上定位到具体哪一支。