package orchestration import ( "context" "sync" "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" ) // SubAgent 是单个已派发的子代理实例。身份字段(ID、Goal、RoleID、PersonaID、Mode、 // AttachInfo、ParentSession、FlowOwner、Data、ToolWhitelist、Context、WorkingMemory) // 在 Spawn 时一次设定,此后永不改动。生命周期字段(Status、Summary、Result、Err)由 // 子代理的后台 goroutine 写入,并经下方同步访问器读取,因此在子代理结束前读取它们 // 是安全的。当且仅当 Done 报告 true 时子代理处于终结态;Join 在全部子代理终结后 // 返回,Wait 在指名的 id 全部终结后返回。 // // SubAgent 由 (*Scheduler).Spawn 创建;不应直接构造。 type SubAgent struct { // ID 在所属调度器内唯一标识该子代理。 ID string // Goal 是子代理追求的目标。 Goal string // RoleID 是绑定的角色/persona。 RoleID string // PersonaID 是子代理运行时充当的目标角色:其指令与记忆作用域(design/22 §8-2)。 // 留空表示「按派发回合的角色运行」,即既有行为。派发工具填充它(已对照按角色 // agent 池校验),Spawn 把它盖到运行的会话 env 上。 PersonaID string // Mode 是绑定的行为模式。 Mode string // AttachInfo 是派发时提供的自由形式附加文本。 AttachInfo string // ParentSession 是拥有本子代理分支的流程图所属会话(该分支不被可视化时为空)。 ParentSession string // FlowOwner 是本子代理 span 落进**哪一张** flow 图的归属桶 // (`{会话}[#{路}]`,见 loop.FlowOwnerFromContext):真并行时派发它的那一路 // 有自己的图,本子代理必须落进那一路,而不是会话主干那张 —— 否则两路的 // 子代理节点会互相长在对方的图上(design/15 §2.5.10-1)。留空时回落到 // ParentSession(=会话主干那张,单路/旧调用方的既有行为)。 FlowOwner string // Data 是派发时提供的可选结构化载荷。 Data any // ToolWhitelist 是该子代理的工具白名单(design/05 §7 裁决 02.4.1): // nil/空=继承主 agent 全量工具集;非空=只放行所列工具。Spawn 时深拷贝。 ToolWhitelist []string // Context 是隔离的会话上下文。它是 AgentSpec.Context 的深拷贝加上任务 user // 消息,绝不与前台会话共享。 Context []model.Message // WorkingMemory 是该子代理自己的工作记忆窗口,在 Spawn 时全新创建,因此绝不与 // 另一个子代理或前台会话重叠。 WorkingMemory *memory.WorkingMemory // timeout 是该子代理生效的运行时限(按调用覆盖优先于调度器默认, // design/05 §7 裁决 02.4.4);非正表示除取消外无期限。Spawn 时落定,只读。 timeout time.Duration mu sync.RWMutex status Status summary string result *loop.Result runErr error started time.Time finished time.Time done chan struct{} cancel context.CancelFunc abortReq bool doneClosed bool // finishedCh 在子代理进入终结态时恰好关闭一次,因此 Wait 能按 id 阻塞而无需 // 轮询。 finishedCh chan struct{} launchOnce sync.Once finishOnce sync.Once } // Status 返回子代理当前的生命周期状态。 func (s *SubAgent) Status() Status { s.mu.RLock() defer s.mu.RUnlock() return s.status } // Summary 返回子代理收尾时写入的执行摘要:成功时是完成摘要,否则是失败/中止原因。 // 子代理进入终结态之前它为空。 func (s *SubAgent) Summary() string { s.mu.RLock() defer s.mu.RUnlock() return s.summary } // Result 返回运行成功时的循环结果;运行失败或被中止时返回 nil。 func (s *SubAgent) Result() *loop.Result { s.mu.RLock() defer s.mu.RUnlock() return s.result } // Err 返回结束失败或中止运行的那个错误(成功时为 nil)。 func (s *SubAgent) Err() error { s.mu.RLock() defer s.mu.RUnlock() return s.runErr } // StartedAt 返回子代理转入 running 的时间(从未启动时为零值)。 func (s *SubAgent) StartedAt() time.Time { s.mu.RLock() defer s.mu.RUnlock() return s.started } // FinishedAt 返回子代理进入终结态的时间(尚未终结时为零值)。 func (s *SubAgent) FinishedAt() time.Time { s.mu.RLock() defer s.mu.RUnlock() return s.finished } // Done 报告子代理是否已进入终结态。 func (s *SubAgent) Done() bool { return s.Status().Terminal() } // flowOwner 返回本子代理 span 该落进哪一张 flow 图的归属桶 // (`{会话}[#{路}]`):显式声明了 FlowOwner 时用它(真并行路派发的子代理因此落进 // **那一路**的图),否则回落到 ParentSession(会话主干那张图,即单路与旧调用方 // 的既有行为)。两者都为空时返回空串 —— 调用方据此抑制发布(不被可视化的分支)。 func (s *SubAgent) flowOwner() string { if s == nil { return "" } if s.FlowOwner != "" { return s.FlowOwner } return s.ParentSession } // waitDone 返回子代理进入终结态时被关闭的 channel。它是 (*Scheduler).Wait 按 id 阻塞 // 时等待的对象,无需轮询。 func (s *SubAgent) waitDone() <-chan struct{} { return s.finishedCh } // markFinished 恰好关闭一次完成 channel。它在 finish 里于 finishOnce 之下调用,因此 // 每个子代理恰好唤醒它的每个 Wait 调用方一次。 func (s *SubAgent) markFinished() { close(s.finishedCh) } // Snapshot 返回子代理状态的某一时刻拷贝,在子代理仍在运行时读取也安全。 func (s *SubAgent) Snapshot() SubAgentResult { s.mu.RLock() defer s.mu.RUnlock() snap := SubAgentResult{ ID: s.ID, Goal: s.Goal, RoleID: s.RoleID, PersonaID: s.PersonaID, Mode: s.Mode, AttachInfo: s.AttachInfo, Status: s.status, Summary: s.summary, Err: s.runErr, StartedAt: s.started, FinishedAt: s.finished, } if s.result != nil { snap.Text = s.result.Text snap.Steps = s.result.Steps } return snap } // transitionRunning 把 pending 的子代理转入 running。子代理已被中止(因此不得运行) // 或当前不是 pending 时返回 false。 func (s *SubAgent) transitionRunning() bool { s.mu.Lock() defer s.mu.Unlock() if s.abortReq || s.status != StatusPending { return false } s.status = StatusRunning s.started = time.Now() return true } // progress 返回用于进度事件的当前状态快照。 func (s *SubAgent) progress() Progress { s.mu.RLock() defer s.mu.RUnlock() return Progress{ AgentID: s.ID, Goal: s.Goal, RoleID: s.RoleID, Mode: s.Mode, Status: s.status, } } // setOutcome 在运行结束后记录终结状态、摘要与错误。每个子代理至多调用一次,由 // finishOnce 把守。 func (s *SubAgent) setOutcome(status Status, summary string, err error) { s.mu.Lock() defer s.mu.Unlock() s.status = status s.summary = summary s.runErr = err s.finished = time.Now() } // setResult 记录成功时的循环结果。 func (s *SubAgent) setResult(res *loop.Result) { s.mu.Lock() defer s.mu.Unlock() s.result = res } // setCancel 保存运行上下文的取消函数,使 Abort 能中断循环。若子代理已被中止 // (cancelRun 先行),则立即调用保存的 cancel。 func (s *SubAgent) setCancel(c context.CancelFunc) { s.mu.Lock() s.cancel = c aborted := s.abortReq s.mu.Unlock() if aborted { c() } } // cancelRun 请求取消该子代理。它幂等且可并发使用:done channel 只关闭一次,已保存的 // cancel(若已设置)只调用一次。排队中的子代理经 done 解除对并行槽位的阻塞;运行中的 // 子代理经 cancel 函数被中断。 func (s *SubAgent) cancelRun() { s.mu.Lock() s.abortReq = true c := s.cancel first := !s.doneClosed s.doneClosed = true s.mu.Unlock() if first { close(s.done) } if c != nil { c() } } // aborted 报告该子代理是否已被请求取消。 func (s *SubAgent) aborted() bool { s.mu.RLock() defer s.mu.RUnlock() return s.abortReq }