diff --git a/internal/runtime/orchestration/doc.go b/internal/runtime/orchestration/doc.go new file mode 100644 index 0000000..baf8664 --- /dev/null +++ b/internal/runtime/orchestration/doc.go @@ -0,0 +1,43 @@ +// Package orchestration implements the sub-agent orchestration layer of the +// Cyrene v2 runtime (DevDocs/design/04-运行时.md §2 and DevDocs/design/05-子代理.md). +// +// When the front session (agent loop) decides a task is complex it hot-switches to +// orchestration mode and spawns background sub-agents that run concurrently +// without blocking the front session. Each sub-agent is an agent instance bound +// to a role + behavior mode, executing in an isolated context with its own +// working memory. Progress and status are streamed back to the front session via +// the agent-to-agent bus, and once all sub-agents finish their outcomes are +// merged for the front session. +// +// # Lifecycle +// +// A Scheduler is created once via New with its injected dependencies (a loop +// template, the bus, an optional comm probe, memory/sink, tracer, approver). A +// sub-agent is created with Spawn, which registers it, publishes a "dispatched" +// progress event and launches its execution in the background. The execution is +// gated by MaxParallel so at most that many sub-agents run at once; excess ones +// queue. Each run is bounded by PerAgentTimeout. The sub-agent moves +// pending -> running -> one of {done, failed, aborted}, and on every terminal +// transition a summary is written and a progress event is published. Abort and +// CancelAll stop a single sub-agent or every sub-agent (anti-runaway). +// +// # Result merging +// +// Join waits until every spawned sub-agent reaches a terminal state (or ctx is +// canceled) and returns a MergedResult aggregating each sub-agent's summary and +// loop result. +// +// # Isolation +// +// Sub-agent context and working memory are isolated by construction: Spawn deep +// copies Spec.Context into SubAgent.Context and creates a fresh WorkingMemory per +// sub-agent, so a sub-agent never shares the front session's context. +// +// # Governance +// +// Sensitive operations are audited through the injected *audit.Approver (see +// Approve): policy judgement belongs to the upper layer; the scheduler only +// triggers the audit record. Task-memory sinking is available through SinkMemory, +// which delegates to the injected *memory.Store so scope classification +// (user/persona/project) stays at the call site. +package orchestration diff --git a/internal/runtime/orchestration/merge.go b/internal/runtime/orchestration/merge.go new file mode 100644 index 0000000..7efb27f --- /dev/null +++ b/internal/runtime/orchestration/merge.go @@ -0,0 +1,94 @@ +package orchestration + +import ( + "fmt" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" +) + +// SubAgentResult is a point-in-time snapshot of a SubAgent. It is used both as +// the per-sub-agent entry in MergedResult and as the return value of +// (*SubAgent).Snapshot for monitoring. +type SubAgentResult struct { + // ID is the sub-agent id. + ID string + // Goal is the sub-agent's objective. + Goal string + // RoleID is the bound role/persona. + RoleID string + // Mode is the bound behavior mode. + Mode string + // AttachInfo is the supplied attachment text. + AttachInfo string + // Status is the terminal (or current) status. + Status Status + // Summary is the execution summary written on finalization. + Summary string + // Text is the loop result text on a successful run. + Text string + // Steps is the number of loop steps executed on a successful run. + Steps int + // Err is the error that ended a failed/aborted run (nil on success). + Err error + // StartedAt is when the sub-agent began running. + StartedAt time.Time + // FinishedAt is when the sub-agent finalized. + FinishedAt time.Time +} + +// MergedResult aggregates the outcomes of every spawned sub-agent, produced by +// (*Scheduler).Join. +type MergedResult struct { + // Agents holds one entry per spawned sub-agent, in spawn order. + Agents []SubAgentResult + // Total is the number of spawned sub-agents. + Total int + // Succeeded is how many sub-agents finished with StatusDone. + Succeeded int + // Failed is how many sub-agents finished with StatusFailed. + Failed int + // Aborted is how many sub-agents finished with StatusAborted. + Aborted int + // Summary is a one-line aggregate summary of all sub-agents. + Summary string +} + +// summaryTextMax caps how many runes of the result text appear in a summary. +const summaryTextMax = 80 + +// doneSummary builds the execution summary written when a sub-agent succeeds. +func doneSummary(res *loop.Result) string { + if res == nil { + return "完成" + } + text := compactText(res.Text, summaryTextMax) + if res.StopReason != "" { + return fmt.Sprintf("完成 (%s, 步骤=%d): %s", res.StopReason, res.Steps, text) + } + return fmt.Sprintf("完成 (步骤=%d): %s", res.Steps, text) +} + +// failedSummary builds the summary written when a sub-agent fails. +func failedSummary(err error) string { + if err == nil { + return "失败" + } + return "失败: " + err.Error() +} + +// canceledSummary builds the summary written when a sub-agent is aborted. +func canceledSummary() string { + return "中止: 已取消" +} + +// timeoutSummary builds the summary written when a sub-agent times out. +func timeoutSummary(d time.Duration) string { + return fmt.Sprintf("中止: 超时 (%s)", d) +} + +// summarizeMerged produces the one-line aggregate summary of a merged result. +func summarizeMerged(r *MergedResult) string { + return fmt.Sprintf("子代理汇总: 共 %d 个, %d 成功, %d 失败, %d 中止", + r.Total, r.Succeeded, r.Failed, r.Aborted) +} diff --git a/internal/runtime/orchestration/options.go b/internal/runtime/orchestration/options.go new file mode 100644 index 0000000..4407fc7 --- /dev/null +++ b/internal/runtime/orchestration/options.go @@ -0,0 +1,70 @@ +package orchestration + +import ( + "log/slog" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + "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/orchestration/audit" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration/comm" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// Defaults for the orchestration layer. +const ( + // DefaultMaxParallel is the maximum number of sub-agents that run at once + // when Options.MaxParallel is non-positive. + DefaultMaxParallel = 4 + + // defaultAgentID is the comm probe identity used when Options.AgentID is + // blank and a probe must be built from Bus. + defaultAgentID = "orchestrator" +) + +// Options configures a Scheduler built by New. All fields are optional; each nil +// dependency degrades gracefully (a missing Loop makes sub-agents fail fast, a +// missing Comm skips progress events, a missing Approver makes Approve error, +// and so on). +type Options struct { + // MaxParallel caps how many sub-agents execute concurrently. Excess sub-agents + // queue until a slot frees. Non-positive selects DefaultMaxParallel (4). + MaxParallel int + + // PerAgentTimeout bounds every sub-agent's run. When it elapses the sub-agent + // is aborted (status aborted). Non-positive means no timeout. + PerAgentTimeout time.Duration + + // Loop is the execution template: a *loop.Loop used to run every sub-agent. + // It must be safe for concurrent Run calls (as loop.Loop documents); the + // template's agent/tools/guardrails/persona apply to every sub-agent. When + // nil, sub-agents fail fast with ErrNoRunner. + Loop *loop.Loop + + // Bus is the in-process event bus used for progress events and, when Comm is + // nil, to build a default comm probe. When nil no events are published. + Bus bus.Bus + + // Comm is the comm probe used to publish progress events. When nil, a probe + // bound to AgentID is built from Bus (when Bus is non-nil). + Comm *comm.Probe + + // Memory is the long-term memory store used by SinkMemory. Optional. + Memory *memory.Store + + // Tracer emits a per-sub-agent span (kind subagent) around each run. Optional. + Tracer *trace.Tracer + + // Approver records sensitive-operation review-and-approve (代审) triggered by + // Approve. Policy judgement belongs to the upper layer. Optional. + Approver *audit.Approver + + // Logger is the base logger. Nil is replaced with a discard logger tagged with + // source "orchestration". + Logger *slog.Logger + + // AgentID is the comm probe identity. When blank and a probe must be built + // from Bus, "orchestrator" is used. + AgentID string +} diff --git a/internal/runtime/orchestration/orchestration_test.go b/internal/runtime/orchestration/orchestration_test.go new file mode 100644 index 0000000..ec3a7b6 --- /dev/null +++ b/internal/runtime/orchestration/orchestration_test.go @@ -0,0 +1,249 @@ +package orchestration + +import ( + "context" + "errors" + "fmt" + "runtime" + "strings" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration/comm" +) + +func TestJoinMergesResults(t *testing.T) { + t.Parallel() + s := newScheduler(t, Options{MaxParallel: 2}) + m := &mockRunner{result: &loop.Result{Text: "answer", Steps: 3, StopReason: "stop"}} + s.runner = m + + ctx := context.Background() + for i := 0; i < 3; i++ { + if _, err := s.Spawn(ctx, AgentSpec{Goal: fmt.Sprintf("goal %d", i)}); err != nil { + t.Fatalf("Spawn %d: %v", i, err) + } + } + + res, err := s.Join(ctx) + if err != nil { + t.Fatalf("Join: %v", err) + } + if res.Total != 3 || res.Succeeded != 3 || res.Failed != 0 || res.Aborted != 0 { + t.Fatalf("merged = %d total / %d ok / %d failed / %d aborted, want 3/3/0/0", + res.Total, res.Succeeded, res.Failed, res.Aborted) + } + if len(res.Agents) != 3 { + t.Fatalf("Agents len = %d, want 3", len(res.Agents)) + } + for _, a := range res.Agents { + if a.Status != StatusDone { + t.Fatalf("agent %s status = %s, want done", a.ID, a.Status) + } + if a.Summary == "" { + t.Fatalf("agent %s has no summary", a.ID) + } + if a.Text != "answer" || a.Steps != 3 { + t.Fatalf("agent %s result = text %q steps %d, want answer/3", a.ID, a.Text, a.Steps) + } + } + if !strings.Contains(res.Summary, "3 成功") { + t.Fatalf("merged summary = %q, want a 3-成功 aggregate", res.Summary) + } +} + +func TestFailureSummary(t *testing.T) { + t.Parallel() + s := newScheduler(t, Options{MaxParallel: 2}) + boom := errors.New("boom") + m := &mockRunner{err: boom} + s.runner = m + + sub, err := s.Spawn(context.Background(), AgentSpec{Goal: "x"}) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + waitFor(t, func() bool { return sub.Status() == StatusFailed }, waitTimeout) + if !strings.Contains(sub.Summary(), "boom") { + t.Fatalf("Summary = %q, want the failure recorded", sub.Summary()) + } + if !errors.Is(sub.Err(), boom) { + t.Fatalf("Err = %v, want boom", sub.Err()) + } + if sub.Result() != nil { + t.Fatalf("Result = %v, want nil on failure", sub.Result()) + } + + res, err := s.Join(context.Background()) + if err != nil { + t.Fatalf("Join: %v", err) + } + if res.Total != 1 || res.Failed != 1 || res.Succeeded != 0 { + t.Fatalf("merged = %d total / %d ok / %d failed, want 1/0/1", + res.Total, res.Succeeded, res.Failed) + } +} + +func TestContextIsolation(t *testing.T) { + t.Parallel() + s := newScheduler(t, Options{MaxParallel: 1}) + m := &mockRunner{result: &loop.Result{Text: "done"}} + s.runner = m + + specCtx := []model.Message{model.TextMessage(model.RoleUser, "hello")} + sub, err := s.Spawn(context.Background(), AgentSpec{Goal: "x", Context: specCtx}) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + + // Mutating the caller's slice after Spawn must not leak into the sub-agent. + specCtx[0].Content[0].Text = "mutated" + + res, err := s.Join(context.Background()) + if err != nil { + t.Fatalf("Join: %v", err) + } + if len(res.Agents) != 1 { + t.Fatalf("Agents len = %d, want 1", len(res.Agents)) + } + if m.total != 1 { + t.Fatalf("runner calls = %d, want 1", m.total) + } + got := m.captured(0) + if len(got) != 1 || got[0].Content[0].Text != "hello" { + t.Fatalf("runner received context = %#v, want the isolated 'hello' context", got) + } + if sub.Context[0].Content[0].Text != "hello" { + t.Fatalf("sub-agent context text = %q, want hello", sub.Context[0].Content[0].Text) + } +} + +func TestProgressEvents(t *testing.T) { + t.Parallel() + s := newScheduler(t, Options{MaxParallel: 1}) + gate := make(chan struct{}) + m := &mockRunner{gate: gate} + s.runner = m + + listener := comm.New(s.Bus(), "main") + events := make(chan comm.Event, 8) + unsub, err := listener.SubscribeEvent(context.Background(), func(_ context.Context, e comm.Event) error { + events <- e + return nil + }) + if err != nil { + t.Fatalf("SubscribeEvent: %v", err) + } + defer unsub() + + if _, err := s.Spawn(context.Background(), AgentSpec{Goal: "x"}); err != nil { + t.Fatalf("Spawn: %v", err) + } + waitFor(t, func() bool { return m.activeCount() >= 1 }, waitTimeout) + close(gate) + + names := collectEventNames(t, events, waitTimeout) + seen := make(map[string]bool) + for _, n := range names { + seen[n] = true + } + for _, want := range []string{EventDispatched, EventStarted, EventDone} { + if !seen[want] { + t.Fatalf("missing progress event %q, got %v", want, names) + } + } +} + +func TestJoin_CanceledCtx(t *testing.T) { + t.Parallel() + s := newScheduler(t, Options{MaxParallel: 1, PerAgentTimeout: 2 * time.Second}) + m := &mockRunner{gate: make(chan struct{})} + s.runner = m + + if _, err := s.Spawn(context.Background(), AgentSpec{Goal: "x"}); err != nil { + t.Fatalf("Spawn: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + res, err := s.Join(ctx) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Join err = %v, want DeadlineExceeded", err) + } + if res == nil { + t.Fatal("Join returned a nil merged result on canceled ctx") + } +} + +func TestApprove_NoApprover(t *testing.T) { + t.Parallel() + s := newScheduler(t, Options{}) + if _, err := s.Approve(context.Background(), "bob", "run", "do something"); !errors.Is(err, ErrNoApprover) { + t.Fatalf("Approve err = %v, want ErrNoApprover", err) + } +} + +func TestSinkMemory_NoStore(t *testing.T) { + t.Parallel() + s := newScheduler(t, Options{}) + sub, err := s.Spawn(context.Background(), AgentSpec{Goal: "x"}) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + if _, err := s.SinkMemory(context.Background(), sub, "remember this", "persona", "relationship"); !errors.Is(err, ErrNoMemory) { + t.Fatalf("SinkMemory err = %v, want ErrNoMemory", err) + } +} + +func TestRunWithRealLoop(t *testing.T) { + t.Parallel() + ad := &mockAdapter{resp: model.Response{Content: "final answer", Model: "mock"}} + ag, err := agent.New("You are a helper.", nil, nil, agent.WithModel(ad, nil)) + if err != nil { + t.Fatalf("agent.New: %v", err) + } + lp := loop.New(loop.Options{Agent: ag, MaxRounds: 3}) + s := newScheduler(t, Options{Loop: lp, MaxParallel: 2}) + + sub, err := s.Spawn(context.Background(), AgentSpec{Goal: "x"}) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + waitFor(t, func() bool { return sub.Status() == StatusDone }, waitTimeout) + if sub.Result() == nil { + t.Fatal("real loop produced no result") + } + if sub.Result().Text != "final answer" { + t.Fatalf("result text = %q, want final answer", sub.Result().Text) + } + + res, err := s.Join(context.Background()) + if err != nil { + t.Fatalf("Join: %v", err) + } + if res.Succeeded != 1 || res.Failed != 0 { + t.Fatalf("merged = %d ok / %d failed, want 1 ok", res.Succeeded, res.Failed) + } +} + +func TestNoGoroutineLeak(t *testing.T) { + // Not parallel: a goroutine-count assertion is noisy under parallel tests. + s := newScheduler(t, Options{MaxParallel: 2}) + m := &mockRunner{result: &loop.Result{Text: "ok"}} + s.runner = m + + baseline := runtime.NumGoroutine() + ctx := context.Background() + for i := 0; i < 20; i++ { + if _, err := s.Spawn(ctx, AgentSpec{Goal: fmt.Sprintf("g%d", i)}); err != nil { + t.Fatalf("Spawn %d: %v", i, err) + } + } + if _, err := s.Join(ctx); err != nil { + t.Fatalf("Join: %v", err) + } + waitFor(t, func() bool { return runtime.NumGoroutine() <= baseline+5 }, waitTimeout) +} diff --git a/internal/runtime/orchestration/scheduler.go b/internal/runtime/orchestration/scheduler.go new file mode 100644 index 0000000..78f166f --- /dev/null +++ b/internal/runtime/orchestration/scheduler.go @@ -0,0 +1,494 @@ +package orchestration + +import ( + "context" + "crypto/rand" + "encoding/hex" + "errors" + "fmt" + "io" + "log/slog" + "strings" + "sync" + "sync/atomic" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "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" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration/audit" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration/comm" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// runner executes one sub-agent's loop. A *loop.Loop satisfies it; tests inject a +// mock so the scheduler can be exercised without a model backend. +type runner interface { + Run(ctx context.Context, sessionID string, messages []model.Message) (*loop.Result, error) +} + +// Scheduler creates and runs sub-agents in the background, bounding concurrency, +// applying a per-agent timeout and publishing progress over the agent-to-agent +// bus. It is safe for concurrent use. +// +// The lifecycle is: Spawn registers a sub-agent and launches it asynchronously; +// the sub-agent's goroutine acquires a parallel slot (gated by MaxParallel), +// transitions to running, runs the injected loop template in its isolated +// context under the per-agent timeout, writes a summary and publishes a terminal +// progress event. Abort/CancelAll interrupt running or queued sub-agents. Join +// blocks until every sub-agent is terminal and returns the merged outcome. +type Scheduler struct { + maxParallel int + perAgentTimeout time.Duration + runner runner + bus bus.Bus + comm *comm.Probe + memory *memory.Store + tracer *trace.Tracer + approver *audit.Approver + log *slog.Logger + agentID string + + mu sync.Mutex + agents map[string]*SubAgent + active int + allDone chan struct{} + allOne sync.Once + sem chan struct{} + seq atomic.Uint64 + prefix string +} + +// noopRunner is the fallback runner used when no *loop.Loop template was +// configured. It makes every sub-agent fail fast with ErrNoRunner rather than +// silently doing nothing. +type noopRunner struct{} + +// Run returns a wrapped ErrNoRunner. +func (noopRunner) Run(context.Context, string, []model.Message) (*loop.Result, error) { + return nil, fmt.Errorf("orchestration: run sub-agent: %w", ErrNoRunner) +} + +// New builds a Scheduler from opts, applying defaults. MaxParallel non-positive +// selects DefaultMaxParallel; a nil Loop is replaced by a no-op runner; a nil +// Comm is built from Bus (when present) using AgentID (default "orchestrator"); a +// nil Logger is replaced by a discard logger tagged with source "orchestration". +func New(opts Options) *Scheduler { + maxParallel := opts.MaxParallel + if maxParallel <= 0 { + maxParallel = DefaultMaxParallel + } + agentID := opts.AgentID + if agentID == "" { + agentID = defaultAgentID + } + + s := &Scheduler{ + maxParallel: maxParallel, + perAgentTimeout: opts.PerAgentTimeout, + runner: wrapLoop(opts.Loop), + bus: opts.Bus, + memory: opts.Memory, + tracer: opts.Tracer, + approver: opts.Approver, + log: sourcedLog(opts.Logger), + agentID: agentID, + agents: make(map[string]*SubAgent), + allDone: make(chan struct{}), + sem: make(chan struct{}, maxParallel), + prefix: randPrefix(), + } + if opts.Comm != nil { + s.comm = opts.Comm + } else if opts.Bus != nil { + s.comm = comm.New(opts.Bus, agentID) + } + return s +} + +// Bus returns the injected event bus (may be nil). +func (s *Scheduler) Bus() bus.Bus { return s.bus } + +// Comm returns the injected (or auto-built) comm probe (may be nil). +func (s *Scheduler) Comm() *comm.Probe { return s.comm } + +// Spawn creates a sub-agent from spec, registers it, publishes a "dispatched" +// progress event and launches its execution in the background. It returns +// immediately so the front session is not blocked; the sub-agent runs +// asynchronously and reports progress over the bus. It returns ErrMissingGoal +// when spec.Goal is blank. +// +// The sub-agent's Context is a deep copy of spec.Context and its WorkingMemory is +// a fresh, isolated window, so it never shares state with the front session. +func (s *Scheduler) Spawn(ctx context.Context, spec AgentSpec) (*SubAgent, error) { + if strings.TrimSpace(spec.Goal) == "" { + return nil, fmt.Errorf("orchestration: spawn: %w", ErrMissingGoal) + } + if ctx == nil { + ctx = context.Background() + } + + sub := &SubAgent{ + ID: s.nextID(), + Goal: spec.Goal, + RoleID: spec.RoleID, + Mode: spec.Mode, + AttachInfo: spec.AttachInfo, + Data: spec.Data, + Context: cloneMessages(spec.Context), + WorkingMemory: memory.NewWorkingMemory(0), + status: StatusPending, + done: make(chan struct{}), + } + + s.mu.Lock() + if s.active == 0 { + s.allDone = make(chan struct{}) + s.allOne = sync.Once{} + } + s.active++ + s.agents[sub.ID] = sub + s.mu.Unlock() + + s.publish(EventDispatched, sub.progress()) + s.launch(ctx, sub) + return sub, nil +} + +// Run explicitly starts every registered sub-agent that has not started yet. It +// is idempotent and is provided for callers that prefer to defer execution; Spawn +// already auto-launches, so this is normally a no-op on freshly spawned batches. +func (s *Scheduler) Run(ctx context.Context) { + if ctx == nil { + ctx = context.Background() + } + s.mu.Lock() + subs := make([]*SubAgent, 0, len(s.agents)) + for _, sub := range s.agents { + subs = append(subs, sub) + } + s.mu.Unlock() + for _, sub := range subs { + s.launch(ctx, sub) + } +} + +// Join blocks until every spawned sub-agent is terminal (or ctx is canceled) and +// returns the merged outcome. On ctx cancellation it returns the partial merged +// outcome together with the wrapped ctx error. +func (s *Scheduler) Join(ctx context.Context) (*MergedResult, error) { + if ctx == nil { + ctx = context.Background() + } + s.mu.Lock() + ch := s.allDone + s.mu.Unlock() + + select { + case <-ch: + return s.merge(), nil + case <-ctx.Done(): + return s.merge(), ctx.Err() + } +} + +// Abort cancels the sub-agent with the given id. If it is running its loop is +// interrupted; if it is queued it is released without running; if it already +// finished the call is a no-op. It returns ErrAgentNotFound for an unknown id. +func (s *Scheduler) Abort(ctx context.Context, id string) error { + if ctx == nil { + ctx = context.Background() + } + s.mu.Lock() + sub, ok := s.agents[id] + s.mu.Unlock() + if !ok { + return fmt.Errorf("orchestration: abort: %w", ErrAgentNotFound) + } + sub.cancelRun() + return nil +} + +// CancelAll cancels every spawned sub-agent, queued or running. It is the +// anti-runaway escape hatch: after it returns each sub-agent converges to the +// aborted state. +func (s *Scheduler) CancelAll(ctx context.Context) error { + if ctx == nil { + ctx = context.Background() + } + s.mu.Lock() + subs := make([]*SubAgent, 0, len(s.agents)) + for _, sub := range s.agents { + subs = append(subs, sub) + } + s.mu.Unlock() + for _, sub := range subs { + sub.cancelRun() + } + return nil +} + +// Agents returns a snapshot of every registered sub-agent in spawn order. It is +// safe to read the returned snapshots at any time. +func (s *Scheduler) Agents() []SubAgentResult { + s.mu.Lock() + subs := make([]*SubAgent, 0, len(s.agents)) + for _, sub := range s.agents { + subs = append(subs, sub) + } + s.mu.Unlock() + + out := make([]SubAgentResult, 0, len(subs)) + for _, sub := range subs { + out = append(out, sub.Snapshot()) + } + return out +} + +// Approve triggers a main-agent review-and-approve (代审) of a sub-agent's +// sensitive action through the injected *audit.Approver. Whether an action is +// sensitive and whether it is granted is decided by the upper policy layer; this +// method only records the audit entry. It returns ErrNoApprover when no approver +// was injected. +func (s *Scheduler) Approve(ctx context.Context, actor, action, describe string) (*audit.Approval, error) { + if s.approver == nil { + return nil, fmt.Errorf("orchestration: approve: %w", ErrNoApprover) + } + return s.approver.ReviewAndApprove(ctx, actor, action, describe) +} + +// SinkMemory persists a task memory item for a sub-agent through the injected +// *memory.Store. The sub-agent's RoleID is used as the persona/role key; content +// and the scope/kind classification (user/persona/project) are supplied by the +// caller, keeping scope policy at the call site. It returns ErrNoMemory when no +// store was injected. +func (s *Scheduler) SinkMemory(ctx context.Context, sub *SubAgent, content string, scope memory.Scope, kind memory.Kind) (memory.MemoryItem, error) { + if s.memory == nil { + return memory.MemoryItem{}, fmt.Errorf("orchestration: sink memory: %w", ErrNoMemory) + } + if sub == nil { + return memory.MemoryItem{}, fmt.Errorf("orchestration: sink memory: %w", ErrAgentNotFound) + } + item := memory.MemoryItem{ + PersonaID: sub.RoleID, + RoleID: sub.RoleID, + Scope: scope, + Kind: kind, + Content: content, + } + return s.memory.Add(ctx, item) +} + +// launch starts the sub-agent's background execution exactly once. It is called +// by Spawn (auto-launch) and by Run (explicit start); launchOnce guarantees a +// single run even if both happen. +func (s *Scheduler) launch(ctx context.Context, sub *SubAgent) { + sub.launchOnce.Do(func() { + go s.run(ctx, sub) + }) +} + +// run drives one sub-agent: acquire a parallel slot, transition to running, run +// the loop in the sub-agent's isolated context under the per-agent timeout, then +// finalize with a summary and a terminal progress event. It always reaches +// exactly one finish, so active is decremented once and Join unblocks exactly +// when every sub-agent is terminal. +func (s *Scheduler) run(ctx context.Context, sub *SubAgent) { + if ctx == nil { + ctx = context.Background() + } + + select { + case s.sem <- struct{}{}: + defer func() { <-s.sem }() + case <-ctx.Done(): + s.finish(sub, StatusAborted, canceledSummary(), ctx.Err()) + return + case <-sub.done: + s.finish(sub, StatusAborted, canceledSummary(), context.Canceled) + return + } + + if sub.aborted() || !sub.transitionRunning() { + s.finish(sub, StatusAborted, canceledSummary(), context.Canceled) + return + } + s.publish(EventStarted, sub.progress()) + + runCtx, cancel := s.subCtx(ctx) + runCtx, span := s.subSpan(runCtx, sub) + sub.setCancel(cancel) + defer cancel() + + res, err := s.runner.Run(runCtx, sub.ID, sub.Context) + if err != nil { + s.spanErr(span, err) + s.abortOrFail(sub, err) + return + } + s.spanOK(span, res) + sub.setResult(res) + s.finish(sub, StatusDone, doneSummary(res), nil) +} + +// abortOrFail classifies an error after a sub-agent's run: a deadline exceeded or +// a cancellation yields the aborted status (timeout/stop), anything else is a +// real failure. +func (s *Scheduler) abortOrFail(sub *SubAgent, err error) { + switch { + case errors.Is(err, context.DeadlineExceeded): + s.finish(sub, StatusAborted, timeoutSummary(s.perAgentTimeout), err) + case errors.Is(err, context.Canceled): + s.finish(sub, StatusAborted, canceledSummary(), err) + default: + s.finish(sub, StatusFailed, failedSummary(err), err) + } +} + +// finish finalizes a sub-agent exactly once: it records the terminal status and +// summary, publishes the terminal progress event and decrements the active count, +// closing the allDone channel when the last sub-agent finalizes. +func (s *Scheduler) finish(sub *SubAgent, status Status, summary string, err error) { + sub.finishOnce.Do(func() { + sub.setOutcome(status, summary, err) + s.publish(eventFor(status), sub.progress()) + + s.mu.Lock() + s.active-- + if s.active == 0 { + s.allOne.Do(func() { close(s.allDone) }) + } + s.mu.Unlock() + }) +} + +// subCtx derives the run context, applying the per-agent timeout when configured. +func (s *Scheduler) subCtx(parent context.Context) (context.Context, context.CancelFunc) { + if s.perAgentTimeout > 0 { + return context.WithTimeout(parent, s.perAgentTimeout) + } + return context.WithCancel(parent) +} + +// subSpan starts a sub-agent span when a tracer is configured, chaining it as a +// child of any span already in ctx. It returns the ctx carrying the span and the +// span itself (nil when the tracer is nil). +func (s *Scheduler) subSpan(ctx context.Context, sub *SubAgent) (context.Context, *trace.Span) { + if s.tracer == nil { + return ctx, nil + } + return s.tracer.StartSpan(ctx, "subagent."+sub.ID, trace.SpanKindSubagent) +} + +// spanOK finishes a sub-agent span as ok, recording the result text. +func (s *Scheduler) spanOK(span *trace.Span, res *loop.Result) { + if span == nil { + return + } + if res != nil { + _ = span.SetProp("text", res.Text) + } + span.Finish(trace.SpanStatusOK) +} + +// spanErr finishes a sub-agent span as an error, recording the cause. +func (s *Scheduler) spanErr(span *trace.Span, err error) { + if span == nil { + return + } + _ = span.SetProp("error", err.Error()) + span.Finish(trace.SpanStatusError) +} + +// publish broadcasts a progress event over the bus. It is a no-op when no comm +// probe is configured. Publishing uses a background context so a canceled run +// still delivers its terminal event before the sub-agent is done. +func (s *Scheduler) publish(name string, data Progress) { + if s.comm == nil { + return + } + if err := s.comm.PublishEvent(context.Background(), comm.Event{Name: name, Data: data}); err != nil { + s.log.Warn("publish progress", "event", name, "agent_id", data.AgentID, "error", err) + } +} + +// nextID mints a unique sub-agent id from the scheduler's random prefix and an +// atomic sequence. It is safe for concurrent use. +func (s *Scheduler) nextID() string { + return fmt.Sprintf("%s-%d", s.prefix, s.seq.Add(1)) +} + +// 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() + + 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 +} + +// eventFor maps a terminal status to its progress event name. +func eventFor(status Status) string { + switch status { + case StatusDone: + return EventDone + case StatusFailed: + return EventFailed + case StatusAborted: + return EventAborted + default: + return EventStarted + } +} + +// wrapLoop adapts a *loop.Loop into the runner interface, substituting a no-op +// runner (fail-fast) when the template is nil. +func wrapLoop(l *loop.Loop) runner { + if l == nil { + return noopRunner{} + } + return l +} + +// sourcedLog returns the base logger tagged with source "orchestration", +// replacing a nil logger with a discard one so the scheduler can always log. +func sourcedLog(l *slog.Logger) *slog.Logger { + if l == nil { + l = slog.New(slog.NewTextHandler(io.Discard, nil)) + } + return logger.WithSource(l, "orchestration") +} + +// prefixBytes is the number of random bytes used to build the sub-agent id +// prefix. +const prefixBytes = 4 + +// randPrefix returns a short random prefix for sub-agent ids, falling back to +// "sub" if the system entropy source is unavailable. +func randPrefix() string { + var b [prefixBytes]byte + if _, err := rand.Read(b[:]); err != nil { + return "sub" + } + return "sub-" + hex.EncodeToString(b[:]) +} diff --git a/internal/runtime/orchestration/scheduler_test.go b/internal/runtime/orchestration/scheduler_test.go new file mode 100644 index 0000000..a7abd34 --- /dev/null +++ b/internal/runtime/orchestration/scheduler_test.go @@ -0,0 +1,321 @@ +package orchestration + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration/comm" +) + +const waitTimeout = 3 * time.Second + +// mockRunner is a configurable runner for scheduler tests. It counts concurrent +// invocations (for the parallel-limit test), captures the messages it receives +// (for the context-isolation test), and can block on a gate channel until closed +// or until ctx is done (to hold a sub-agent "running" for abort/timeout tests). +type mockRunner struct { + mu sync.Mutex + gate chan struct{} + active int + maxActive int + total int + calls [][]model.Message + result *loop.Result + err error +} + +func (m *mockRunner) Run(ctx context.Context, _ string, msgs []model.Message) (*loop.Result, error) { + m.mu.Lock() + m.active++ + if m.active > m.maxActive { + m.maxActive = m.active + } + m.total++ + m.calls = append(m.calls, cloneMessages(msgs)) + m.mu.Unlock() + + var gerr error + if m.gate != nil { + select { + case <-m.gate: + case <-ctx.Done(): + gerr = ctx.Err() + } + } + + m.mu.Lock() + m.active-- + m.mu.Unlock() + + if gerr != nil { + return nil, gerr + } + if m.err != nil { + return nil, m.err + } + if m.result != nil { + return m.result, nil + } + return &loop.Result{Text: "ok", Steps: 1, StopReason: "stop"}, nil +} + +func (m *mockRunner) activeCount() int { + m.mu.Lock() + defer m.mu.Unlock() + return m.active +} + +func (m *mockRunner) maxActiveCount() int { + m.mu.Lock() + defer m.mu.Unlock() + return m.maxActive +} + +func (m *mockRunner) captured(i int) []model.Message { + m.mu.Lock() + defer m.mu.Unlock() + return m.calls[i] +} + +// mockAdapter is a minimal model.Adapter so a real loop.Loop can be driven in +// tests without a provider. Chat returns a scripted final response. +type mockAdapter struct { + name string + resp model.Response + err error +} + +func (m *mockAdapter) Name() string { + if m.name == "" { + return "mock" + } + return m.name +} + +func (m *mockAdapter) Chat(context.Context, model.Request, model.Options) (model.Response, error) { + return m.resp, m.err +} + +func (m *mockAdapter) ChatStream(context.Context, model.Request, func(string)) (model.Response, error) { + return m.resp, m.err +} + +// newScheduler builds a Scheduler over a fresh in-process bus, wiring the default +// comm probe. The bus is closed on test cleanup. Callers set sched.runner after +// the call, or pass Options.Loop to drive a real loop. +func newScheduler(t *testing.T, opts Options) *Scheduler { + t.Helper() + b := bus.New() + t.Cleanup(func() { _ = b.Close() }) + opts.Bus = b + return New(opts) +} + +func waitFor(t *testing.T, cond func() bool, d time.Duration) { + t.Helper() + deadline := time.Now().Add(d) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("condition not met within %v", d) +} + +func countStatus(subs []*SubAgent, st Status) int { + n := 0 + for _, s := range subs { + if s.Status() == st { + n++ + } + } + return n +} + +// collectEventNames reads events from ch until a terminal event or the deadline +// elapses, returning the event names in order. +func collectEventNames(t *testing.T, ch <-chan comm.Event, d time.Duration) []string { + t.Helper() + var names []string + deadline := time.After(d) + for { + select { + case e := <-ch: + names = append(names, e.Name) + if e.Name == EventDone || e.Name == EventFailed || e.Name == EventAborted { + return names + } + case <-deadline: + return names + } + } +} + +func TestSpawnStatusFlow(t *testing.T) { + t.Parallel() + s := newScheduler(t, Options{MaxParallel: 2}) + m := &mockRunner{gate: make(chan struct{})} + s.runner = m + + sub, err := s.Spawn(context.Background(), AgentSpec{Goal: "do X"}) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + if sub.ID == "" { + t.Fatal("Spawn returned an empty id") + } + if sub.Goal != "do X" { + t.Fatalf("Goal = %q, want do X", sub.Goal) + } + if sub.WorkingMemory == nil { + t.Fatal("sub-agent has no isolated working memory") + } + + // pending -> running -> done. + waitFor(t, func() bool { return sub.Status() == StatusRunning }, waitTimeout) + close(m.gate) + waitFor(t, func() bool { return sub.Status() == StatusDone }, waitTimeout) + + if !sub.Done() { + t.Fatal("sub-agent not terminal after completion") + } + if sub.Summary() == "" { + t.Fatal("summary was not written on success") + } + if sub.Result() == nil { + t.Fatal("result was not stored on success") + } + if sub.Err() != nil { + t.Fatalf("Err = %v, want nil", sub.Err()) + } + if sub.FinishedAt().IsZero() { + t.Fatal("FinishedAt is not set on a terminal sub-agent") + } +} + +func TestSpawn_MissingGoal(t *testing.T) { + t.Parallel() + s := newScheduler(t, Options{}) + if _, err := s.Spawn(context.Background(), AgentSpec{}); !errors.Is(err, ErrMissingGoal) { + t.Fatalf("Spawn err = %v, want ErrMissingGoal", err) + } +} + +func TestParallelLimit(t *testing.T) { + t.Parallel() + s := newScheduler(t, Options{MaxParallel: 2}) + gate := make(chan struct{}) + m := &mockRunner{gate: gate} + s.runner = m + + ctx := context.Background() + var subs []*SubAgent + for i := 0; i < 5; i++ { + sub, err := s.Spawn(ctx, AgentSpec{Goal: fmt.Sprintf("task %d", i)}) + if err != nil { + t.Fatalf("Spawn %d: %v", i, err) + } + subs = append(subs, sub) + } + + // Wait until the two parallel slots are occupied, then confirm never more + // than MaxParallel run concurrently. + waitFor(t, func() bool { return countStatus(subs, StatusRunning) == 2 }, waitTimeout) + if got := m.maxActiveCount(); got != 2 { + t.Fatalf("max concurrent runs = %d, want 2 while gated", got) + } + if running := countStatus(subs, StatusRunning); running != 2 { + t.Fatalf("running = %d, want 2", running) + } + if pending := countStatus(subs, StatusPending); pending != 3 { + t.Fatalf("pending = %d, want 3 (queued behind the parallel limit)", pending) + } + + close(gate) + waitFor(t, func() bool { return countStatus(subs, StatusDone) == 5 }, waitTimeout) + + res, err := s.Join(ctx) + if err != nil { + t.Fatalf("Join: %v", err) + } + if res.Total != 5 || res.Succeeded != 5 { + t.Fatalf("merged = %d total / %d succeeded, want 5/5", res.Total, res.Succeeded) + } +} + +func TestPerAgentTimeout(t *testing.T) { + t.Parallel() + s := newScheduler(t, Options{MaxParallel: 1, PerAgentTimeout: 50 * time.Millisecond}) + m := &mockRunner{gate: make(chan struct{})} // never closed: block until timeout + s.runner = m + + sub, err := s.Spawn(context.Background(), AgentSpec{Goal: "slow"}) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + waitFor(t, func() bool { return sub.Status() == StatusRunning }, waitTimeout) + waitFor(t, func() bool { return sub.Status() == StatusAborted }, waitTimeout) + if !strings.Contains(sub.Summary(), "超时") { + t.Fatalf("Summary = %q, want a timeout notice", sub.Summary()) + } + if !errors.Is(sub.Err(), context.DeadlineExceeded) { + t.Fatalf("Err = %v, want DeadlineExceeded", sub.Err()) + } +} + +func TestAbort(t *testing.T) { + t.Parallel() + s := newScheduler(t, Options{MaxParallel: 1}) + m := &mockRunner{gate: make(chan struct{})} + s.runner = m + + sub, err := s.Spawn(context.Background(), AgentSpec{Goal: "x"}) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + waitFor(t, func() bool { return sub.Status() == StatusRunning }, waitTimeout) + + if err := s.Abort(context.Background(), sub.ID); err != nil { + t.Fatalf("Abort: %v", err) + } + waitFor(t, func() bool { return sub.Status() == StatusAborted }, waitTimeout) + + // Aborting an unknown id reports ErrAgentNotFound. + if err := s.Abort(context.Background(), "ghost"); !errors.Is(err, ErrAgentNotFound) { + t.Fatalf("Abort(ghost) err = %v, want ErrAgentNotFound", err) + } +} + +func TestCancelAll(t *testing.T) { + t.Parallel() + s := newScheduler(t, Options{MaxParallel: 2}) + m := &mockRunner{gate: make(chan struct{})} + s.runner = m + + ctx := context.Background() + var subs []*SubAgent + for i := 0; i < 3; i++ { + sub, err := s.Spawn(ctx, AgentSpec{Goal: fmt.Sprintf("task %d", i)}) + if err != nil { + t.Fatalf("Spawn %d: %v", i, err) + } + subs = append(subs, sub) + } + waitFor(t, func() bool { return countStatus(subs, StatusRunning) >= 1 }, waitTimeout) + + if err := s.CancelAll(ctx); err != nil { + t.Fatalf("CancelAll: %v", err) + } + for _, sub := range subs { + waitFor(t, func() bool { return sub.Status() == StatusAborted }, waitTimeout) + } +} diff --git a/internal/runtime/orchestration/spec.go b/internal/runtime/orchestration/spec.go new file mode 100644 index 0000000..65f1759 --- /dev/null +++ b/internal/runtime/orchestration/spec.go @@ -0,0 +1,133 @@ +package orchestration + +import ( + "errors" + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// Status is the lifecycle state of a spawned sub-agent. +// +// The lifecycle is pending -> running -> one of the terminal states +// (done, failed, aborted). A sub-agent is terminal exactly when Terminal +// reports true. +type Status string + +const ( + // StatusPending means the sub-agent was created but has not started. + StatusPending Status = "pending" + // StatusRunning means the sub-agent's loop is executing. + StatusRunning Status = "running" + // StatusDone means the loop completed and produced a result. + StatusDone Status = "done" + // StatusFailed means the loop returned an error. + StatusFailed Status = "failed" + // StatusAborted means the sub-agent was canceled or timed out. + StatusAborted Status = "aborted" +) + +// Terminal reports whether s is a final state. Pending and running are not +// terminal; every sub-agent ends in one of the three terminal states. +func (s Status) Terminal() bool { + return s == StatusDone || s == StatusFailed || s == StatusAborted +} + +// AgentSpec describes a sub-agent to spawn. Goal is required; the rest are +// optional and bound onto the sub-agent for the upper layer to interpret. +type AgentSpec struct { + // Goal is the objective the sub-agent works toward. It is required. + Goal string + // RoleID selects the role/persona the sub-agent is bound to. + RoleID string + // Mode selects the behavior mode (e.g. "engineer") for the role. + Mode string + // Context is the initial, isolated conversation context. Spawn deep-copies + // it, so later mutation by the caller does not affect the sub-agent. + Context []model.Message + // AttachInfo is free-form attachment/injection text supplied at spawn. + AttachInfo string + // Data carries an optional structured payload for the caller's use. + Data any +} + +// Progress event names published on the agent-to-agent bus via the Scheduler's +// comm probe. External subscribers can match on these to drive a monitoring +// panel (DevDocs/design/05-子代理.md §9). +const ( + // EventDispatched is published when a sub-agent is created. + EventDispatched = "subagent.dispatched" + // EventStarted is published when a sub-agent begins running. + EventStarted = "subagent.started" + // EventDone is published when a sub-agent finishes successfully. + EventDone = "subagent.done" + // EventFailed is published when a sub-agent fails. + EventFailed = "subagent.failed" + // EventAborted is published when a sub-agent is canceled or times out. + EventAborted = "subagent.aborted" +) + +// Progress is the status snapshot carried by a progress event's Data payload. +type Progress struct { + // AgentID is the spawned sub-agent id. + AgentID string + // Goal is the sub-agent's objective. + Goal string + // RoleID is the bound role/persona. + RoleID string + // Mode is the bound behavior mode. + Mode string + // Status is the sub-agent's status at publish time. + Status Status +} + +// Sentinel errors returned by the Scheduler. They are wrapped with %w at call +// sites so callers can test with errors.Is. +var ( + // ErrMissingGoal is returned by Spawn when the spec Goal is blank. + ErrMissingGoal = errors.New("orchestration: missing goal") + + // ErrNoRunner is returned when a sub-agent runs with no loop runner + // configured (Options.Loop was nil). + ErrNoRunner = errors.New("orchestration: no loop runner configured") + + // ErrNoApprover is returned by Approve when no *audit.Approver was injected. + ErrNoApprover = errors.New("orchestration: no approver configured") + + // ErrNoMemory is returned by SinkMemory when no *memory.Store was injected. + ErrNoMemory = errors.New("orchestration: no memory store configured") + + // ErrAgentNotFound is returned by Abort when the sub-agent id is unknown. + ErrAgentNotFound = errors.New("orchestration: sub-agent not found") +) + +// cloneMessages returns a deep copy of msgs: the message slice and each message's +// content parts are copied so the result shares no memory with the input. It is +// used at Spawn to give each sub-agent an isolated context. A nil input yields +// nil; a non-nil empty input yields a non-nil empty slice. +func cloneMessages(msgs []model.Message) []model.Message { + if msgs == nil { + return nil + } + out := make([]model.Message, len(msgs)) + for i, m := range msgs { + if m.Content != nil { + parts := make([]model.Content, len(m.Content)) + copy(parts, m.Content) + m.Content = parts + } + out[i] = m + } + return out +} + +// compactText shortens s to at most n runes for inclusion in a summary, appending +// an ellipsis when it was truncated. It never splits a multi-byte rune. +func compactText(s string, n int) string { + s = strings.TrimSpace(s) + r := []rune(s) + if len(r) <= n { + return s + } + return string(r[:n]) + "…" +} diff --git a/internal/runtime/orchestration/subagent.go b/internal/runtime/orchestration/subagent.go new file mode 100644 index 0000000..d732835 --- /dev/null +++ b/internal/runtime/orchestration/subagent.go @@ -0,0 +1,214 @@ +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 is a single spawned sub-agent instance. The identity fields (ID, Goal, +// RoleID, Mode, AttachInfo, Data, Context, WorkingMemory) are set once at Spawn +// and never mutated. The lifecycle fields (Status, Summary, Result, Err) are +// written by the sub-agent's background goroutine and read through the +// synchronized accessors below, so reading them before the sub-agent finishes is +// safe. A sub-agent is terminal exactly once its Done reports true; Join returns +// once every sub-agent is terminal. +// +// SubAgent is created by (*Scheduler).Spawn; it should not be built directly. +type SubAgent struct { + // ID uniquely identifies the sub-agent within its scheduler. + ID string + // Goal is the objective the sub-agent pursues. + Goal string + // RoleID is the bound role/persona. + RoleID string + // Mode is the bound behavior mode. + Mode string + // AttachInfo is the free-form attachment text supplied at spawn. + AttachInfo string + // Data is the optional structured payload supplied at spawn. + Data any + // Context is the isolated conversation context. It is a deep copy of the + // AgentSpec.Context and is never shared with the front session. + Context []model.Message + // WorkingMemory is the sub-agent's own working-memory window, created fresh + // at Spawn so it never overlaps another sub-agent or the front session. + WorkingMemory *memory.WorkingMemory + + 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 + + launchOnce sync.Once + finishOnce sync.Once +} + +// Status returns the sub-agent's current lifecycle status. +func (s *SubAgent) Status() Status { + s.mu.RLock() + defer s.mu.RUnlock() + return s.status +} + +// Summary returns the execution summary written when the sub-agent finalized: +// a completion summary on success, or a failure/abort reason otherwise. It is +// empty until the sub-agent is terminal. +func (s *SubAgent) Summary() string { + s.mu.RLock() + defer s.mu.RUnlock() + return s.summary +} + +// Result returns the loop result on a successful run, or nil when the run failed +// or was aborted. +func (s *SubAgent) Result() *loop.Result { + s.mu.RLock() + defer s.mu.RUnlock() + return s.result +} + +// Err returns the error that ended a failed or aborted run (nil on success). +func (s *SubAgent) Err() error { + s.mu.RLock() + defer s.mu.RUnlock() + return s.runErr +} + +// StartedAt returns when the sub-agent transitioned to running (zero if it never +// started). +func (s *SubAgent) StartedAt() time.Time { + s.mu.RLock() + defer s.mu.RUnlock() + return s.started +} + +// FinishedAt returns when the sub-agent reached a terminal state (zero if not +// terminal yet). +func (s *SubAgent) FinishedAt() time.Time { + s.mu.RLock() + defer s.mu.RUnlock() + return s.finished +} + +// Done reports whether the sub-agent reached a terminal state. +func (s *SubAgent) Done() bool { return s.Status().Terminal() } + +// Snapshot returns a point-in-time copy of the sub-agent's state, safe to read +// while the sub-agent is still running. +func (s *SubAgent) Snapshot() SubAgentResult { + s.mu.RLock() + defer s.mu.RUnlock() + snap := SubAgentResult{ + ID: s.ID, + Goal: s.Goal, + RoleID: s.RoleID, + 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 moves a pending sub-agent to running. It returns false when +// the sub-agent was already aborted (so it must not run) or is not pending. +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 returns the current status snapshot for a progress event. +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 records the terminal status, summary and error after the run ends. +// It is called at most once per sub-agent, guarded by 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 records the successful loop result. +func (s *SubAgent) setResult(res *loop.Result) { + s.mu.Lock() + defer s.mu.Unlock() + s.result = res +} + +// setCancel stores the run context's cancel function so Abort can interrupt the +// loop. If the sub-agent was already aborted (cancelRun raced ahead), the stored +// cancel is invoked immediately. +func (s *SubAgent) setCancel(c context.CancelFunc) { + s.mu.Lock() + s.cancel = c + aborted := s.abortReq + s.mu.Unlock() + if aborted { + c() + } +} + +// cancelRun requests cancellation of the sub-agent. It is idempotent and safe for +// concurrent use: the done channel is closed once and the stored cancel (when set) +// is invoked once. A pending sub-agent unblocks from its parallel slot via done; a +// running sub-agent is interrupted via the cancel func. +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 reports whether a cancellation was requested for the sub-agent. +func (s *SubAgent) aborted() bool { + s.mu.RLock() + defer s.mu.RUnlock() + return s.abortReq +}