Let subagents run in the background and be reviewed
Why: A parent turn should be able to finish while children keep working, and a review should keep only findings that a second pass confirms. Users also need to override a profile's tools, model, and turn cap without adding dispatch tools. Evidence: go test ./... passed. New cases cover overlay rejection, role mapping, queued notices, and dropping FAIL or unparsed findings. Impact: Notices live only in memory, and each conversation allows three background children. Squad writer steps now use the writer profile. Refs: feature/agent-harness-p3b Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Vendored
+1
-1
Submodule deps/agent-system-core updated: fdc0c1b933...9b8cf79c96
@@ -0,0 +1,19 @@
|
||||
---
|
||||
id: agent.sub.review
|
||||
layer: agent
|
||||
version: 1
|
||||
title: 评审子代理(只读找问题)
|
||||
description: 只读审查材料,输出 JSON 问题清单,不修改任何内容
|
||||
tools: knowledge_search, web_search, tavily_search, time_now, list_conversations, read_conversation, fs_read_file, fs_list_dir, quality_review
|
||||
model: lightweight
|
||||
maxTurns: 8
|
||||
---
|
||||
你是只读评审子代理。只找问题,不修改材料,也不下最终通过结论。
|
||||
|
||||
- 严格只读:不写入、不修改、不删除。
|
||||
- 每条问题都要有材料中的定位和证据,不接受没有出处的猜测。
|
||||
- 只输出一个 JSON 数组,不要加解释。元素字段为 location、severity、issue、evidence。
|
||||
- severity 取 critical、high、medium、low 之一。
|
||||
|
||||
示例:
|
||||
[{"location":"第 2 段","severity":"high","issue":"结论缺少来源","evidence":"原文只写了「显然」"}]
|
||||
+23
-36
@@ -107,8 +107,12 @@ func ParseModuleFile(name, content string) (core.PromptModule, error) {
|
||||
m.OutputContract = v
|
||||
case "triggers":
|
||||
// 值留空;列表项由后续 "- kind:" 行提供。
|
||||
case "tools", "model", "maxTurns":
|
||||
// 子代理 profile 只从内置库文件读取,不写入 PromptModule。
|
||||
case "tools":
|
||||
m.Tools = splitList(v)
|
||||
case "model":
|
||||
m.Model = v
|
||||
case "maxTurns":
|
||||
m.MaxTurns = atoi(v)
|
||||
default:
|
||||
return m, fmt.Errorf("%s: unknown field %q", name, k)
|
||||
}
|
||||
@@ -122,50 +126,33 @@ func ParseModuleFile(name, content string) (core.PromptModule, error) {
|
||||
return m, nil
|
||||
}
|
||||
|
||||
// SubagentSpec 是内置子代理 profile。用户覆盖正文时这三项保持内置值。
|
||||
// SubagentSpec 是子代理 profile。空字段表示沿用内置。
|
||||
type SubagentSpec struct {
|
||||
Tools []string
|
||||
Model string
|
||||
MaxTurns int
|
||||
}
|
||||
|
||||
// specFromModule 由模块字段派生 profile。未写的 model / maxTurns 用默认值。
|
||||
func specFromModule(m core.PromptModule) SubagentSpec {
|
||||
spec := SubagentSpec{Tools: m.Tools, Model: m.Model, MaxTurns: m.MaxTurns}
|
||||
if spec.Model == "" {
|
||||
spec.Model = "lightweight"
|
||||
}
|
||||
if spec.MaxTurns <= 0 {
|
||||
spec.MaxTurns = 8
|
||||
}
|
||||
return spec
|
||||
}
|
||||
|
||||
// ParseSubagentSpec 从库文件 frontmatter 取出 tools、model、maxTurns。
|
||||
// 标准模块字段可以同时出现。其他未知键报错。
|
||||
func ParseSubagentSpec(content string) (SubagentSpec, error) {
|
||||
spec := SubagentSpec{Model: "lightweight", MaxTurns: 8}
|
||||
if !strings.HasPrefix(content, "---\n") {
|
||||
return spec, fmt.Errorf("subagent spec: missing frontmatter")
|
||||
m, err := ParseModuleFile("subagent", content)
|
||||
if err != nil {
|
||||
return SubagentSpec{}, err
|
||||
}
|
||||
end := strings.Index(content[4:], "\n---\n")
|
||||
if end < 0 {
|
||||
return spec, fmt.Errorf("subagent spec: unterminated frontmatter")
|
||||
}
|
||||
for _, line := range strings.Split(content[4:4+end], "\n") {
|
||||
trimmed := strings.TrimSpace(line)
|
||||
if trimmed == "" || strings.HasPrefix(trimmed, "- ") {
|
||||
continue
|
||||
}
|
||||
if line[0] == ' ' || line[0] == '\t' {
|
||||
continue
|
||||
}
|
||||
k, v := splitKV(trimmed)
|
||||
switch k {
|
||||
case "id", "layer", "version", "title", "description", "priority", "requiresAttachments", "outputContract", "triggers":
|
||||
case "tools":
|
||||
spec.Tools = splitList(v)
|
||||
case "model":
|
||||
if v != "" {
|
||||
spec.Model = v
|
||||
}
|
||||
case "maxTurns":
|
||||
if n := atoi(v); n > 0 {
|
||||
spec.MaxTurns = n
|
||||
}
|
||||
default:
|
||||
return spec, fmt.Errorf("subagent spec: unknown field %q", k)
|
||||
}
|
||||
}
|
||||
return spec, nil
|
||||
return specFromModule(m), nil
|
||||
}
|
||||
|
||||
func appendBlock(m *core.PromptModule, key, line string) {
|
||||
|
||||
@@ -368,7 +368,7 @@ func TestBuiltinSubagentSpecs(t *testing.T) {
|
||||
"agent.sub.writer": {"chat", 12, "run_python", "delegate_task"},
|
||||
}
|
||||
for id, w := range want {
|
||||
spec, ok := svc.SubagentSpec(id)
|
||||
spec, ok := svc.SubagentSpec("", id)
|
||||
if !ok {
|
||||
t.Errorf("missing spec %s", id)
|
||||
continue
|
||||
@@ -377,11 +377,79 @@ func TestBuiltinSubagentSpecs(t *testing.T) {
|
||||
t.Errorf("%s = %+v", id, spec)
|
||||
}
|
||||
}
|
||||
if _, ok := svc.SubagentSpec("agent.sub.missing"); ok {
|
||||
if _, ok := svc.SubagentSpec("", "agent.sub.missing"); ok {
|
||||
t.Error("unknown spec should be absent")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSubagentOverrideChain(t *testing.T) {
|
||||
st, err := store.OpenSQLite(t.TempDir() + "/t.db")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer st.Close()
|
||||
lib, _ := DefaultLibrary()
|
||||
svc, err := NewService(st, lib)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
base, ok := svc.SubagentSpec("alice", "agent.sub.explore")
|
||||
if !ok || !containsStr(base.Tools, "knowledge_search") || base.Model != "lightweight" {
|
||||
t.Fatalf("builtin base = %+v ok=%v", base, ok)
|
||||
}
|
||||
if err := svc.Save("alice", core.PromptModule{
|
||||
ID: "agent.sub.explore", Layer: core.LayerAgent, Title: "探索", Body: "只读。",
|
||||
Model: "chat",
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got, _ := svc.SubagentSpec("alice", "agent.sub.explore")
|
||||
if got.Model != "chat" || !containsStr(got.Tools, "knowledge_search") || got.MaxTurns != base.MaxTurns {
|
||||
t.Fatalf("partial overlay = %+v, base tools should stay", got)
|
||||
}
|
||||
if err := svc.Save("alice", core.PromptModule{
|
||||
ID: "agent.sub.explore", Layer: core.LayerAgent, Title: "探索", Body: "只读。",
|
||||
Tools: []string{"time_now"}, MaxTurns: 4,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got, _ = svc.SubagentSpec("alice", "agent.sub.explore")
|
||||
if len(got.Tools) != 1 || got.Tools[0] != "time_now" || got.MaxTurns != 4 || got.Model != "lightweight" {
|
||||
t.Fatalf("tools/turns overlay = %+v", got)
|
||||
}
|
||||
other, _ := svc.SubagentSpec("bob", "agent.sub.explore")
|
||||
if other.Model != "lightweight" || !containsStr(other.Tools, "knowledge_search") {
|
||||
t.Fatalf("other owner changed: %+v", other)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSubagentOverrideRejected(t *testing.T) {
|
||||
st, err := store.OpenSQLite(t.TempDir() + "/t.db")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer st.Close()
|
||||
lib, _ := DefaultLibrary()
|
||||
svc, err := NewService(st, lib)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cases := []core.PromptModule{
|
||||
{ID: "task.summarize", Layer: core.LayerTask, Title: "总结", Body: "正文", Tools: []string{"time_now"}},
|
||||
{ID: "agent.sub.explore", Layer: core.LayerAgent, Title: "探索", Body: "正文", Model: "huge"},
|
||||
{ID: "agent.sub.explore", Layer: core.LayerAgent, Title: "探索", Body: "正文", MaxTurns: 31},
|
||||
{ID: "agent.sub.explore", Layer: core.LayerAgent, Title: "探索", Body: "正文", MaxTurns: -1},
|
||||
{ID: "agent.sub.explore", Layer: core.LayerAgent, Title: "探索", Body: "正文", Tools: []string{"delegate_task"}},
|
||||
{ID: "agent.sub.explore", Layer: core.LayerAgent, Title: "探索", Body: "正文", Tools: []string{"squad_orchestrate"}},
|
||||
{ID: "agent.sub.explore", Layer: core.LayerAgent, Title: "探索", Body: "正文", Tools: []string{"review_findings"}},
|
||||
}
|
||||
for _, m := range cases {
|
||||
if err := svc.Save("alice", m); err == nil {
|
||||
t.Errorf("expected reject %+v", m)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func containsStr(list []string, want string) bool {
|
||||
for _, s := range list {
|
||||
if s == want {
|
||||
|
||||
@@ -77,13 +77,38 @@ func NewService(st *store.Store, lib []core.PromptModule) (*Service, error) {
|
||||
return &Service{st: st, builtin: lib, subagents: specs}, nil
|
||||
}
|
||||
|
||||
// SubagentSpec 返回内置子代理 profile。用户覆盖不能改 tools、model、maxTurns。
|
||||
func (s *Service) SubagentSpec(id string) (SubagentSpec, bool) {
|
||||
// SubagentSpec 以内置 spec 为底,用该 owner 的用户行里非空字段逐项覆盖。
|
||||
func (s *Service) SubagentSpec(owner, id string) (SubagentSpec, bool) {
|
||||
if s == nil {
|
||||
return SubagentSpec{}, false
|
||||
}
|
||||
spec, ok := s.subagents[id]
|
||||
return spec, ok
|
||||
if !ok {
|
||||
return SubagentSpec{}, false
|
||||
}
|
||||
if s.st == nil || owner == "" {
|
||||
return spec, true
|
||||
}
|
||||
user, err := s.st.ListPromptModules(owner)
|
||||
if err != nil {
|
||||
return spec, true
|
||||
}
|
||||
for _, u := range user {
|
||||
if u.ID != id {
|
||||
continue
|
||||
}
|
||||
if len(u.Tools) > 0 {
|
||||
spec.Tools = u.Tools
|
||||
}
|
||||
if u.Model != "" {
|
||||
spec.Model = u.Model
|
||||
}
|
||||
if u.MaxTurns > 0 {
|
||||
spec.MaxTurns = u.MaxTurns
|
||||
}
|
||||
break
|
||||
}
|
||||
return spec, true
|
||||
}
|
||||
|
||||
func builtinSubagentSpecs() (map[string]SubagentSpec, error) {
|
||||
@@ -105,11 +130,7 @@ func builtinSubagentSpecs() (map[string]SubagentSpec, error) {
|
||||
if mod.Layer != "agent" || !strings.HasPrefix(mod.ID, "agent.sub.") {
|
||||
continue
|
||||
}
|
||||
spec, err := ParseSubagentSpec(text)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out[mod.ID] = spec
|
||||
out[mod.ID] = specFromModule(mod)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
@@ -161,8 +182,38 @@ func (s *Service) Get(owner, id string) (core.PromptModule, bool, error) {
|
||||
return core.PromptModule{}, false, nil
|
||||
}
|
||||
|
||||
var dispatchTools = map[string]struct{}{
|
||||
"delegate_task": {}, "squad_orchestrate": {}, "review_findings": {},
|
||||
}
|
||||
|
||||
// validateSubagentOverride 只允许 agent.sub.* 携带 tools/model/maxTurns,
|
||||
// 并且不能把派发类工具加进子代理。
|
||||
func validateSubagentOverride(m core.PromptModule) error {
|
||||
if len(m.Tools) == 0 && m.Model == "" && m.MaxTurns == 0 {
|
||||
return nil
|
||||
}
|
||||
if !strings.HasPrefix(m.ID, "agent.sub.") {
|
||||
return fmt.Errorf("tools/model/maxTurns 只允许出现在 agent.sub.*")
|
||||
}
|
||||
if m.Model != "" && m.Model != "lightweight" && m.Model != "chat" {
|
||||
return fmt.Errorf("model 只能是 lightweight 或 chat")
|
||||
}
|
||||
if m.MaxTurns < 0 || m.MaxTurns > 30 {
|
||||
return fmt.Errorf("maxTurns 必须在 0 到 30 之间")
|
||||
}
|
||||
for _, name := range m.Tools {
|
||||
if _, bad := dispatchTools[name]; bad {
|
||||
return fmt.Errorf("子代理不能使用派发工具 %s", name)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Save 用户覆盖/新建:覆盖内置时记录 BasedOnVersion 供 drift 提示。
|
||||
func (s *Service) Save(owner string, m core.PromptModule) error {
|
||||
if err := validateSubagentOverride(m); err != nil {
|
||||
return err
|
||||
}
|
||||
m.Source = string(core.SourceUser)
|
||||
m.Owner = owner
|
||||
for _, b := range s.builtin {
|
||||
|
||||
@@ -0,0 +1,231 @@
|
||||
package runmgr
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"agent-system-backend/internal/runtime"
|
||||
core "agent-system-core"
|
||||
)
|
||||
|
||||
type gateRunner struct {
|
||||
entered chan struct{}
|
||||
release chan struct{}
|
||||
once sync.Once
|
||||
parts []core.Part
|
||||
err error
|
||||
spawn func(runtime.RunCtx) error
|
||||
}
|
||||
|
||||
func (g *gateRunner) Name() string { return "gate" }
|
||||
|
||||
func (g *gateRunner) Execute(rc runtime.RunCtx) error {
|
||||
if g.entered != nil {
|
||||
g.once.Do(func() { close(g.entered) })
|
||||
}
|
||||
if g.release != nil {
|
||||
<-g.release
|
||||
}
|
||||
if g.spawn != nil {
|
||||
return g.spawn(rc)
|
||||
}
|
||||
if g.err != nil {
|
||||
return g.err
|
||||
}
|
||||
if len(g.parts) > 0 {
|
||||
return rc.SetAssistantParts(g.parts)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type noopRunner struct{}
|
||||
|
||||
func (noopRunner) Name() string { return "noop" }
|
||||
func (noopRunner) Execute(rc runtime.RunCtx) error {
|
||||
return rc.SetAssistantParts([]core.Part{{Type: core.PartText, Text: "已汇报"}})
|
||||
}
|
||||
|
||||
func waitUntil(t *testing.T, cond func() bool) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if cond() {
|
||||
return
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
t.Fatal("timed out")
|
||||
}
|
||||
|
||||
func systemTexts(t *testing.T, c *runCtx) []string {
|
||||
t.Helper()
|
||||
msgs, err := c.mgr.st.ListMessages(c.run.ConversationID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var out []string
|
||||
for _, m := range msgs {
|
||||
if m.Role != core.RoleSystem {
|
||||
continue
|
||||
}
|
||||
for _, p := range m.Parts {
|
||||
if p.Type == core.PartText {
|
||||
out = append(out, p.Text)
|
||||
}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func TestBackgroundDoesNotBlockParent(t *testing.T) {
|
||||
entered := make(chan struct{})
|
||||
release := make(chan struct{})
|
||||
runner := &gateRunner{
|
||||
entered: entered, release: release,
|
||||
parts: []core.Part{{Type: core.PartText, Text: "后台结果"}},
|
||||
}
|
||||
c, _ := newSpawnEnv(t, runner)
|
||||
c.mgr.runner = noopRunner{}
|
||||
id, err := c.SpawnBackground("agent.sub.explore", "检索", "查资料", "")
|
||||
if err != nil || id == "" {
|
||||
t.Fatalf("spawn: %v %s", err, id)
|
||||
}
|
||||
select {
|
||||
case <-entered:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("child did not start")
|
||||
}
|
||||
close(release)
|
||||
waitUntil(t, func() bool {
|
||||
if len(systemTexts(t, c)) != 1 {
|
||||
return false
|
||||
}
|
||||
c.mgr.convMu.Lock()
|
||||
busy := c.mgr.convActive[c.run.ConversationID]
|
||||
c.mgr.convMu.Unlock()
|
||||
return !busy
|
||||
})
|
||||
text := systemTexts(t, c)[0]
|
||||
if !strings.Contains(text, "检索") || !strings.Contains(text, "<system-reminder>") || !strings.Contains(text, "后台结果") {
|
||||
t.Fatalf("notice = %s", text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBackgroundQueuesUntilIdleThenMerges(t *testing.T) {
|
||||
runner := &gateRunner{parts: []core.Part{{Type: core.PartText, Text: "完成了"}}}
|
||||
c, st := newSpawnEnv(t, runner)
|
||||
c.mgr.runner = noopRunner{}
|
||||
if err := c.mgr.claimConversation(c.run.ConversationID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := c.SpawnBackground("agent.sub.general", "甲", "任务甲", ""); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := c.SpawnBackground("agent.sub.general", "乙", "任务乙", ""); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waitUntil(t, func() bool {
|
||||
c.mgr.noticeMu.Lock()
|
||||
defer c.mgr.noticeMu.Unlock()
|
||||
return len(c.mgr.notices[c.run.ConversationID]) == 2
|
||||
})
|
||||
if len(systemTexts(t, c)) != 0 {
|
||||
t.Fatal("notice delivered while conversation was busy")
|
||||
}
|
||||
c.mgr.releaseConversation(c.run.ConversationID)
|
||||
c.mgr.deliverNotices(c.run.ConversationID)
|
||||
waitUntil(t, func() bool {
|
||||
if len(systemTexts(t, c)) != 1 {
|
||||
return false
|
||||
}
|
||||
c.mgr.convMu.Lock()
|
||||
busy := c.mgr.convActive[c.run.ConversationID]
|
||||
c.mgr.convMu.Unlock()
|
||||
return !busy
|
||||
})
|
||||
text := systemTexts(t, c)[0]
|
||||
if !strings.Contains(text, "甲") || !strings.Contains(text, "乙") {
|
||||
t.Fatalf("merged notice = %s", text)
|
||||
}
|
||||
runs, err := st.ListRunsByConversation(c.run.ConversationID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
conv := 0
|
||||
for _, r := range runs {
|
||||
if r.Kind == core.RunConversation {
|
||||
conv++
|
||||
}
|
||||
}
|
||||
if conv != 1 {
|
||||
t.Fatalf("conversation runs = %d, want 1", conv)
|
||||
}
|
||||
hist := c.History()
|
||||
seen := false
|
||||
for _, m := range hist {
|
||||
if m.Role == core.RoleSystem {
|
||||
seen = true
|
||||
}
|
||||
}
|
||||
if !seen {
|
||||
t.Fatal("history dropped the notice")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBackgroundChildCannotDelegate(t *testing.T) {
|
||||
got := make(chan error, 1)
|
||||
runner := &gateRunner{spawn: func(rc runtime.RunCtx) error {
|
||||
_, err := rc.(interface {
|
||||
SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error)
|
||||
}).SpawnAgent("agent.sub.general", "再派", "继续", "")
|
||||
got <- err
|
||||
return nil
|
||||
}}
|
||||
c, _ := newSpawnEnv(t, runner)
|
||||
c.mgr.runner = noopRunner{}
|
||||
if _, err := c.SpawnBackground("agent.sub.general", "后台", "任务", ""); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() {
|
||||
waitUntil(t, func() bool {
|
||||
c.mgr.convMu.Lock()
|
||||
busy := c.mgr.convActive[c.run.ConversationID]
|
||||
c.mgr.convMu.Unlock()
|
||||
return !busy && len(systemTexts(t, c)) == 1
|
||||
})
|
||||
}()
|
||||
select {
|
||||
case err := <-got:
|
||||
if err == nil || !strings.Contains(err.Error(), "子代理不能再派发子代理") {
|
||||
t.Fatalf("err = %v", err)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("child did not try to delegate")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBackgroundCap(t *testing.T) {
|
||||
release := make(chan struct{})
|
||||
runner := &gateRunner{release: release}
|
||||
c, _ := newSpawnEnv(t, runner)
|
||||
c.mgr.runner = noopRunner{}
|
||||
if err := c.mgr.claimConversation(c.run.ConversationID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for i := 0; i < 3; i++ {
|
||||
if _, err := c.SpawnBackground("agent.sub.general", "n", "t", ""); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if _, err := c.SpawnBackground("agent.sub.general", "overflow", "t", ""); err == nil || !strings.Contains(err.Error(), "3") {
|
||||
t.Fatalf("cap err = %v", err)
|
||||
}
|
||||
close(release)
|
||||
waitUntil(t, func() bool {
|
||||
c.mgr.noticeMu.Lock()
|
||||
defer c.mgr.noticeMu.Unlock()
|
||||
return c.mgr.bgInFlight[c.run.ConversationID] == 0
|
||||
})
|
||||
}
|
||||
+204
-17
@@ -50,6 +50,21 @@ type Manager struct {
|
||||
// 会话内串行(设计并发策略):一个会话同时至多一个在途 conversation run。
|
||||
convMu sync.Mutex
|
||||
convActive map[string]bool // conversationID → 有在途 run
|
||||
|
||||
// 后台子代理的完成通知只留在内存里。重启后未投递的通知丢失。
|
||||
noticeMu sync.Mutex
|
||||
notices map[string][]bgNotice
|
||||
bgInFlight map[string]int
|
||||
delivering map[string]bool
|
||||
}
|
||||
|
||||
const maxBackgroundChildren = 3
|
||||
|
||||
// bgNotice 是一条尚未投递给父会话的后台子代理结果。
|
||||
type bgNotice struct {
|
||||
owner string
|
||||
name string
|
||||
payload core.AgentResultPayload
|
||||
}
|
||||
|
||||
// ErrConversationBusy 表示会话已有在途 run(HTTP 层映射 409)。
|
||||
@@ -77,6 +92,7 @@ func New(st *store.Store, b *bus.Bus, reg *tools.Registry, p *profiles.Service,
|
||||
st: st, bus: b, registry: reg, profiles: p, settings: s,
|
||||
runner: runner, subRunner: subRunner, mock: mockRunner, mockSub: mockSubRunner,
|
||||
active: map[string]*runHandle{}, convActive: map[string]bool{},
|
||||
notices: map[string][]bgNotice{}, bgInFlight: map[string]int{}, delivering: map[string]bool{},
|
||||
events: newEventBuffer(st, b),
|
||||
}
|
||||
}
|
||||
@@ -299,6 +315,10 @@ func (c *runCtx) SpawnAgent(agentID, name, task, hint string) (core.AgentResultP
|
||||
return c.mgr.spawnSubagent(c.ctx, c.run, agentID, name, task, hint)
|
||||
}
|
||||
|
||||
func (c *runCtx) SpawnBackground(agentID, name, task, hint string) (core.ID, error) {
|
||||
return c.mgr.spawnBackground(c.run, agentID, name, task, hint)
|
||||
}
|
||||
|
||||
func (c *runCtx) SubagentID() string { return c.subAgent }
|
||||
|
||||
func (c *runCtx) Subhint() string { return c.subHint }
|
||||
@@ -688,15 +708,26 @@ func selectConversationRunner(m *Manager, eff core.EffectiveSettings) runtime.Ru
|
||||
// attachments 为已解析的附件文档(server 层经 ingest 处理)。
|
||||
// inputs 是同一批附件的原件,落到会话工作区 inputs/,供模板和仿制使用。
|
||||
func (m *Manager) StartConversationRun(conv core.Conversation, userMessage core.Message, attachments []core.ParsedDocument, inputs []WorkspaceFile) (core.Run, error) {
|
||||
// 会话内串行(设计策略落地):在途 run(含 HITL 挂起)未终态前拒绝新 run。
|
||||
m.convMu.Lock()
|
||||
if m.convActive[conv.ID] {
|
||||
m.convMu.Unlock()
|
||||
return core.Run{}, ErrConversationBusy
|
||||
if err := m.claimConversation(conv.ID); err != nil {
|
||||
return core.Run{}, err
|
||||
}
|
||||
m.convActive[conv.ID] = true
|
||||
m.convMu.Unlock()
|
||||
return m.startClaimedRun(conv, userMessage, attachments, inputs)
|
||||
}
|
||||
|
||||
func (m *Manager) claimConversation(conversationID string) error {
|
||||
m.convMu.Lock()
|
||||
defer m.convMu.Unlock()
|
||||
if m.convActive == nil {
|
||||
m.convActive = map[string]bool{}
|
||||
}
|
||||
if m.convActive[conversationID] {
|
||||
return ErrConversationBusy
|
||||
}
|
||||
m.convActive[conversationID] = true
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) startClaimedRun(conv core.Conversation, userMessage core.Message, attachments []core.ParsedDocument, inputs []WorkspaceFile) (core.Run, error) {
|
||||
prof, _ := m.profiles.Get(conv.Owner, conv.ProfileID)
|
||||
eff := m.settings.Effective(conv.Owner, conv.ProjectID)
|
||||
run := core.Run{
|
||||
@@ -734,6 +765,7 @@ func (m *Manager) execute(rc *runCtx, runner runtime.Runner) {
|
||||
m.mu.Lock()
|
||||
delete(m.active, rc.run.ID)
|
||||
m.mu.Unlock()
|
||||
m.deliverNotices(rc.run.ConversationID)
|
||||
}()
|
||||
m.setRunStatus(&rc.run, core.RunRunning, core.PhaseThinking)
|
||||
m.emit(rc.run.ConversationID, rc.run.ID, core.EvRunPhase, core.RunPhasePayload{Phase: core.PhaseThinking})
|
||||
@@ -968,8 +1000,18 @@ func (m *Manager) ResolveHITL(owner, cardID string, resp core.HITLResponse) (piv
|
||||
// ---- 子代理(任务树)----
|
||||
|
||||
func (m *Manager) spawnSubagent(parentCtx context.Context, parent core.Run, agentID, name, task, hint string) (core.AgentResultPayload, error) {
|
||||
rc, runner, err := m.startChild(parentCtx, parent, agentID, name, task, hint, false)
|
||||
if err != nil {
|
||||
return core.AgentResultPayload{}, err
|
||||
}
|
||||
return m.finishChild(rc, runner, true)
|
||||
}
|
||||
|
||||
// startChild 做深度检查、建 run、发 spawn 事件。detach 时子 run 使用独立 ctx,
|
||||
// 父 run 结束不会取消它;cancel 仍登记在 active 里,CancelRun 和停机中断照常生效。
|
||||
func (m *Manager) startChild(parentCtx context.Context, parent core.Run, agentID, name, task, hint string, detach bool) (*runCtx, runtime.Runner, error) {
|
||||
if parent.Kind == core.RunSubagent {
|
||||
return core.AgentResultPayload{}, fmt.Errorf("子代理不能再派发子代理")
|
||||
return nil, nil, fmt.Errorf("子代理不能再派发子代理")
|
||||
}
|
||||
if strings.TrimSpace(agentID) == "" {
|
||||
agentID = "agent.sub.general"
|
||||
@@ -980,11 +1022,15 @@ func (m *Manager) spawnSubagent(parentCtx context.Context, parent core.Run, agen
|
||||
Status: core.RunPending, Phase: core.PhaseThinking, AgentName: name,
|
||||
}
|
||||
if err := m.st.CreateRun(&child); err != nil {
|
||||
return core.AgentResultPayload{}, err
|
||||
return nil, nil, err
|
||||
}
|
||||
m.emit(parent.ConversationID, parent.ID, core.EvAgentSpawn, core.AgentSpawnPayload{AgentRun: child.ID, Name: name, Task: task})
|
||||
|
||||
ctx, cancel := context.WithCancel(parentCtx)
|
||||
base := parentCtx
|
||||
if detach {
|
||||
base = context.Background()
|
||||
}
|
||||
ctx, cancel := context.WithCancel(base)
|
||||
h := &runHandle{cancel: cancel, hitlCh: make(chan hitlAnswer, 1), started: time.Now()}
|
||||
m.mu.Lock()
|
||||
m.active[child.ID] = h
|
||||
@@ -999,6 +1045,12 @@ func (m *Manager) spawnSubagent(parentCtx context.Context, parent core.Run, agen
|
||||
if eff.GetBool("runtime.mock", false) || runner == nil {
|
||||
runner = m.mockSub
|
||||
}
|
||||
return rc, runner, nil
|
||||
}
|
||||
|
||||
// finishChild 执行子 run、落库并汇总。emitResult 为假时不发 agent.result,
|
||||
// 留给后台投递成功之后再发。
|
||||
func (m *Manager) finishChild(rc *runCtx, runner runtime.Runner, emitResult bool) (core.AgentResultPayload, error) {
|
||||
execErr := func() (err error) {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
@@ -1008,12 +1060,12 @@ func (m *Manager) spawnSubagent(parentCtx context.Context, parent core.Run, agen
|
||||
return runner.Execute(rc)
|
||||
}()
|
||||
|
||||
usage := core.Usage{DurationMs: float64(time.Since(h.started).Milliseconds())}
|
||||
usage := core.Usage{DurationMs: float64(time.Since(rc.h.started).Milliseconds())}
|
||||
status := core.RunCompleted
|
||||
summary := "完成"
|
||||
payload := core.AgentResultPayload{AgentRun: child.ID, Status: status, Summary: summary}
|
||||
payload := core.AgentResultPayload{AgentRun: rc.run.ID, Status: status, Summary: summary}
|
||||
if execErr != nil {
|
||||
if ctx.Err() != nil {
|
||||
if rc.ctx.Err() != nil {
|
||||
status = core.RunCancelled
|
||||
payload.Status = status
|
||||
payload.Summary = "已取消"
|
||||
@@ -1024,16 +1076,18 @@ func (m *Manager) spawnSubagent(parentCtx context.Context, parent core.Run, agen
|
||||
}
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
child := rc.run
|
||||
child.EndedAt = &now
|
||||
child.Status = status
|
||||
child.Phase = core.PhaseDone
|
||||
child.Usage = &usage
|
||||
m.st.UpdateRun(&child)
|
||||
m.emit(child.ConversationID, child.ID, core.EvAgentResult, payload)
|
||||
if emitResult {
|
||||
m.emit(child.ConversationID, child.ID, core.EvAgentResult, payload)
|
||||
}
|
||||
|
||||
// 子代理中间过程落库(AgentTracker 弹窗的权威数据源;消息树带 run_id 归属)。
|
||||
m.mu.Lock()
|
||||
subParts := append([]core.Part{}, h.assistant...)
|
||||
subParts := append([]core.Part{}, rc.h.assistant...)
|
||||
m.mu.Unlock()
|
||||
fullText := partsText(subParts)
|
||||
if len(subParts) > 0 {
|
||||
@@ -1053,11 +1107,144 @@ func (m *Manager) spawnSubagent(parentCtx context.Context, parent core.Run, agen
|
||||
delete(m.active, child.ID)
|
||||
m.mu.Unlock()
|
||||
if status != core.RunCompleted {
|
||||
return payload, fmt.Errorf("subagent %s: %s", name, payload.Summary)
|
||||
return payload, fmt.Errorf("subagent %s: %s", child.AgentName, payload.Summary)
|
||||
}
|
||||
return payload, nil
|
||||
}
|
||||
|
||||
func (m *Manager) spawnBackground(parent core.Run, agentID, name, task, hint string) (core.ID, error) {
|
||||
if parent.Kind == core.RunSubagent {
|
||||
return "", fmt.Errorf("子代理不能再派发子代理")
|
||||
}
|
||||
m.noticeMu.Lock()
|
||||
if m.bgInFlight == nil {
|
||||
m.bgInFlight = map[string]int{}
|
||||
}
|
||||
if m.bgInFlight[parent.ConversationID] >= maxBackgroundChildren {
|
||||
m.noticeMu.Unlock()
|
||||
return "", fmt.Errorf("本会话已有 %d 个后台子代理在运行", maxBackgroundChildren)
|
||||
}
|
||||
m.bgInFlight[parent.ConversationID]++
|
||||
m.noticeMu.Unlock()
|
||||
|
||||
rc, runner, err := m.startChild(context.Background(), parent, agentID, name, task, hint, true)
|
||||
if err != nil {
|
||||
m.decBackground(parent.ConversationID)
|
||||
return "", err
|
||||
}
|
||||
go func() {
|
||||
defer m.decBackground(parent.ConversationID)
|
||||
payload, _ := m.finishChild(rc, runner, false)
|
||||
m.enqueueNotice(parent.ConversationID, bgNotice{owner: parent.Owner, name: name, payload: payload})
|
||||
m.deliverNotices(parent.ConversationID)
|
||||
}()
|
||||
return rc.run.ID, nil
|
||||
}
|
||||
|
||||
func (m *Manager) decBackground(convID string) {
|
||||
m.noticeMu.Lock()
|
||||
defer m.noticeMu.Unlock()
|
||||
if m.bgInFlight == nil {
|
||||
return
|
||||
}
|
||||
if m.bgInFlight[convID] > 0 {
|
||||
m.bgInFlight[convID]--
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) enqueueNotice(convID string, n bgNotice) {
|
||||
m.noticeMu.Lock()
|
||||
defer m.noticeMu.Unlock()
|
||||
if m.notices == nil {
|
||||
m.notices = map[string][]bgNotice{}
|
||||
}
|
||||
m.notices[convID] = append(m.notices[convID], n)
|
||||
}
|
||||
|
||||
// deliverNotices 在会话空闲时把全部待投递结果合并成一条 system 消息,并启动新的父回合。
|
||||
// 会话忙、或启动时发现忙,就把结果放回队列。
|
||||
func (m *Manager) deliverNotices(convID string) {
|
||||
m.convMu.Lock()
|
||||
busy := m.convActive != nil && m.convActive[convID]
|
||||
m.convMu.Unlock()
|
||||
if busy {
|
||||
return
|
||||
}
|
||||
|
||||
m.noticeMu.Lock()
|
||||
if m.delivering == nil {
|
||||
m.delivering = map[string]bool{}
|
||||
}
|
||||
if m.delivering[convID] {
|
||||
m.noticeMu.Unlock()
|
||||
return
|
||||
}
|
||||
items := append([]bgNotice{}, m.notices[convID]...)
|
||||
if len(items) == 0 {
|
||||
m.noticeMu.Unlock()
|
||||
return
|
||||
}
|
||||
delete(m.notices, convID)
|
||||
m.delivering[convID] = true
|
||||
m.noticeMu.Unlock()
|
||||
|
||||
defer func() {
|
||||
m.noticeMu.Lock()
|
||||
delete(m.delivering, convID)
|
||||
more := len(m.notices[convID]) > 0
|
||||
m.noticeMu.Unlock()
|
||||
if more {
|
||||
m.deliverNotices(convID)
|
||||
}
|
||||
}()
|
||||
|
||||
if err := m.startNoticeRun(convID, items); err != nil {
|
||||
m.noticeMu.Lock()
|
||||
if m.notices == nil {
|
||||
m.notices = map[string][]bgNotice{}
|
||||
}
|
||||
m.notices[convID] = append(items, m.notices[convID]...)
|
||||
m.noticeMu.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) startNoticeRun(convID string, items []bgNotice) error {
|
||||
if len(items) == 0 {
|
||||
return nil
|
||||
}
|
||||
if err := m.claimConversation(convID); err != nil {
|
||||
return err
|
||||
}
|
||||
owner := items[0].owner
|
||||
conv, err := m.st.GetConversation(owner, convID)
|
||||
if err != nil {
|
||||
m.releaseConversation(convID)
|
||||
return err
|
||||
}
|
||||
var b strings.Builder
|
||||
b.WriteString("<system-reminder>\n")
|
||||
for _, n := range items {
|
||||
fmt.Fprintf(&b, "后台子代理「%s」已结束。\n状态:%s\n%s\n", n.name, n.payload.Status, strings.TrimSpace(n.payload.Summary))
|
||||
}
|
||||
b.WriteString("</system-reminder>")
|
||||
msg := core.Message{
|
||||
ConversationID: convID, Role: core.RoleSystem,
|
||||
Parts: []core.Part{{Type: core.PartText, Text: b.String()}},
|
||||
}
|
||||
if err := m.st.AppendMessage(&msg); err != nil {
|
||||
m.releaseConversation(convID)
|
||||
return err
|
||||
}
|
||||
if _, err := m.startClaimedRun(conv, msg, nil, nil); err != nil {
|
||||
m.releaseConversation(convID)
|
||||
return err
|
||||
}
|
||||
for _, n := range items {
|
||||
m.emit(convID, n.payload.AgentRun, core.EvAgentResult, n.payload)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// InterruptRunning 服务停机前把在途 run 标记为 interrupted(4.2 恢复策略)。
|
||||
func (m *Manager) InterruptRunning() {
|
||||
m.mu.Lock()
|
||||
|
||||
+21
-16
@@ -47,6 +47,7 @@ func durationMs(start, end time.Time) float64 {
|
||||
|
||||
const (
|
||||
toolDelegate = "delegate_task"
|
||||
toolReview = "review_findings"
|
||||
toolPlan = "update_plan"
|
||||
toolTodos = "update_todos"
|
||||
toolSquad = "squad_orchestrate"
|
||||
@@ -209,13 +210,7 @@ func (l *LLMRunner) Execute(rc RunCtx) error {
|
||||
|
||||
// 组装初始上下文。
|
||||
hist := rc.History()
|
||||
var currentUser core.Message
|
||||
for i := len(hist) - 1; i >= 0; i-- {
|
||||
if hist[i].Role == core.RoleUser {
|
||||
currentUser = hist[i]
|
||||
break
|
||||
}
|
||||
}
|
||||
currentUser, _ := lastAddressed(hist)
|
||||
var sys []agentllm.Message
|
||||
var promptObs []string
|
||||
if l.Prompts != nil {
|
||||
@@ -240,7 +235,8 @@ func (l *LLMRunner) Execute(rc RunCtx) error {
|
||||
decls = append(decls, agentllm.ToolDecl{Name: toolSquad, Description: l.toolDesc(prof.Owner, toolSquad, "用小队编排引擎执行多步骤目标:规则规划器拆解为 梳理→专项→成文→终审 步骤并逐步执行(每步一个子代理),挂起决策会以提问卡片征求用户"), Parameters: json.RawMessage(`{"type":"object","properties":{"goal":{"type":"string","description":"要完成的整体目标描述"}},"required":["goal"]}`)})
|
||||
}
|
||||
decls = append(decls,
|
||||
agentllm.ToolDecl{Name: toolDelegate, Description: l.toolDesc(prof.Owner, toolDelegate, "派生一个子代理执行独立子任务(如检索、撰写),返回其结果摘要。模型和工具由子代理 profile 决定,调用方不能指定模型"), Parameters: json.RawMessage(`{"type":"object","properties":{"agent":{"type":"string","enum":["general","explore","research","plan","verify","writer"],"description":"子代理 profile,缺省 general"},"name":{"type":"string","description":"显示名"},"task":{"type":"string","description":"子任务描述"},"hint":{"type":"string","description":"执行提示"}},"required":["name","task"]}`)},
|
||||
agentllm.ToolDecl{Name: toolDelegate, Description: l.toolDesc(prof.Owner, toolDelegate, "派生一个子代理执行独立子任务(如检索、撰写),返回其结果摘要。模型和工具由子代理 profile 决定,调用方不能指定模型。background 为真时立即返回,子代理完成后系统会自动发起一轮汇报"), Parameters: json.RawMessage(`{"type":"object","properties":{"agent":{"type":"string","enum":["general","explore","research","plan","verify","writer"],"description":"子代理 profile,缺省 general"},"name":{"type":"string","description":"显示名"},"task":{"type":"string","description":"子任务描述"},"hint":{"type":"string","description":"执行提示"},"background":{"type":"boolean","description":"为真时在后台运行,当前回合不等待结果"}},"required":["name","task"]}`)},
|
||||
agentllm.ToolDecl{Name: toolReview, Description: l.toolDesc(prof.Owner, toolReview, "对一段材料做评审:多个审查子代理并行找问题,再逐条独立验证,只保留验证通过的问题"), Parameters: json.RawMessage(`{"type":"object","properties":{"target":{"type":"string","description":"要评审的文本或工件"},"focus":{"type":"string","description":"评审重点"},"reviewers":{"type":"integer","minimum":1,"maximum":3,"description":"并行审查人数,1 到 3,缺省 1"}},"required":["target"]}`)},
|
||||
agentllm.ToolDecl{Name: toolPlan, Description: l.toolDesc(prof.Owner, toolPlan, "创建或更新当前会话的计划(步骤列表)"), Parameters: json.RawMessage(`{"type":"object","properties":{"title":{"type":"string"},"steps":{"type":"array","items":{"type":"object","properties":{"id":{"type":"string"},"text":{"type":"string"},"status":{"type":"string","enum":["pending","in_progress","done","skipped"]}},"required":["id","text","status"]}}},"required":["title","steps"]}`)},
|
||||
agentllm.ToolDecl{Name: toolTodos, Description: l.toolDesc(prof.Owner, toolTodos, "更新当前会话的待办清单"), Parameters: json.RawMessage(`{"type":"object","properties":{"items":{"type":"array","items":{"type":"object","properties":{"id":{"type":"string"},"text":{"type":"string"},"status":{"type":"string","enum":["pending","in_progress","done"]}},"required":["id","text","status"]}}},"required":["items"]}`)},
|
||||
)
|
||||
@@ -300,19 +296,28 @@ type SubLLMRunner struct {
|
||||
}
|
||||
|
||||
func currentUserText(rc RunCtx) string {
|
||||
hist := rc.History()
|
||||
for i := len(hist) - 1; i >= 0; i-- {
|
||||
if hist[i].Role == core.RoleUser {
|
||||
for _, p := range hist[i].Parts {
|
||||
if p.Type == core.PartText {
|
||||
return p.Text
|
||||
}
|
||||
}
|
||||
msg, ok := lastAddressed(rc.History())
|
||||
if !ok {
|
||||
return ""
|
||||
}
|
||||
for _, p := range msg.Parts {
|
||||
if p.Type == core.PartText {
|
||||
return p.Text
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// lastAddressed 取最后一条用户消息或系统通知,避免通知回合重答更早的问题。
|
||||
func lastAddressed(hist []core.Message) (core.Message, bool) {
|
||||
for i := len(hist) - 1; i >= 0; i-- {
|
||||
if hist[i].Role == core.RoleUser || hist[i].Role == core.RoleSystem {
|
||||
return hist[i], true
|
||||
}
|
||||
}
|
||||
return core.Message{}, false
|
||||
}
|
||||
|
||||
func toolList(rc RunCtx) []core.ToolDesc { return rc.Tools() }
|
||||
|
||||
// squadToolEnabled 在小队已经是对话入口时关掉 squad_orchestrate,避免再套一层编排。
|
||||
|
||||
@@ -80,6 +80,15 @@ func (t directTool) Execute(ctx context.Context, args json.RawMessage) (agenttoo
|
||||
func (l *LLMRunner) runDirect(rc RunCtx, stream agentllm.StreamFunc, model string, msgs []agentllm.Message, decls []agentllm.ToolDecl, streamOpts agentllm.StreamOptions, cardflow bool, eff core.EffectiveSettings, maxTurns, maxTools int) error {
|
||||
st := &directRun{l: l, rc: rc, eff: eff, cardflow: cardflow}
|
||||
st.loadChecklist()
|
||||
if msg, ok := lastAddressed(rc.History()); ok && msg.Role == core.RoleSystem {
|
||||
var text string
|
||||
for _, p := range msg.Parts {
|
||||
if p.Type == core.PartText {
|
||||
text += p.Text
|
||||
}
|
||||
}
|
||||
st.pinSubagentArtifacts(text)
|
||||
}
|
||||
defer func() {
|
||||
if sink, ok := rc.(interface{ AddUsage(core.Usage) }); ok {
|
||||
sink.AddUsage(st.usageTotal)
|
||||
@@ -343,8 +352,24 @@ func (st *directRun) runTool(name, args string) agenttools.Result {
|
||||
out = summary
|
||||
}
|
||||
case toolDelegate:
|
||||
var a struct{ Agent, Name, Task, Hint string }
|
||||
var a struct {
|
||||
Agent, Name, Task, Hint string
|
||||
Background bool
|
||||
}
|
||||
_ = json.Unmarshal([]byte(call.Arguments), &a)
|
||||
if a.Background {
|
||||
if sp, ok := st.rc.(interface {
|
||||
SpawnBackground(agentID, name, task, hint string) (core.ID, error)
|
||||
}); ok {
|
||||
id, serr := sp.SpawnBackground(subagentProfileID(a.Agent), orDefault(a.Name, "子代理"), a.Task, a.Hint)
|
||||
if serr != nil {
|
||||
out = "子代理失败:" + serr.Error()
|
||||
} else {
|
||||
out = fmt.Sprintf("已在后台启动 %s,完成后会自动通知", id)
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
var res core.AgentResultPayload
|
||||
var serr error
|
||||
if sp, ok := st.rc.(interface {
|
||||
@@ -360,6 +385,8 @@ func (st *directRun) runTool(name, args string) agenttools.Result {
|
||||
out = orDefault(res.Summary, "子代理完成")
|
||||
st.pinSubagentArtifacts(res.Summary + "\n" + res.Text)
|
||||
}
|
||||
case toolReview:
|
||||
out = st.reviewFindings(call.Arguments)
|
||||
case toolPlan:
|
||||
var p core.PlanSnapshot
|
||||
if uerr := json.Unmarshal([]byte(call.Arguments), &p); uerr == nil {
|
||||
|
||||
@@ -35,6 +35,10 @@ type fakeRunCtx struct {
|
||||
verify bool // runtime.verify;缺省关闭,避免改变既有流调用次数
|
||||
missingArtifact bool // 工具登记了工件,但 Artifact 查不到
|
||||
emptyArtifact bool // Artifact 查到了,内容为空
|
||||
history []core.Message // 非 nil 时覆盖默认历史
|
||||
bgID string
|
||||
bgErr error
|
||||
bgCalls int
|
||||
}
|
||||
|
||||
func (f *fakeRunCtx) Context() context.Context { return f.ctx }
|
||||
@@ -76,8 +80,21 @@ func (f *fakeRunCtx) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult
|
||||
return res
|
||||
}
|
||||
func (f *fakeRunCtx) History() []core.Message {
|
||||
if f.history != nil {
|
||||
return f.history
|
||||
}
|
||||
return []core.Message{{ID: "m1", Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "hi"}}}}
|
||||
}
|
||||
func (f *fakeRunCtx) SpawnBackground(agentID, name, task, hint string) (core.ID, error) {
|
||||
f.bgCalls++
|
||||
if f.bgErr != nil {
|
||||
return "", f.bgErr
|
||||
}
|
||||
if f.bgID == "" {
|
||||
f.bgID = "run_bg"
|
||||
}
|
||||
return f.bgID, nil
|
||||
}
|
||||
func (f *fakeRunCtx) Attachments() []core.ParsedDocument { return nil }
|
||||
func (f *fakeRunCtx) Tools() []core.ToolDesc { return nil }
|
||||
func (f *fakeRunCtx) Subtask() string { return "" }
|
||||
|
||||
@@ -0,0 +1,187 @@
|
||||
package runtime
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
core "agent-system-core"
|
||||
)
|
||||
|
||||
const reviewProfileID = "agent.sub.review"
|
||||
const verifyProfileID = "agent.sub.verify"
|
||||
|
||||
type reviewFinding struct {
|
||||
Location string `json:"location"`
|
||||
Severity string `json:"severity"`
|
||||
Issue string `json:"issue"`
|
||||
Evidence string `json:"evidence"`
|
||||
}
|
||||
|
||||
func (st *directRun) reviewFindings(args string) string {
|
||||
var a struct {
|
||||
Target string
|
||||
Focus string
|
||||
Reviewers int
|
||||
}
|
||||
if err := json.Unmarshal([]byte(args), &a); err != nil {
|
||||
return "评审参数错误: " + err.Error()
|
||||
}
|
||||
sp, ok := st.rc.(interface {
|
||||
SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error)
|
||||
})
|
||||
if !ok {
|
||||
return "当前运行不能派发评审子代理"
|
||||
}
|
||||
return runReview(sp.SpawnAgent, a.Target, a.Focus, a.Reviewers)
|
||||
}
|
||||
|
||||
func runReview(spawn func(agentID, name, task, hint string) (core.AgentResultPayload, error), target, focus string, reviewers int) string {
|
||||
if reviewers < 1 {
|
||||
reviewers = 1
|
||||
}
|
||||
if reviewers > 3 {
|
||||
reviewers = 3
|
||||
}
|
||||
found := make([][]reviewFinding, reviewers)
|
||||
parallel(reviewers, func(i int) {
|
||||
task := fmt.Sprintf("评审以下材料,重点是%s。只输出 JSON 数组,元素含 location、severity、issue、evidence。\n\n%s", orDefault(focus, "正确性与遗漏"), target)
|
||||
res, err := spawn(reviewProfileID, fmt.Sprintf("审查%d", i+1), task, focus)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
found[i] = parseFindings(res.Text + "\n" + res.Summary)
|
||||
})
|
||||
unique := dedupeFindings(found, 10)
|
||||
kept := make([]reviewFinding, len(unique))
|
||||
pass := make([]bool, len(unique))
|
||||
parallel(len(unique), func(i int) {
|
||||
f := unique[i]
|
||||
body, _ := json.Marshal(f)
|
||||
task := fmt.Sprintf("独立核实这条问题是否成立。材料如下:\n%s\n\n问题:%s\n只在结尾给出一行 VERDICT: PASS、FAIL 或 PARTIAL。", target, body)
|
||||
res, err := spawn(verifyProfileID, "核实", task, f.Location)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if reviewVerdict(res.Text+"\n"+res.Summary) == "PASS" {
|
||||
kept[i] = f
|
||||
pass[i] = true
|
||||
}
|
||||
})
|
||||
var verified []reviewFinding
|
||||
for i, ok := range pass {
|
||||
if ok {
|
||||
verified = append(verified, kept[i])
|
||||
}
|
||||
}
|
||||
sort.SliceStable(verified, func(i, j int) bool {
|
||||
return severityRank(verified[i].Severity) < severityRank(verified[j].Severity)
|
||||
})
|
||||
dropped := len(unique) - len(verified)
|
||||
var b strings.Builder
|
||||
if len(verified) == 0 {
|
||||
b.WriteString("没有验证通过的问题。\n")
|
||||
}
|
||||
for _, f := range verified {
|
||||
fmt.Fprintf(&b, "- [%s] %s:%s\n 证据:%s\n", orDefault(f.Severity, "medium"), orDefault(f.Location, "未定位"), f.Issue, f.Evidence)
|
||||
}
|
||||
fmt.Fprintf(&b, "已丢弃 %d 条未验证\n", dropped)
|
||||
return b.String()
|
||||
}
|
||||
|
||||
func parallel(n int, fn func(i int)) {
|
||||
if n <= 0 {
|
||||
return
|
||||
}
|
||||
sem := make(chan struct{}, 3)
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < n; i++ {
|
||||
wg.Add(1)
|
||||
go func(i int) {
|
||||
defer wg.Done()
|
||||
sem <- struct{}{}
|
||||
defer func() { <-sem }()
|
||||
fn(i)
|
||||
}(i)
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
func parseFindings(text string) []reviewFinding {
|
||||
start := strings.Index(text, "[")
|
||||
end := strings.LastIndex(text, "]")
|
||||
if start < 0 || end <= start {
|
||||
return nil
|
||||
}
|
||||
var out []reviewFinding
|
||||
if err := json.Unmarshal([]byte(text[start:end+1]), &out); err != nil {
|
||||
return nil
|
||||
}
|
||||
var kept []reviewFinding
|
||||
for _, f := range out {
|
||||
if strings.TrimSpace(f.Issue) == "" {
|
||||
continue
|
||||
}
|
||||
kept = append(kept, f)
|
||||
}
|
||||
return kept
|
||||
}
|
||||
|
||||
func dedupeFindings(groups [][]reviewFinding, limit int) []reviewFinding {
|
||||
seen := map[string]bool{}
|
||||
var out []reviewFinding
|
||||
for _, group := range groups {
|
||||
for _, f := range group {
|
||||
key := strings.TrimSpace(f.Location) + "\n" + strings.TrimSpace(f.Issue)
|
||||
if seen[key] {
|
||||
continue
|
||||
}
|
||||
seen[key] = true
|
||||
out = append(out, f)
|
||||
if len(out) >= limit {
|
||||
return out
|
||||
}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func reviewVerdict(text string) string {
|
||||
lines := strings.Split(text, "\n")
|
||||
for i := len(lines) - 1; i >= 0; i-- {
|
||||
line := strings.TrimSpace(lines[i])
|
||||
if line == "" || !strings.HasPrefix(strings.ToUpper(line), "VERDICT:") {
|
||||
continue
|
||||
}
|
||||
v := strings.TrimSpace(line[len("VERDICT:"):])
|
||||
v = strings.ToUpper(v)
|
||||
switch {
|
||||
case strings.HasPrefix(v, "PASS"):
|
||||
return "PASS"
|
||||
case strings.HasPrefix(v, "FAIL"):
|
||||
return "FAIL"
|
||||
case strings.HasPrefix(v, "PARTIAL"):
|
||||
return "PARTIAL"
|
||||
default:
|
||||
return ""
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func severityRank(s string) int {
|
||||
switch strings.ToLower(strings.TrimSpace(s)) {
|
||||
case "critical":
|
||||
return 0
|
||||
case "high":
|
||||
return 1
|
||||
case "medium":
|
||||
return 2
|
||||
case "low":
|
||||
return 3
|
||||
default:
|
||||
return 4
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,108 @@
|
||||
package runtime
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
agentllm "agent-native/runtime/llm"
|
||||
core "agent-system-core"
|
||||
)
|
||||
|
||||
func TestReviewKeepsPassAndDropsTheRest(t *testing.T) {
|
||||
release := make(chan struct{})
|
||||
var once sync.Once
|
||||
var mu sync.Mutex
|
||||
var entered int
|
||||
var verifyCalls int
|
||||
spawn := func(agentID, name, task, hint string) (core.AgentResultPayload, error) {
|
||||
if agentID == reviewProfileID {
|
||||
mu.Lock()
|
||||
entered++
|
||||
n := entered
|
||||
mu.Unlock()
|
||||
if n == 3 {
|
||||
once.Do(func() { close(release) })
|
||||
}
|
||||
select {
|
||||
case <-release:
|
||||
case <-time.After(time.Second):
|
||||
return core.AgentResultPayload{}, fmt.Errorf("reviewers did not run together")
|
||||
}
|
||||
switch name {
|
||||
case "审查1":
|
||||
return core.AgentResultPayload{Text: `[{"location":"第2段","severity":"high","issue":"无来源","evidence":"显然"},{"location":"第1段","severity":"low","issue":"错字","evidence":"的地"},{"location":"表1","severity":"medium","issue":"过期数据","evidence":"2020"}]`}, nil
|
||||
case "审查2":
|
||||
return core.AgentResultPayload{Text: `[{"location":"第2段","severity":"high","issue":"无来源","evidence":"显然"},{"location":"附录","severity":"medium","issue":"数字不一致","evidence":"3 与 4"}]`}, nil
|
||||
default:
|
||||
return core.AgentResultPayload{Text: `[{"location":"结尾","severity":"low","issue":"没有结论","evidence":"最后一句"}]`}, nil
|
||||
}
|
||||
}
|
||||
mu.Lock()
|
||||
verifyCalls++
|
||||
mu.Unlock()
|
||||
switch {
|
||||
case strings.Contains(task, "无来源"):
|
||||
return core.AgentResultPayload{Text: "核对原文确实没有来源。\nVERDICT: PASS"}, nil
|
||||
case strings.Contains(task, "错字"):
|
||||
return core.AgentResultPayload{Text: "核对后成立。\nVERDICT: PASS"}, nil
|
||||
case strings.Contains(task, "过期数据"):
|
||||
return core.AgentResultPayload{Text: "数据仍有效。\nVERDICT: FAIL"}, nil
|
||||
default:
|
||||
return core.AgentResultPayload{Text: "看不出对错"}, nil
|
||||
}
|
||||
}
|
||||
out := runReview(spawn, "材料正文", "事实", 3)
|
||||
if !strings.Contains(out, "无来源") || !strings.Contains(out, "错字") || strings.Contains(out, "数字不一致") || strings.Contains(out, "没有结论") || strings.Contains(out, "过期数据") {
|
||||
t.Fatalf("filter = %s", out)
|
||||
}
|
||||
if !strings.Contains(out, "已丢弃 3 条未验证") {
|
||||
t.Fatalf("dropped = %s", out)
|
||||
}
|
||||
if verifyCalls != 5 {
|
||||
t.Fatalf("verify calls = %d, want 5 after dedupe", verifyCalls)
|
||||
}
|
||||
if strings.Index(out, "无来源") > strings.Index(out, "错字") {
|
||||
t.Fatalf("severity order: %s", out)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDelegateBackgroundUsesSpawnBackground(t *testing.T) {
|
||||
rc := &fakeRunCtx{ctx: context.Background(), bgID: "run_bg1"}
|
||||
st := &directRun{rc: rc}
|
||||
out := st.runTool(toolDelegate, `{"name":"检索","task":"查资料","background":true,"agent":"explore"}`)
|
||||
if rc.bgCalls != 1 || !strings.Contains(out.Content, "已在后台启动 run_bg1") {
|
||||
t.Fatalf("calls=%d out=%s", rc.bgCalls, out.Content)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNoticeTurnPinsArtifacts(t *testing.T) {
|
||||
rc := &fakeRunCtx{
|
||||
ctx: context.Background(),
|
||||
prof: core.AgentProfile{Model: core.ModelRoute{Chat: "test-model"}},
|
||||
history: []core.Message{{
|
||||
Role: core.RoleSystem,
|
||||
Parts: []core.Part{{Type: core.PartText, Text: "<system-reminder>\n后台子代理「检索」已结束。\n工件:art_0123456789ab 说明.md\n</system-reminder>"}},
|
||||
}},
|
||||
execArtifact: &core.Artifact{ID: "art_0123456789ab", Name: "说明.md", Kind: "document"},
|
||||
}
|
||||
r := &LLMRunner{Stream: scriptStream([][]agentllm.Event{{
|
||||
{Kind: agentllm.KindTextDelta, Text: "检索已经完成。"},
|
||||
{Kind: agentllm.KindMessageEnd},
|
||||
}}, nil)}
|
||||
if err := r.Execute(rc); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var pinned bool
|
||||
for _, p := range rc.assistant {
|
||||
if p.Type == core.PartArtifactRef && p.Artifact != nil && p.Artifact.ArtifactID == "art_0123456789ab" {
|
||||
pinned = true
|
||||
}
|
||||
}
|
||||
if !pinned {
|
||||
t.Fatalf("parts = %+v", rc.assistant)
|
||||
}
|
||||
}
|
||||
@@ -39,7 +39,7 @@ func (l *SubLLMRunner) Execute(rc RunCtx) error {
|
||||
}
|
||||
spec := fallbackSubagentSpec()
|
||||
if l.Prompts != nil {
|
||||
if got, ok := l.Prompts.SubagentSpec(id); ok {
|
||||
if got, ok := l.Prompts.SubagentSpec(rc.Profile().Owner, id); ok {
|
||||
spec = got
|
||||
}
|
||||
}
|
||||
|
||||
@@ -156,13 +156,38 @@ type stepBackend struct {
|
||||
|
||||
func (b *stepBackend) Name() string { return "agent-system" }
|
||||
|
||||
// profileForRole 把小队角色键映射到子代理 profile。未列出的角色走 general。
|
||||
func profileForRole(role string) string {
|
||||
switch role {
|
||||
case "planner", "organizer":
|
||||
return "agent.sub.plan"
|
||||
case "researcher", "analyst":
|
||||
return "agent.sub.research"
|
||||
case "writer":
|
||||
return "agent.sub.writer"
|
||||
case "reviewer", "leader":
|
||||
return "agent.sub.verify"
|
||||
default:
|
||||
return "agent.sub.general"
|
||||
}
|
||||
}
|
||||
|
||||
func (b *stepBackend) Execute(ctx context.Context, req exec.Request) (*exec.Result, error) {
|
||||
task := strings.TrimSpace(req.Prompt)
|
||||
if task == "" {
|
||||
task = req.Instructions
|
||||
}
|
||||
hint := strings.TrimSpace(req.Instructions)
|
||||
out, err := b.rc.SpawnSubagent(firstNonEmpty(b.role.Name, b.role.Role), task, hint)
|
||||
name := firstNonEmpty(b.role.Name, b.role.Role)
|
||||
var out core.AgentResultPayload
|
||||
var err error
|
||||
if sp, ok := b.rc.(interface {
|
||||
SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error)
|
||||
}); ok {
|
||||
out, err = sp.SpawnAgent(profileForRole(b.role.Role), name, task, hint)
|
||||
} else {
|
||||
out, err = b.rc.SpawnSubagent(name, task, hint)
|
||||
}
|
||||
res := &exec.Result{Success: err == nil, Logs: firstNonEmpty(out.Summary, "子代理完成")}
|
||||
if err != nil {
|
||||
res.ErrorReason = err.Error()
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"agent-native/exec"
|
||||
"agent-native/runtime/subagent"
|
||||
"agent-squad/squad"
|
||||
core "agent-system-core"
|
||||
@@ -115,6 +116,53 @@ func TestSubagentFactoryBridgesSpawnSubagent(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
type agentFakeRc struct {
|
||||
subagentFakeRc
|
||||
agentID string
|
||||
}
|
||||
|
||||
func (f *agentFakeRc) SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) {
|
||||
f.agentID = agentID
|
||||
f.gotName, f.gotTask = name, task
|
||||
return f.payload, f.err
|
||||
}
|
||||
|
||||
func TestStepRoleMapsToProfile(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
cases := []struct {
|
||||
role string
|
||||
want string
|
||||
}{
|
||||
{"planner", "agent.sub.plan"},
|
||||
{"organizer", "agent.sub.plan"},
|
||||
{"researcher", "agent.sub.research"},
|
||||
{"analyst", "agent.sub.research"},
|
||||
{"writer", "agent.sub.writer"},
|
||||
{"reviewer", "agent.sub.verify"},
|
||||
{"leader", "agent.sub.verify"},
|
||||
{"specialist", "agent.sub.general"},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
rc := &agentFakeRc{subagentFakeRc: subagentFakeRc{payload: core.AgentResultPayload{Text: "ok"}}}
|
||||
b := &stepBackend{rc: rc, role: squad.RoleSpec{Role: tc.role, Name: "显示名"}}
|
||||
res, err := b.Execute(context.Background(), exec.Request{Prompt: "做这件事", WorkDir: dir, Targets: []string{"output.md"}})
|
||||
if err != nil || !res.Success {
|
||||
t.Fatalf("%s: %+v %v", tc.role, res, err)
|
||||
}
|
||||
if rc.agentID != tc.want || rc.gotName != "显示名" {
|
||||
t.Fatalf("%s mapped to %s name %s", tc.role, rc.agentID, rc.gotName)
|
||||
}
|
||||
}
|
||||
plain := &subagentFakeRc{payload: core.AgentResultPayload{Text: "ok"}}
|
||||
b := &stepBackend{rc: plain, role: squad.RoleSpec{Role: "writer", Name: "撰稿"}}
|
||||
if _, err := b.Execute(context.Background(), exec.Request{Prompt: "写", WorkDir: dir}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if plain.gotName != "撰稿" {
|
||||
t.Fatalf("fallback name = %s", plain.gotName)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuiltinRolesExplicitExecConfig(t *testing.T) {
|
||||
roles := builtinRoles()
|
||||
byKey := map[string]squad.RoleSpec{}
|
||||
|
||||
Reference in New Issue
Block a user