AskaEth
de1fa83b79
修的是「文档说 ✅、生产不可达」：此前 Scheduler.Spawn 只有测试调用方、无 bus 入口、无派发工具；Span.Record 无生产调用方 → 泳道画不出、节点日志恒空。
- 入口：subagent_spawn 工具（internal/runtime/orchestration/spawntool.go，同包便于用真 Scheduler/loop/Tracer/Flow 做集成测试；注册在 cmd/cyrene/subagent_tool.go 照 terminal_tool 范式）。默认 owner-only，不设 AllowInGroup/AllowInStranger；安全仍由 restrictSchemas + applyToolSecurityGate 双层保证，未绕代审
- 父会话归属：AgentSpec.ParentSession（请求语义）→ Scheduler 转运行期 ctx 标记 loop.WithFlowBranch → 分支 span 以父 session 发布；**allow 谓词一行未改**；单图语义靠「分支 Run 不调 StartRound/EndRound」+ FlowPublisher 不新建图；未归属 spawn 一张图都不发（有测试锁）
- 前台最多等 2s（对齐 task.autoBackgroundThreshold）→ 合并结果作工具结果；转后台则广播 bus subagent.result（design/05 §9 回填 hook）；新增 Scheduler.Wait(ids) 按批等待（Join 会串并发批次，故不重用）
- 子代理跑在 WithoutCancel(ctx)：保留 span 归属、不被工具超时/回合结束顺手杀掉；生命周期由 PerAgentTimeout + CancelAll 约束；内部 WithMutedBroadcast 静音（不进前台 stream）
- 节点日志：loop/spanlog.go 新增 llm.done / tool.done / subagent.done；工具参数**只记键名不记值**；仅步骤真正结束才追加（跳过/被拒路径不发，不伪造）
- 脱敏（红线）：新增 trace/redact.go（形状规则 sk-/Bearer/JWT/key:value + 键名规则 api_key/token/authorization/password…，递归嵌套、不改调用方事件），graph.go 发布前必经；**顺带修掉一个既有泄漏**：result_summary prop 原先未脱敏（工具结果里的密钥会进图），由图级变异测试抓出
- 测试：新增 22 个（工具级 / 图级 / 日志级），含 4 次变异验证（去掉 WithFlowBranch、去掉 record*、还原未脱敏、去掉静音 各自 FAIL）；-race 过 orchestration+trace+loop
- 已知限制：真实模型端到端未验（无凭据/不起服务）；工具白名单是 design gap 未做；subagent.result 暂无消费者（design/15 未实现）；后台子代理跨轮次边界迟到 span 会丢泳道根；PerAgentTimeout=10min 待确认
86 lines
2.6 KiB
Go
86 lines
2.6 KiB
Go
package orchestration
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
)
|
|
|
|
// Wait blocks until every sub-agent named by ids reaches a terminal state (or ctx
|
|
// is canceled) and returns the merged outcome of exactly those sub-agents, in the
|
|
// order the ids were given.
|
|
//
|
|
// Unlike Join — which covers the whole scheduler — Wait scopes both the wait and
|
|
// the merge to one caller's batch, so concurrent spawns from different sessions
|
|
// never wait on, or merge, each other's sub-agents. It is what the spawn tool uses
|
|
// to collect the results it is about to hand back.
|
|
//
|
|
// It returns ErrAgentNotFound (wrapped with %w) for an unknown id, and the wrapped
|
|
// ctx error together with the partial merged outcome when ctx is canceled first.
|
|
func (s *Scheduler) Wait(ctx context.Context, ids []string) (*MergedResult, error) {
|
|
if ctx == nil {
|
|
ctx = context.Background()
|
|
}
|
|
subs, err := s.pick(ids)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for _, sub := range subs {
|
|
select {
|
|
case <-sub.waitDone():
|
|
case <-ctx.Done():
|
|
return mergeSubs(subs), fmt.Errorf("orchestration: wait: %w", ctx.Err())
|
|
}
|
|
}
|
|
return mergeSubs(subs), nil
|
|
}
|
|
|
|
// pick resolves ids to their registered sub-agents in the given order. It returns
|
|
// ErrAgentNotFound wrapped for the first unknown id.
|
|
func (s *Scheduler) pick(ids []string) ([]*SubAgent, error) {
|
|
subs := make([]*SubAgent, 0, len(ids))
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
for _, id := range ids {
|
|
sub, ok := s.agents[id]
|
|
if !ok {
|
|
return nil, fmt.Errorf("orchestration: wait %q: %w", id, ErrAgentNotFound)
|
|
}
|
|
subs = append(subs, sub)
|
|
}
|
|
return subs, nil
|
|
}
|
|
|
|
// merge aggregates every registered sub-agent into a MergedResult. It is called by
|
|
// Join once all sub-agents are terminal.
|
|
func (s *Scheduler) merge() *MergedResult {
|
|
s.mu.Lock()
|
|
subs := make([]*SubAgent, 0, len(s.agents))
|
|
for _, sub := range s.agents {
|
|
subs = append(subs, sub)
|
|
}
|
|
s.mu.Unlock()
|
|
return mergeSubs(subs)
|
|
}
|
|
|
|
// mergeSubs aggregates the given sub-agents into a MergedResult, counting the
|
|
// terminal statuses and appending each snapshot in slice order. A sub-agent that
|
|
// is still running contributes its current snapshot and is counted in neither
|
|
// Succeeded/Failed/Aborted.
|
|
func mergeSubs(subs []*SubAgent) *MergedResult {
|
|
res := &MergedResult{Total: len(subs), Agents: make([]SubAgentResult, 0, len(subs))}
|
|
for _, sub := range subs {
|
|
snap := sub.Snapshot()
|
|
res.Agents = append(res.Agents, snap)
|
|
switch snap.Status {
|
|
case StatusDone:
|
|
res.Succeeded++
|
|
case StatusFailed:
|
|
res.Failed++
|
|
case StatusAborted:
|
|
res.Aborted++
|
|
}
|
|
}
|
|
res.Summary = summarizeMerged(res)
|
|
return res
|
|
}
|