From 7e7f53bb4120e2b202a7f3ef9414d7ece6cfe5e5 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sun, 4 Oct 2026 11:30:39 +0800 Subject: [PATCH] =?UTF-8?q?fix(orchestration,loop):=20=E5=AD=90=E4=BB=A3?= =?UTF-8?q?=E7=90=86=E7=8E=B0=E5=9C=A8=E7=9C=9F=E7=9A=84=E6=94=B6=E5=88=B0?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=EF=BC=88P0=EF=BC=89+=20=E5=B7=A5=E5=85=B7?= =?UTF-8?q?=E7=99=BD=E5=90=8D=E5=8D=95=20+=20=E8=B6=85=E6=97=B6=E5=8F=AF?= =?UTF-8?q?=E9=85=8D=EF=BC=88=E8=A3=81=E5=86=B3=2002.4.2/02.4.1/02.4.4?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 执行方案阶段 2 · P2-1。**02.4.2 是最伤的一条 P0**：`spawntool` 从不填 `AgentSpec.Context`、 `scheduler` 把 nil 交给 loop ⇒ 子代理模型上下文 = 角色 system + 记忆召回 + **0 条消息**， goal/attach_info/data 一个字都到不了模型（此前测试用 mock adapter 恒答固定文本，测不出）。 - **02.4.2**：新增 `taskContext/taskUserMessage`（`spec.go`）把 **goal 正文 + attach_info + data 渲染成同一条 user 消息**（`[任务目标]/[附加信息]/[任务数据]` 独立段落）；`Spawn()` 设 `Context: taskContext(spec)`；**system 只保留角色指令**，任务绝不进 system。 真断言（捕获请求体的 adapter + 真 loop/agent）：首请求恰好 **1 条 user 消息**、同时含 goal 与 attach_info，且**所有 system 消息不含 goal 文本**。 - **02.4.1**：`AgentSpec.ToolWhitelist`（nil/空 = 继承全量）；schema 增 `tool_whitelist` （顶层 + 每 task 条目，task 未带时继承批次级）；`scheduler.run()` 经 ctx 传递 + `scopedRunner` 派生；`loop/options.go` **纯增量（+75 / −0）**：`WithToolWhitelist`/ `ToolWhitelistFrom` + `(*Loop).WithToolWhitelist` 派生副本（浅拷贝仅替换 `hooks.CallAdmit`， 链式保留原钩子），名单外调用否决为**软结果**回灌模型。 - **02.4.4**：`SUBAGENT_TIMEOUT_SEC`（默认 600）入 `internal/config` + `.env.example`； 优先级 = 按调用 `timeout_seconds`（1s–24h 校验）> `WithSpawnTimeout`（生产由 cfg 填充）> 调度器 `PerAgentTimeout`（`wire*.go` 仍是 10min 常量，未动）；迟到 span 续根由测试锁定。 验证：**全树 `go build`/`go vet`/`go test ./... -count=1` = 58 包 ok / 0 FAIL**； `GOOS=linux/amd64` + `GOOS=darwin/arm64` 交叉编译绿；前端 `tsc --noEmit` + `npm run build` 绿； 本批文件全部 ≤500 行、函数 ≤80 行。 变异验证（四条，均带还原证明）：① `Context` 改回 `cloneMessages(spec.Context)` → 「1 条 user 消息」断言 FAIL（`user messages = 0`，即 P0 症状）；② `WithToolWhitelist` 原样返回 → 显式白名单测试 FAIL（`alpha executed 1 times despite not being whitelisted`）； ③ `subCtx` 改回全局默认 → 3 个超时测试 FAIL；④ `foldSpanFinish` 丢掉 `SpanFinished` → 迟到 span 状态断言 FAIL。 ⚠️ 已知缺口（**登记，未假装已解决**）：① 无真实模型端到端（全部捕获请求体 mock）； ② 白名单**只挡执行、不过滤工具广告面**（模型仍能看见未列工具，调用即被拒并拿到可读原因； 过滤需改 `loop/request.go`）；③ **迟到 span 的"新一轮已开始"窗口未修**（`loop/flow.go` 的 `StartRound` 会整体替换 `graphs[session]` ⇒ 新轮开始后迟到的 span 仍落进新轮图；需按轮键控图， 属后续）；④ per-role 超时覆盖未实现（需 persona 旋钮管道）。 --- .env.example | 5 + cmd/cyrene/subagent_tool.go | 30 +- cmd/cyrene/subagent_tool_test.go | 24 ++ internal/config/config.go | 11 + internal/config/config_test.go | 23 ++ internal/runtime/loop/options.go | 75 ++++ internal/runtime/orchestration/merge.go | 9 + internal/runtime/orchestration/options.go | 35 ++ .../orchestration/orchestration_test.go | 12 +- internal/runtime/orchestration/scheduler.go | 39 +- .../runtime/orchestration/spawntask_test.go | 332 ++++++++++++++++++ internal/runtime/orchestration/spawntool.go | 107 +++++- .../orchestration/spawnwhitelist_test.go | 229 ++++++++++++ internal/runtime/orchestration/spec.go | 80 +++++ internal/runtime/orchestration/subagent.go | 13 +- 15 files changed, 974 insertions(+), 50 deletions(-) create mode 100644 internal/runtime/orchestration/spawntask_test.go create mode 100644 internal/runtime/orchestration/spawnwhitelist_test.go diff --git a/.env.example b/.env.example index 2398926..147d86e 100644 --- a/.env.example +++ b/.env.example @@ -120,6 +120,11 @@ THINK_KIND_WINDOW_SEC=7200 QUIET_START_HOUR=23 QUIET_END_HOUR=7 +# ---- 子代理（orchestration，design/05 §7） ---- +# 单个子代理整次运行的时限（秒）。默认 600 = 10 分钟（裁决 02.4.4：agent_timeout +# 可配置）；模型派发时可用 subagent_spawn 的 timeout_seconds 参数按调用覆盖。 +SUBAGENT_TIMEOUT_SEC=600 + # ---- 推送出口（D1，主动推送的两级投递）---- # 主动消息在 WS/IM 投递之外，异步镜像到「主人自建统一推送服务」（主通道）。 # 该服务不可达/未配时，日志注明「个推(Getui)回退待接入」——当前无 Getui SDK，不伪造投递。 diff --git a/cmd/cyrene/subagent_tool.go b/cmd/cyrene/subagent_tool.go index f34a01b..c4bfe65 100644 --- a/cmd/cyrene/subagent_tool.go +++ b/cmd/cyrene/subagent_tool.go @@ -7,15 +7,27 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration" ) -// defaultSubagentTimeout bounds one sub-agent's whole run. DevDocs/design/05-子代理.md -// §7 requires a timeout and an anti-runaway bound; the scheduler already -// implements both (PerAgentTimeout, Abort/CancelAll), so production must supply a -// value or the per-agent timeout stays inert (the scheduler's zero means "no -// timeout"). The number is deliberately generous — a sub-agent runs up to 8 tool -// rounds like any loop — and is a single knob to revisit when design/05 fixes a -// concrete figure. +// defaultSubagentTimeout bounds one sub-agent's whole run when nothing else +// configures it. DevDocs/design/05-子代理.md §7 requires a timeout and an +// anti-runaway bound; the scheduler already implements both (PerAgentTimeout, +// Abort/CancelAll), so production must supply a value or the per-agent timeout +// stays inert (the scheduler's zero means "no timeout"). The number is +// deliberately generous — a sub-agent runs up to 8 tool rounds like any loop — +// and is the fallback behind SUBAGENT_TIMEOUT_SEC (裁决 02.4.4: 默认 10min， +// 可配置；subagentToolTimeout resolves the effective value). const defaultSubagentTimeout = 10 * time.Minute +// subagentToolTimeout resolves the dispatch tool's default sub-agent timeout: +// SUBAGENT_TIMEOUT_SEC wins when positive, otherwise the 10-minute code default +// applies (design/05 §7 裁决 02.4.4). A per-dispatch timeout_seconds argument +// still overrides this per call at the tool layer. +func (a *app) subagentToolTimeout() time.Duration { + if a.cfg != nil && a.cfg.Subagent.TimeoutSec > 0 { + return time.Duration(a.cfg.Subagent.TimeoutSec) * time.Second + } + return defaultSubagentTimeout +} + // wireSubagentTool registers the sub-agent dispatch tool with the dispatcher, so // the agent can finally reach the orchestration layer at runtime (before this, // (*orchestration.Scheduler).Spawn had no production caller and design/05 was @@ -33,7 +45,9 @@ func (a *app) wireSubagentTool() { log.Warn("sub-agent tool skipped: no scheduler") return } - spawner := orchestration.NewSpawnTool(a.scheduler, orchestration.WithSpawnLogger(a.log)) + spawner := orchestration.NewSpawnTool(a.scheduler, + orchestration.WithSpawnLogger(a.log), + orchestration.WithSpawnTimeout(a.subagentToolTimeout())) if err := a.dispatcher.Register(spawner); err != nil { log.Warn("sub-agent tool register skipped", "error", err) return diff --git a/cmd/cyrene/subagent_tool_test.go b/cmd/cyrene/subagent_tool_test.go index 2b6fae2..ae1fd94 100644 --- a/cmd/cyrene/subagent_tool_test.go +++ b/cmd/cyrene/subagent_tool_test.go @@ -3,9 +3,11 @@ package main import ( "context" "testing" + "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/restrict" + "git.yeij.top/AskaEth/Cyrene_v2/internal/config" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool" @@ -105,3 +107,25 @@ func TestPersonaExistsMirrorsAgentPool(t *testing.T) { t.Fatal("a persona-less dispatch was rejected by an empty pool; want it unaffected") } } + +// TestSubagentToolTimeoutFromConfig 锁定组合根的配置接法（design/05 §7 裁决 +// 02.4.4）：未配置时回落 10min 代码默认，SUBAGENT_TIMEOUT_SEC 生效时被真实 +// 采用——派发工具拿到的是配置值而不是常量。 +func TestSubagentToolTimeoutFromConfig(t *testing.T) { + a := &app{log: testLog()} + if got := a.subagentToolTimeout(); got != 10*time.Minute { + t.Fatalf("subagentToolTimeout() = %v, want the 10min code default", got) + } + + a.cfg = &config.Config{} + a.cfg.Subagent.TimeoutSec = 90 + if got := a.subagentToolTimeout(); got != 90*time.Second { + t.Fatalf("subagentToolTimeout() = %v, want 90s from config", got) + } + + // 非法（非正）配置不缩小时限：保持代码默认而不是把子代理变成「立即超时」。 + a.cfg.Subagent.TimeoutSec = -5 + if got := a.subagentToolTimeout(); got != 10*time.Minute { + t.Fatalf("subagentToolTimeout() = %v for a non-positive config, want the code default", got) + } +} diff --git a/internal/config/config.go b/internal/config/config.go index e51db5a..bd61de7 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -30,6 +30,7 @@ type Config struct { IoT IoTConfig Voice VoiceConfig Background BackgroundConfig + Subagent SubagentConfig PushOutlet PushOutletConfig WSHub WSHubConfig ExecSandbox ExecSandbox @@ -178,6 +179,16 @@ type BackgroundConfig struct { KindWindowSec int `env:"THINK_KIND_WINDOW_SEC" default:"7200"` } +// SubagentConfig holds the sub-agent orchestration settings (design/05 §7). +type SubagentConfig struct { + // TimeoutSec bounds one sub-agent's whole run. It feeds the dispatch tool's + // default (orchestration.WithSpawnTimeout), overriding the scheduler's code + // default; a per-dispatch timeout_seconds argument still overrides it per call + // (裁决 02.4.4: agent_timeout 可配置，默认 10min，支持按调用覆盖). Zero or + // negative selects the code default (600s = 10 minutes). + TimeoutSec int `env:"SUBAGENT_TIMEOUT_SEC" default:"600"` +} + // PushOutletConfig configures the proactive push outlet (D1, design/12 §4/§6). // Proactive messages are mirrored to the owner's self-hosted unified push // service (primary); Getui (个推) is the deferred fallback for when that service diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 35db90b..f4e8385 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -361,3 +361,26 @@ EMPTY= t.Errorf("EMPTY = %q (present=%v), want empty", v, ok) } } + +// TestLoad_SubagentTimeout 验证子代理时限配置（design/05 §7 裁决 02.4.4： +// agent_timeout 可配置，默认 10min）：缺省 600 秒，环境变量可覆盖。 +func TestLoad_SubagentTimeout(t *testing.T) { + t.Setenv("JWT_SECRET", strings.Repeat("s", 40)) + + cfg, err := Load("") + if err != nil { + t.Fatalf("Load() error: %v", err) + } + if cfg.Subagent.TimeoutSec != 600 { + t.Errorf("Subagent.TimeoutSec = %d, want the 600s (10min) default", cfg.Subagent.TimeoutSec) + } + + t.Setenv("SUBAGENT_TIMEOUT_SEC", "90") + cfg, err = Load("") + if err != nil { + t.Fatalf("Load() with override error: %v", err) + } + if cfg.Subagent.TimeoutSec != 90 { + t.Errorf("Subagent.TimeoutSec = %d, want 90 from the environment", cfg.Subagent.TimeoutSec) + } +} diff --git a/internal/runtime/loop/options.go b/internal/runtime/loop/options.go index 79d11c4..04a8acb 100644 --- a/internal/runtime/loop/options.go +++ b/internal/runtime/loop/options.go @@ -176,3 +176,78 @@ type Options struct { // "runtime/loop" 的丢弃 logger。 Log *slog.Logger } + +// toolWhitelistKey 是每 Run 工具白名单的 context key（design/05 §7 +// 裁决 02.4.1）。编排层在驱动子代理 Run 之前挂载它。 +type toolWhitelistKey struct{} + +// WithToolWhitelist 把该次运行的工具白名单挂到 ctx 上（design/05 §7）： +// nil/空名单原样返回 ctx，即「缺省全开」——继承主 agent 的全量工具集， +// 与挂载前的行为逐字节一致。名单会被复制，调用方后续修改不影响已挂载值。 +func WithToolWhitelist(ctx context.Context, names []string) context.Context { + if ctx == nil || len(names) == 0 { + return ctx + } + return context.WithValue(ctx, toolWhitelistKey{}, append([]string(nil), names...)) +} + +// ToolWhitelistFrom 返回 WithToolWhitelist 挂到 ctx 上的白名单； +// 未挂载或为空返回 nil，语义为「全开」。它是 WithToolWhitelist 的读侧， +// 供拿到 context 的调用方（子代理 runner、测试）观察本次运行的能力面。 +func ToolWhitelistFrom(ctx context.Context) []string { + if ctx == nil { + return nil + } + names, _ := ctx.Value(toolWhitelistKey{}).([]string) + return names +} + +// toolWhitelistAllows 报告 tool 是否在名单内。 +func toolWhitelistAllows(names []string, tool string) bool { + for _, n := range names { + if n == tool { + return true + } + } + return false +} + +// whitelistVetoReason 是白名单外调用被否决后回灌给模型的软提示， +// 与 design/11 §3 安全门的 not_allowed 文案区分开：这是子代理的 +// 能力面收缩（design/05 §7），不是会话身份限制。 +func whitelistVetoReason(name string) string { + return "not_allowed: tool " + name + " is not in this sub-agent's tool whitelist" +} + +// WithToolWhitelist 返回工具面被限定为 names 的 Loop 派生副本（design/05 +// §7 裁决 02.4.1：显式列出 = 只放行所列）。它是浅拷贝：除 hooks 外的所有 +// 依赖（agent、dispatcher、记忆、追踪）与原模板共享，因此并发 Run 契约 +// 与原 Loop 一致；唯一的差异是 CallAdmit 钩子——白名单外的调用被否决为 +// 携带原因的跳过结果回灌给模型（软结果，绝不 panic），白名单内的调用 +// 继续交给模板上已配置的钩子链（先跑原钩子，被原钩子拦截的保持拦截）。 +// 空/nil 名单返回原模板（全开）。编排层按子代理逐个派生，主会话的 +// Loop 不受影响。 +func (l *Loop) WithToolWhitelist(names []string) *Loop { + if l == nil || len(names) == 0 { + return l + } + cp := *l + prev := l.hooks + cp.hooks = ToolLoopHooks{ + BatchAdmit: prev.BatchAdmit, + AfterCall: prev.AfterCall, + AfterOutcome: prev.AfterOutcome, + CallAdmit: func(ctx context.Context, call agent.ToolCall) *ToolCallVeto { + if prev.CallAdmit != nil { + if v := prev.CallAdmit(ctx, call); v != nil && v.Block { + return v + } + } + if !toolWhitelistAllows(names, call.Name) { + return &ToolCallVeto{Block: true, Reason: whitelistVetoReason(call.Name)} + } + return nil + }, + } + return &cp +} diff --git a/internal/runtime/orchestration/merge.go b/internal/runtime/orchestration/merge.go index 05236d7..75aeddd 100644 --- a/internal/runtime/orchestration/merge.go +++ b/internal/runtime/orchestration/merge.go @@ -90,6 +90,15 @@ func timeoutSummary(d time.Duration) string { return fmt.Sprintf("中止: 超时 (%s)", d) } +// timeoutSummaryFor 是按子代理生效时限渲染的超时摘要：无期限（0）时不再 +// 显示误导性的 "(0s)"，只说明超时本身。 +func timeoutSummaryFor(d time.Duration) string { + if d <= 0 { + return "中止: 超时" + } + return timeoutSummary(d) +} + // summarizeMerged produces the one-line aggregate summary of a merged result. func summarizeMerged(r *MergedResult) string { return fmt.Sprintf("子代理汇总: 共 %d 个, %d 成功, %d 失败, %d 中止", diff --git a/internal/runtime/orchestration/options.go b/internal/runtime/orchestration/options.go index 9376825..391daa2 100644 --- a/internal/runtime/orchestration/options.go +++ b/internal/runtime/orchestration/options.go @@ -97,3 +97,38 @@ type FlowSpanSink interface { // SpanFinished folds a just-finished span into the session graph. SpanFinished(sessionID string, sp *trace.Span) } + +// 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 +} + +// scopedRunner picks the runner that executes sub: without a tool whitelist the +// configured template runs as-is（缺省全开，继承主 agent 全量工具集）；with one, +// a *loop.Loop template is derived through WithToolWhitelist so only the listed +// tools pass（design/05 §7 裁决 02.4.1）. A non-loop runner (tests inject mocks) +// is returned unchanged — the whitelist still rides the run context for it to +// observe via loop.ToolWhitelistFrom. +func (s *Scheduler) scopedRunner(sub *SubAgent) runner { + if len(sub.ToolWhitelist) == 0 { + return s.runner + } + if lp, ok := s.runner.(*loop.Loop); ok { + return lp.WithToolWhitelist(sub.ToolWhitelist) + } + return s.runner +} + +// effectiveTimeout resolves the deadline one sub-agent runs under: a positive +// per-call override wins; otherwise the scheduler's PerAgentTimeout applies +// (non-positive = no deadline). +func effectiveTimeout(override, schedulerDefault time.Duration) time.Duration { + if override > 0 { + return override + } + return schedulerDefault +} diff --git a/internal/runtime/orchestration/orchestration_test.go b/internal/runtime/orchestration/orchestration_test.go index ec3a7b6..67b671e 100644 --- a/internal/runtime/orchestration/orchestration_test.go +++ b/internal/runtime/orchestration/orchestration_test.go @@ -112,9 +112,17 @@ func TestContextIsolation(t *testing.T) { if m.total != 1 { t.Fatalf("runner calls = %d, want 1", m.total) } + // The runner receives the isolated caller context plus the task itself as one + // trailing user message (裁决 2026-10-04: 任务作为一条 user 消息注入). 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 len(got) != 2 { + t.Fatalf("runner received %d messages, want the isolated context + the task message: %#v", len(got), got) + } + if got[0].Role != model.RoleUser || got[0].Content[0].Text != "hello" { + t.Fatalf("runner context[0] = %+v, want the isolated 'hello' user message", got[0]) + } + if got[1].Role != model.RoleUser || !strings.Contains(got[1].Content[0].Text, "x") { + t.Fatalf("runner context[1] = %+v, want the task user message carrying the goal", got[1]) } if sub.Context[0].Content[0].Text != "hello" { t.Fatalf("sub-agent context text = %q, want hello", sub.Context[0].Content[0].Text) diff --git a/internal/runtime/orchestration/scheduler.go b/internal/runtime/orchestration/scheduler.go index 0aa653c..3d8fb03 100644 --- a/internal/runtime/orchestration/scheduler.go +++ b/internal/runtime/orchestration/scheduler.go @@ -123,7 +123,8 @@ func (s *Scheduler) Comm() *comm.Probe { return s.comm } // when spec.Goal is blank, and a wrapped ErrUnknownPersona when spec.PersonaID // names a role this process has no agent for (design/22 §8-2). // -// The sub-agent's Context is a deep copy of spec.Context and its WorkingMemory is +// The sub-agent's Context is a deep copy of spec.Context plus the task itself +// as one trailing user message (裁决 2026-10-04), 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) == "" { @@ -145,8 +146,10 @@ func (s *Scheduler) Spawn(ctx context.Context, spec AgentSpec) (*SubAgent, error AttachInfo: spec.AttachInfo, ParentSession: spec.ParentSession, Data: spec.Data, - Context: cloneMessages(spec.Context), + ToolWhitelist: cloneStrings(spec.ToolWhitelist), + Context: taskContext(spec), WorkingMemory: memory.NewWorkingMemory(0), + timeout: effectiveTimeout(spec.Timeout, s.perAgentTimeout), status: StatusPending, done: make(chan struct{}), finishedCh: make(chan struct{}), @@ -304,7 +307,7 @@ func (s *Scheduler) run(ctx context.Context, sub *SubAgent) { } s.publish(EventStarted, sub.progress()) - runCtx, cancel := s.subCtx(ctx) + runCtx, cancel := s.subCtx(ctx, sub) // A sub-agent must not leak its inner monologue into the caller's chat stream // (design/05 §8: 少说话、多做事; design/04 §2: 不污染前台): muting suppresses the // UI stream/capacity broadcast while progress still reaches the frontend over @@ -316,11 +319,14 @@ func (s *Scheduler) run(ctx context.Context, sub *SubAgent) { // of this run (design/22 §8-2). Blank keeps the inherited env: the pre-existing // behavior, where a sub-agent runs as the spawning turn's role. runCtx = withPersonaEnv(runCtx, sub.PersonaID) + // 工具白名单（design/05 §7 裁决 02.4.1）随运行上下文传递：名单非空时 + // 只放行所列工具；空名单＝全开，与挂载前的行为一致。 + runCtx = loop.WithToolWhitelist(runCtx, sub.ToolWhitelist) runCtx, span := s.subSpan(runCtx, sub) sub.setCancel(cancel) defer cancel() - res, err := s.runner.Run(runCtx, sub.ID, sub.Context) + res, err := s.scopedRunner(sub).Run(runCtx, sub.ID, sub.Context) if err != nil { s.spanErr(sub, span, err) s.abortOrFail(sub, err) @@ -333,11 +339,12 @@ func (s *Scheduler) run(ctx context.Context, sub *SubAgent) { // 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. +// real failure. The timeout notice names the deadline that actually applied to +// this sub-agent (per-call override or scheduler default), not the global one. func (s *Scheduler) abortOrFail(sub *SubAgent, err error) { switch { case errors.Is(err, context.DeadlineExceeded): - s.finish(sub, StatusAborted, timeoutSummary(s.perAgentTimeout), err) + s.finish(sub, StatusAborted, timeoutSummaryFor(sub.timeout), err) case errors.Is(err, context.Canceled): s.finish(sub, StatusAborted, canceledSummary(), err) default: @@ -363,10 +370,13 @@ func (s *Scheduler) finish(sub *SubAgent, status Status, summary string, err err }) } -// 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) +// subCtx derives the run context, applying the sub-agent's effective timeout: +// a per-call AgentSpec.Timeout (design/05 §7 裁决 02.4.4 按调用覆盖) wins over +// the scheduler's PerAgentTimeout; neither configured means no deadline beyond +// cancellation. +func (s *Scheduler) subCtx(parent context.Context, sub *SubAgent) (context.Context, context.CancelFunc) { + if sub.timeout > 0 { + return context.WithTimeout(parent, sub.timeout) } return context.WithCancel(parent) } @@ -475,12 +485,3 @@ func eventFor(status Status) string { 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 -} diff --git a/internal/runtime/orchestration/spawntask_test.go b/internal/runtime/orchestration/spawntask_test.go new file mode 100644 index 0000000..5b3f407 --- /dev/null +++ b/internal/runtime/orchestration/spawntask_test.go @@ -0,0 +1,332 @@ +package orchestration + +import ( + "context" + "strings" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" + "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/tool" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// requestCaptureAdapter 记录每次模型请求的完整报文（Messages/Tools）， +// 使测试能断言「子代理的模型真正看到了什么」——这是 02.4.2 的裁决 +// （任务作为一条 user 消息注入）与 02.4.1（工具白名单）的真断言载体： +// 恒答固定文本的 mock 适配器测不出这两类缺陷。 +type requestCaptureAdapter struct { + final string + + mu sync.Mutex + requests []model.Request +} + +func (a *requestCaptureAdapter) Name() string { return "capture" } + +func (a *requestCaptureAdapter) record(req model.Request) { + a.mu.Lock() + a.requests = append(a.requests, req) + a.mu.Unlock() +} + +func (a *requestCaptureAdapter) Chat(_ context.Context, req model.Request, _ model.Options) (model.Response, error) { + a.record(req) + return model.Response{Model: "mock-1", Content: a.final, StopReason: "end_turn"}, nil +} + +func (a *requestCaptureAdapter) ChatStream(_ context.Context, req model.Request, _ func(model.StreamEvent)) (model.Response, error) { + return a.Chat(context.Background(), req, model.Options{}) +} + +// request 返回第 i 次捕获的模型请求。 +func (a *requestCaptureAdapter) request(i int) model.Request { + a.mu.Lock() + defer a.mu.Unlock() + return a.requests[i] +} + +func (a *requestCaptureAdapter) count() int { + a.mu.Lock() + defer a.mu.Unlock() + return len(a.requests) +} + +// captureLoop 在真实 loop.Loop 之上构建带请求捕获的调度器（生产形状， +// 只是模型换成了捕获适配器）。tools 为 nil 时不接 dispatcher。 +func captureLoop(t *testing.T, ad model.Adapter, d *tool.Dispatcher) *Scheduler { + t.Helper() + ag, err := agent.New("你是后台子代理，按角色与安全约束行事", d, nil, + agent.WithModel(ad, nil), + agent.WithOptions(agent.Options{Model: "mock-1"})) + if err != nil { + t.Fatalf("agent.New: %v", err) + } + lp := loop.New(loop.Options{Agent: ag, Tools: d, MaxRounds: 8}) + return newScheduler(t, Options{Loop: lp, MaxParallel: 1}) +} + +// messageTextOf 提取一条消息的纯文本（拼接 text 与 tool_result 分段， +// 后者承载被否决/失败调用的回灌文本）。 +func messageTextOf(m model.Message) string { + var b strings.Builder + for _, c := range m.Content { + if c.Type == model.ContentTypeText || c.Type == model.ContentTypeToolResult { + b.WriteString(c.Text) + } + } + return b.String() +} + +// TestSpawnToolInjectsTaskAsUserMessage 是 02.4.2（P0「子代理收不到任务」） +// 的裁决断言：经生产派发工具创建的子代理，其模型请求里必须有一条 user +// 消息同时承载 goal 正文与 attach_info 独立段落；system 只放角色与安全 +// 约束，绝不包含任务文本。 +func TestSpawnToolInjectsTaskAsUserMessage(t *testing.T) { + t.Parallel() + ad := &requestCaptureAdapter{final: "子代理完成"} + s := captureLoop(t, ad, nil) + sp := NewSpawnTool(s, WithSpawnWait(3*time.Second)) + + raw, err := sp.Execute(context.Background(), map[string]any{ + "goal": "调研量子纠错现状", + "attach_info": "只看 2026 年后的论文", + }) + if err != nil { + t.Fatalf("Execute: %v", err) + } + got := decodeSpawnResult(t, raw) + if got.Status != "done" { + t.Fatalf("Status = %q, want done", got.Status) + } + if ad.count() == 0 { + t.Fatal("the sub-agent never reached a model; nothing to assert the task injection on") + } + msgs := ad.request(0).Messages + + var userTexts, systemTexts []string + for _, m := range msgs { + switch m.Role { + case model.RoleUser: + userTexts = append(userTexts, messageTextOf(m)) + case model.RoleSystem: + systemTexts = append(systemTexts, messageTextOf(m)) + } + } + if len(userTexts) != 1 { + t.Fatalf("user messages = %d (%v), want exactly one carrying the task", len(userTexts), userTexts) + } + task := userTexts[0] + for _, want := range []string{"[任务目标]", "调研量子纠错现状", "[附加信息]", "只看 2026 年后的论文"} { + if !strings.Contains(task, want) { + t.Fatalf("task user message = %q, want it to contain %q", task, want) + } + } + for _, sys := range systemTexts { + if strings.Contains(sys, "调研量子纠错现状") { + t.Fatalf("system message carries the task text (%q); the ruling keeps the task out of system", sys) + } + } +} + +// TestSpawnInjectsDataParagraph 断言 AgentSpec.Data 作为同一条 user 消息内的 +// [任务数据] 段落到达模型（库路径调用方同样受裁决覆盖）。 +func TestSpawnInjectsDataParagraph(t *testing.T) { + t.Parallel() + ad := &requestCaptureAdapter{final: "done"} + s := captureLoop(t, ad, nil) + + sub, err := s.Spawn(context.Background(), AgentSpec{ + Goal: "处理数据", + Data: map[string]any{"repo": "cyrene", "stars": 2}, + }) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + if _, werr := s.Wait(context.Background(), []string{sub.ID}); werr != nil { + t.Fatalf("Wait: %v", werr) + } + msgs := ad.request(0).Messages + var task string + users := 0 + for _, m := range msgs { + if m.Role != model.RoleUser { + continue + } + users++ + task = messageTextOf(m) + } + if users != 1 { + t.Fatalf("user messages = %d, want one", users) + } + for _, want := range []string{"[任务数据]", `"repo":"cyrene"`} { + if !strings.Contains(task, want) { + t.Fatalf("task message = %q, want it to contain %q", task, want) + } + } +} + +// TestSpawnTimeoutFromToolDefault 断言 WithSpawnTimeout（生产经 +// SUBAGENT_TIMEOUT_SEC 填充）覆盖调度器默认时限（02.4.4）。 +func TestSpawnTimeoutFromToolDefault(t *testing.T) { + t.Parallel() + s := newScheduler(t, Options{MaxParallel: 1, PerAgentTimeout: time.Hour}) + s.runner = &mockRunner{gate: make(chan struct{})} // 阻塞直到 ctx 超时 + sp := NewSpawnTool(s, WithSpawnWait(30*time.Millisecond), WithSpawnTimeout(120*time.Millisecond)) + + raw, err := sp.Execute(context.Background(), map[string]any{"goal": "slow"}) + if err != nil { + t.Fatalf("Execute: %v", err) + } + got := decodeSpawnResult(t, raw) + if got.Status != "running" { + t.Fatalf("Status = %q, want running", got.Status) + } + id := got.Agents[0].ID + waitFor(t, func() bool { return statusOf(s, id) == StatusAborted }, waitTimeout) + sub := s.pickOrFail(t, id) + if !strings.Contains(sub.Summary, "120ms") { + t.Fatalf("Summary = %q, want the 120ms tool-default deadline named", sub.Summary) + } +} + +// TestSpawnTimeoutPerCallOverride 断言按调用覆盖（timeout_seconds）优先于 +// 派发默认与调度器默认（design/05 §7 裁决 02.4.4）。 +func TestSpawnTimeoutPerCallOverride(t *testing.T) { + t.Parallel() + s := newScheduler(t, Options{MaxParallel: 1, PerAgentTimeout: time.Hour}) + s.runner = &mockRunner{gate: make(chan struct{})} + sp := NewSpawnTool(s, WithSpawnWait(30*time.Millisecond), WithSpawnTimeout(time.Hour)) + + raw, err := sp.Execute(context.Background(), map[string]any{"goal": "slow", "timeout_seconds": float64(1)}) + if err != nil { + t.Fatalf("Execute: %v", err) + } + id := decodeSpawnResult(t, raw).Agents[0].ID + waitFor(t, func() bool { return statusOf(s, id) == StatusAborted }, 3*time.Second) + sub := s.pickOrFail(t, id) + if !strings.Contains(sub.Summary, "1s") { + t.Fatalf("Summary = %q, want the 1s per-call deadline named", sub.Summary) + } +} + +// TestSpawnTimeoutSpecOverride 是库路径的按调用覆盖断言：AgentSpec.Timeout +// 优先于 Options.PerAgentTimeout。 +func TestSpawnTimeoutSpecOverride(t *testing.T) { + t.Parallel() + s := newScheduler(t, Options{MaxParallel: 1, PerAgentTimeout: time.Hour}) + s.runner = &mockRunner{gate: make(chan struct{})} + + sub, err := s.Spawn(context.Background(), AgentSpec{Goal: "quick", Timeout: 80 * time.Millisecond}) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + waitFor(t, func() bool { return sub.Status() == StatusAborted }, waitTimeout) + if !strings.Contains(sub.Summary(), "80ms") { + t.Fatalf("Summary = %q, want the 80ms spec deadline named", sub.Summary()) + } + if sub.FinishedAt().Sub(sub.StartedAt()) > 5*time.Second { + t.Fatalf("run took %v; the per-call override never applied", sub.FinishedAt().Sub(sub.StartedAt())) + } +} + +// pickOrFail 按 id 取子代理快照，找不到时失败测试。 +func (s *Scheduler) pickOrFail(t *testing.T, id string) SubAgentResult { + t.Helper() + for _, a := range s.Agents() { + if a.ID == id { + return a + } + } + t.Fatalf("sub-agent %q not registered", id) + return SubAgentResult{} +} + +// TestLateSubagentSpanRejoinsOriginalRound 是 02.4.4「迟到 span 续根」的 +// 归属断言（design/05 §9、design/19 §2）：回合（EndRound）先于子代理结束， +// 之后才到达的子代理 span 必须归入原轮（web-1）的泳道——根节点翻转为终态、 +// 内部步骤链完整——而不是被丢弃，也不是另起新轮根。 +func TestLateSubagentSpanRejoinsOriginalRound(t *testing.T) { + t.Parallel() + capture := &graphCapture{} + flow := loop.NewFlowPublisher(capture.publish, nil) + tracer := trace.New(nil) + + d, err := tool.New(registry.New()) + if err != nil { + t.Fatalf("tool.New: %v", err) + } + if rerr := d.Register(&echoTool{name: "echo", result: "ok", delay: 150 * time.Millisecond}); rerr != nil { + t.Fatalf("Register: %v", rerr) + } + s := branchLoop(t, flow, tracer, d, Options{MaxParallel: 1}) + + flow.StartRound(parentSession, "帮我调研") + sub, serr := s.Spawn(context.Background(), AgentSpec{Goal: "慢速调研", ParentSession: parentSession}) + if serr != nil { + t.Fatalf("Spawn: %v", serr) + } + // 回合在子代理仍在运行时就结束：其后到达的 span 都是「迟到 span」。 + flow.EndRound(parentSession, "结果如下") + if _, werr := s.Wait(context.Background(), []string{sub.ID}); werr != nil { + t.Fatalf("Wait: %v", werr) + } + + // 不丢、不另起根：整个会话只出版过一轮（web-1），迟到的泳道落在它里面。 + if ids := capture.ids(parentSession); len(ids) != 1 || ids[0] != parentSession+"-1" { + t.Fatalf("published rounds for %q = %v, want exactly the original round %s-1", parentSession, ids, parentSession) + } + if capture.has(sub.ID) { + t.Fatalf("the late sub-agent opened a graph of its own (%s); it must rejoin the original round", sub.ID) + } + g := capture.graph(t, parentSession) + if g.ID != parentSession+"-1" { + t.Fatalf("graph id = %q, want the original round's %s-1", g.ID, parentSession) + } + // 原轮的收尾仍在，且迟到泳道完整挂靠：根节点终态 + llm>tool>llm 步骤链。 + if flowNodeByID(t, g, "output").Kind != "output" { + t.Fatal("the closed round lost its output node") + } + subNode := flowSubagentNode(t, g, sub.ID) + if subNode.Status != trace.FlowStatusDone { + t.Fatalf("late subagent node status = %q, want done (the late finish must flip its root)", subNode.Status) + } + if chain := flowLaneChain(t, g, subNode); len(chain) != 3 || chainKinds(chain) != "llm>tool>llm" { + t.Fatalf("late lane chain = %s, want llm>tool>llm inside the original round", chainKinds(chain)) + } +} + +// TestTaskContextIsolationFromCallerMutation 锁定注入消息与调用方切片的隔离： +// Spawn 后修改 spec.Context / spec.ToolWhitelist 不得影响已创建的子代理。 +func TestTaskContextIsolationFromCallerMutation(t *testing.T) { + t.Parallel() + s := newScheduler(t, Options{MaxParallel: 1}) + m := &ctxCaptureRunner{} + s.runner = m + + wl := []string{"alpha"} + spec := AgentSpec{Goal: "g", ToolWhitelist: wl, Context: []model.Message{model.TextMessage(model.RoleUser, "seed")}} + sub, err := s.Spawn(context.Background(), spec) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + wl[0] = "mutated" + spec.Context[0].Content[0].Text = "mutated" + + if _, werr := s.Wait(context.Background(), []string{sub.ID}); werr != nil { + t.Fatalf("Wait: %v", werr) + } + if got := loop.ToolWhitelistFrom(m.last()); len(got) != 1 || got[0] != "alpha" { + t.Fatalf("whitelist = %v, want the snapshot [alpha] immune to caller mutation", got) + } + if sub.ToolWhitelist[0] != "alpha" { + t.Fatalf("sub-agent whitelist = %v, want [alpha]", sub.ToolWhitelist) + } + if sub.Context[0].Content[0].Text != "seed" { + t.Fatalf("sub-agent context = %q, want the isolated seed", sub.Context[0].Content[0].Text) + } +} diff --git a/internal/runtime/orchestration/spawntool.go b/internal/runtime/orchestration/spawntool.go index 462ca70..3fd8492 100644 --- a/internal/runtime/orchestration/spawntool.go +++ b/internal/runtime/orchestration/spawntool.go @@ -36,6 +36,12 @@ const ( // front session's context. The scheduler keeps the full text on its snapshot. maxSpawnTextRunes = 4000 + // minSpawnTimeoutSeconds 与 maxSpawnTimeoutSeconds 界定按调用覆盖 + // timeout_seconds 的合法区间（design/05 §7 裁决 02.4.4 的「按调用覆盖」）： + // 下限挡住「立即超时」的畸形调用，上限挡住「永远不超时」的失控委托。 + minSpawnTimeoutSeconds = 1 + maxSpawnTimeoutSeconds = 24 * 60 * 60 + // SubagentResultSubject is the bus subject a finished background batch is // announced on, carrying the *MergedResult. It is the design/05 §9 结果回填 hook // for whichever component owns the session message queue (design/15); the @@ -53,9 +59,10 @@ const ( // // It is safe for concurrent use and holds no per-call state. type SpawnTool struct { - sched *Scheduler - wait time.Duration - log *slog.Logger + sched *Scheduler + wait time.Duration + timeout time.Duration + log *slog.Logger } // SpawnOption customizes a SpawnTool built by NewSpawnTool. @@ -76,6 +83,18 @@ func WithSpawnLogger(l *slog.Logger) SpawnOption { return func(t *SpawnTool) { t.log = spawnLog(l) } } +// WithSpawnTimeout 设置派发层的默认子代理时限（design/05 §7 裁决 02.4.4 +// 「agent_timeout 可配置」的接法）：正数时作为未按调用覆盖的批次默认， +// 覆盖调度器的 PerAgentTimeout；非正保持「不覆盖」，沿用调度器默认。 +// 生产接线用 SUBAGENT_TIMEOUT_SEC（默认 10min）填充它。 +func WithSpawnTimeout(d time.Duration) SpawnOption { + return func(t *SpawnTool) { + if d > 0 { + t.timeout = d + } + } +} + // NewSpawnTool builds the dispatch tool over sched. A nil scheduler is tolerated: // Execute then fails fast with ErrNoScheduler instead of panicking. func NewSpawnTool(sched *Scheduler, opts ...SpawnOption) *SpawnTool { @@ -103,17 +122,21 @@ func (t *SpawnTool) Description() string { // Parameters returns the input JSON Schema: either a single goal shorthand or a // tasks array for a parallel batch, plus an optional persona (the role the -// sub-agent runs as, design/22 §8-2) and an optional background flag. +// sub-agent runs as, design/22 §8-2), an optional background flag, an optional +// per-dispatch tool whitelist (design/05 §7) and an optional per-call timeout +// override. func (t *SpawnTool) Parameters() (map[string]any, error) { return map[string]any{ "type": "object", "properties": map[string]any{ - "goal": spawnStringProp("Objective the sub-agent works toward alone. Required unless tasks is given."), - "persona": spawnPersonaProp(), - "role": spawnStringProp("Role/persona id to bind the sub-agent to (optional). Superseded by persona, which is the role the sub-agent actually runs as."), - "mode": spawnStringProp("Behavior mode for the role, e.g. engineer (optional)."), - "attach_info": spawnStringProp("Extra context or data to hand the sub-agent (optional)."), - "tasks": spawnTasksProp(), + "goal": spawnStringProp("Objective the sub-agent works toward alone. Required unless tasks is given."), + "persona": spawnPersonaProp(), + "role": spawnStringProp("Role/persona id to bind the sub-agent to (optional). Superseded by persona, which is the role the sub-agent actually runs as."), + "mode": spawnStringProp("Behavior mode for the role, e.g. engineer (optional)."), + "attach_info": spawnStringProp("Extra context or data to hand the sub-agent (optional)."), + "tasks": spawnTasksProp(), + "tool_whitelist": spawnWhitelistProp("Applies to every sub-agent of this dispatch when a task does not set its own."), + "timeout_seconds": spawnTimeoutProp(t.timeout), "background": map[string]any{ "type": "boolean", "description": "Dispatch and return immediately without waiting for the foreground window (optional, default false).", @@ -143,7 +166,7 @@ func (t *SpawnTool) Execute(ctx context.Context, args any) (any, error) { if t.sched == nil { return nil, fmt.Errorf("%s: %w", SpawnToolName, ErrNoScheduler) } - specs, err := spawnSpecs(m, loop.FlowOwnerFromContext(ctx)) + specs, err := spawnSpecs(m, loop.FlowOwnerFromContext(ctx), t.timeout) if err != nil { return nil, fmt.Errorf("%s: %w", SpawnToolName, err) } @@ -286,8 +309,9 @@ func runningSpawnResult(ids []string, specs []AgentSpec) *SpawnResult { // spawnSpecs decodes the tool arguments into one AgentSpec per sub-agent. A // non-empty tasks array is a parallel batch and takes precedence over the single -// goal shorthand. -func spawnSpecs(m map[string]any, parentSession string) ([]AgentSpec, error) { +// goal shorthand. dftTimeout is the dispatch-layer default (WithSpawnTimeout); +// a non-positive value leaves the scheduler's per-agent timeout in charge. +func spawnSpecs(m map[string]any, parentSession string, dftTimeout time.Duration) ([]AgentSpec, error) { tasks, err := taskList(m["tasks"]) if err != nil { return nil, err @@ -300,14 +324,27 @@ func spawnSpecs(m map[string]any, parentSession string) ([]AgentSpec, error) { if serr != nil { return nil, serr } + // 派发层默认时限只补位：按调用未给出 timeout_seconds 时才生效。 + if spec.Timeout == 0 { + spec.Timeout = dftTimeout + } return []AgentSpec{spec}, nil } + // 批次级的白名单是未自带名单任务的缺省值（缺省全开：未列出即继承全量）。 + batchWhitelist := stringListArg(m, "tool_whitelist") + batchTimeout := timeoutArg(m, "timeout_seconds", dftTimeout) specs := make([]AgentSpec, 0, len(tasks)) for i, task := range tasks { spec, serr := specFromMap(task, parentSession) if serr != nil { return nil, fmt.Errorf("tasks[%d]: %w", i, serr) } + if len(spec.ToolWhitelist) == 0 { + spec.ToolWhitelist = cloneStrings(batchWhitelist) + } + if spec.Timeout == 0 { + spec.Timeout = batchTimeout + } specs = append(specs, spec) } return specs, nil @@ -337,6 +374,8 @@ func specFromMap(m map[string]any, parentSession string) (AgentSpec, error) { Mode: strings.TrimSpace(stringArg(m, "mode")), AttachInfo: strings.TrimSpace(stringArg(m, "attach_info")), ParentSession: parentSession, + ToolWhitelist: stringListArg(m, "tool_whitelist"), + Timeout: timeoutArg(m, "timeout_seconds", 0), }, nil } @@ -376,17 +415,49 @@ func spawnTasksProp() map[string]any { "items": map[string]any{ "type": "object", "properties": map[string]any{ - "goal": spawnStringProp("Objective this sub-agent works toward alone."), - "persona": spawnPersonaProp(), - "role": spawnStringProp("Role/persona id to bind the sub-agent to (optional). Superseded by persona."), - "mode": spawnStringProp("Behavior mode for the role (optional)."), - "attach_info": spawnStringProp("Extra context or data for this sub-agent (optional)."), + "goal": spawnStringProp("Objective this sub-agent works toward alone."), + "persona": spawnPersonaProp(), + "role": spawnStringProp("Role/persona id to bind the sub-agent to (optional). Superseded by persona."), + "mode": spawnStringProp("Behavior mode for the role (optional)."), + "attach_info": spawnStringProp("Extra context or data for this sub-agent (optional)."), + "tool_whitelist": spawnWhitelistProp("Only this sub-agent; empty means every registered tool."), + "timeout_seconds": spawnTimeoutProp(0), }, "required": []any{"goal"}, }, } } +// spawnWhitelistProp builds the tool_whitelist property schema (design/05 §7 +// 裁决 02.4.1)：缺省全开，显式列出则只放行所列。 +func spawnWhitelistProp(desc string) map[string]any { + return map[string]any{ + "type": "array", + "items": map[string]any{ + "type": "string", + }, + "description": "Tool names the sub-agent may invoke. Omitted or empty means every registered tool " + + "(the default, the main agent's full set); listing names restricts the sub-agent to exactly those " + + "tools. " + desc, + } +} + +// spawnTimeoutProp builds the timeout_seconds property schema (design/05 §7 +// 裁决 02.4.4 的按调用覆盖)。dft 非正时描述为「沿用派发默认」。 +func spawnTimeoutProp(dft time.Duration) map[string]any { + desc := "Per-call timeout override in seconds for the sub-agent run (optional; omitted keeps the " + + "configured sub-agent default)." + if dft > 0 { + desc = fmt.Sprintf("Per-call timeout override in seconds for the sub-agent run (optional, default %s).", dft) + } + return map[string]any{ + "type": "integer", + "minimum": minSpawnTimeoutSeconds, + "maximum": maxSpawnTimeoutSeconds, + "description": desc, + } +} + // spawnPersonaProp builds the persona property schema. It is the design/22 §8-2 派发 // switch: naming a role makes the sub-agent stand in for it (that role's // instructions and memory scope) instead of working as the spawning session's role. diff --git a/internal/runtime/orchestration/spawnwhitelist_test.go b/internal/runtime/orchestration/spawnwhitelist_test.go new file mode 100644 index 0000000..ebd4204 --- /dev/null +++ b/internal/runtime/orchestration/spawnwhitelist_test.go @@ -0,0 +1,229 @@ +package orchestration + +import ( + "context" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" + "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/tool" +) + +// twoToolAdapter 先在一轮里同时请求两个工具调用，见到任意工具结果后给最终 +// 答案——使一次子代理运行同时覆盖「白名单内被放行」与「白名单外被否决」 +// 两条路径。它同时捕获每次模型请求，供断言工具广告面与回灌文本。 +type twoToolAdapter struct { + first, second string + final string + + mu sync.Mutex + requests []model.Request +} + +func (a *twoToolAdapter) Name() string { return "two" } + +func (a *twoToolAdapter) Chat(_ context.Context, req model.Request, _ model.Options) (model.Response, error) { + a.mu.Lock() + a.requests = append(a.requests, req) + a.mu.Unlock() + for _, m := range req.Messages { + if m.Role == model.RoleTool { + return model.Response{Model: "mock-1", Content: a.final, StopReason: "end_turn"}, nil + } + } + return model.Response{Model: "mock-1", ToolCalls: []model.ToolCall{ + {ID: "call-1", Name: a.first, Args: map[string]any{"n": 1}}, + {ID: "call-2", Name: a.second, Args: map[string]any{"n": 2}}, + }}, nil +} + +func (a *twoToolAdapter) ChatStream(ctx context.Context, req model.Request, on func(model.StreamEvent)) (model.Response, error) { + return a.Chat(ctx, req, model.Options{}) +} + +// request 返回第 i 次捕获的模型请求。 +func (a *twoToolAdapter) request(i int) model.Request { + a.mu.Lock() + defer a.mu.Unlock() + return a.requests[i] +} + +func (a *twoToolAdapter) count() int { + a.mu.Lock() + defer a.mu.Unlock() + return len(a.requests) +} + +// hitTool 是一个只计数的工具（原子递增，容忍并行批次），使测试能断言 +// 「真的执行了」而不仅是被广告。 +type hitTool struct { + name string + hits *int32 + delay time.Duration +} + +func (h *hitTool) Name() string { return h.name } +func (h *hitTool) Description() string { return "count its invocations" } +func (h *hitTool) Parameters() (map[string]any, error) { return nil, nil } + +func (h *hitTool) Execute(ctx context.Context, _ any) (any, error) { + if h.delay > 0 { + select { + case <-time.After(h.delay): + case <-ctx.Done(): + return nil, ctx.Err() + } + } + if h.hits != nil { + atomic.AddInt32(h.hits, 1) + } + return h.name + ":ok", nil +} + +// twoToolRig 组装「两个可数工具 + 请求两工具并捕获请求的适配器 + 派发工具」 +// 的测试台。 +func twoToolRig(t *testing.T) (*SpawnTool, *twoToolAdapter, *int32, *int32) { + t.Helper() + var alphaHits, betaHits int32 + d, err := tool.New(registry.New()) + if err != nil { + t.Fatalf("tool.New: %v", err) + } + for _, h := range []*hitTool{ + {name: "alpha", hits: &alphaHits}, + {name: "beta", hits: &betaHits}, + } { + if rerr := d.Register(h); rerr != nil { + t.Fatalf("Register: %v", rerr) + } + } + two := &twoToolAdapter{first: "alpha", second: "beta", final: "完成"} + s := captureLoop(t, two, d) + sp := NewSpawnTool(s, WithSpawnWait(3*time.Second)) + return sp, two, &alphaHits, &betaHits +} + +// TestToolWhitelistDefaultAllTools 断言缺省全开：不列出 tool_whitelist 的 +// 派发，子代理照常执行主 agent 全量工具集里的每个工具（02.4.1）。 +func TestToolWhitelistDefaultAllTools(t *testing.T) { + t.Parallel() + sp, ad, alphaHits, betaHits := twoToolRig(t) + + raw, err := sp.Execute(context.Background(), map[string]any{"goal": "both"}) + if err != nil { + t.Fatalf("Execute: %v", err) + } + if got := decodeSpawnResult(t, raw).Status; got != "done" { + t.Fatalf("Status = %q, want done", got) + } + if *alphaHits == 0 || *betaHits == 0 { + t.Fatalf("default dispatch restricted tools: alpha=%d beta=%d, want both executed", *alphaHits, *betaHits) + } + // 两轮请求都广告了全部工具（能力面未被收缩）。 + for i := 0; i < ad.count(); i++ { + names := toolNamesOf(ad.request(i)) + if !names["alpha"] || !names["beta"] { + t.Fatalf("request %d advertised tools %v, want the full set (default = inherit everything)", i, names) + } + } +} + +// TestToolWhitelistExplicitRestricts 断言显式列出 = 只放行所列（02.4.1）： +// 白名单外的调用被否决为软结果回灌（模型可见 not_allowed 提示），白名单内 +// 的工具照常执行。 +func TestToolWhitelistExplicitRestricts(t *testing.T) { + t.Parallel() + sp, ad, alphaHits, betaHits := twoToolRig(t) + raw, err := sp.Execute(context.Background(), map[string]any{ + "goal": "only beta", + "tool_whitelist": []any{"beta"}, + }) + if err != nil { + t.Fatalf("Execute: %v", err) + } + if got := decodeSpawnResult(t, raw).Status; got != "done" { + t.Fatalf("Status = %q, want done (a veto is a soft result, not a failure)", got) + } + if *alphaHits != 0 { + t.Fatalf("alpha executed %d times despite not being whitelisted", *alphaHits) + } + if *betaHits == 0 { + t.Fatal("beta never executed although it is whitelisted") + } + // 被否决的调用以软结果回灌，模型能看到可读原因。 + vetoed := false + for i := 0; i < ad.count(); i++ { + for _, m := range ad.request(i).Messages { + if m.Role == model.RoleTool && strings.Contains(messageTextOf(m), "not in this sub-agent's tool whitelist") { + vetoed = true + } + } + } + if !vetoed { + t.Fatal("the out-of-whitelist call never surfaced a readable veto to the model") + } +} + +// toolNamesOf 提取一次请求里广告的工具名集合。 +func toolNamesOf(req model.Request) map[string]bool { + out := make(map[string]bool, len(req.Tools)) + for _, ts := range req.Tools { + out[ts.Name] = true + } + return out +} + +// ctxCaptureRunner 记录运行 context，供断言白名单「随运行上下文传递」。 +type ctxCaptureRunner struct { + mu sync.Mutex + ctx []context.Context +} + +func (r *ctxCaptureRunner) Run(ctx context.Context, _ string, _ []model.Message) (*loop.Result, error) { + r.mu.Lock() + r.ctx = append(r.ctx, ctx) + r.mu.Unlock() + return &loop.Result{Text: "ok", Steps: 1, StopReason: "stop"}, nil +} + +func (r *ctxCaptureRunner) last() context.Context { + r.mu.Lock() + defer r.mu.Unlock() + return r.ctx[len(r.ctx)-1] +} + +// TestToolWhitelistRidesRunContext 断言名单通过 loop.WithToolWhitelist +// 挂到运行 context 上（空名单 = 未挂载 = 全开）。 +func TestToolWhitelistRidesRunContext(t *testing.T) { + t.Parallel() + s := newScheduler(t, Options{MaxParallel: 1}) + m := &ctxCaptureRunner{} + s.runner = m + + sub, err := s.Spawn(context.Background(), AgentSpec{Goal: "w", ToolWhitelist: []string{"a", "b"}}) + if err != nil { + t.Fatalf("Spawn: %v", err) + } + if _, werr := s.Wait(context.Background(), []string{sub.ID}); werr != nil { + t.Fatalf("Wait: %v", werr) + } + if got := loop.ToolWhitelistFrom(m.last()); len(got) != 2 || got[0] != "a" || got[1] != "b" { + t.Fatalf("run context whitelist = %v, want [a b]", got) + } + + bare, err := s.Spawn(context.Background(), AgentSpec{Goal: "bare"}) + if err != nil { + t.Fatalf("Spawn bare: %v", err) + } + if _, werr := s.Wait(context.Background(), []string{bare.ID}); werr != nil { + t.Fatalf("Wait bare: %v", werr) + } + if got := loop.ToolWhitelistFrom(m.last()); got != nil { + t.Fatalf("whitelist = %v on a whitelist-less dispatch, want nil (全开)", got) + } +} diff --git a/internal/runtime/orchestration/spec.go b/internal/runtime/orchestration/spec.go index d82a4a4..5f59d6d 100644 --- a/internal/runtime/orchestration/spec.go +++ b/internal/runtime/orchestration/spec.go @@ -1,8 +1,10 @@ package orchestration import ( + "encoding/json" "errors" "strings" + "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" ) @@ -67,6 +69,13 @@ type AgentSpec struct { ParentSession string // Data carries an optional structured payload for the caller's use. Data any + // ToolWhitelist 限定该子代理可执行的工具集合（design/05 §7 裁决 02.4.1）： + // 缺省（nil/空）＝继承主 agent 的全量工具集；显式列出＝只放行所列工具， + // 名单外的调用在工具循环层被否决并以软结果回灌。它只收缩能力面，绝不扩展。 + ToolWhitelist []string + // Timeout 覆盖该子代理的运行时限（design/05 §7 裁决 02.4.4 的按调用覆盖）。 + // 非正时回退到调度器的 Options.PerAgentTimeout。 + Timeout time.Duration } // Progress event names published on the agent-to-agent bus via the Scheduler's @@ -159,3 +168,74 @@ func compactText(s string, n int) string { } return string(r[:n]) + "…" } + +// taskContext 返回子代理的隔离上下文：调用方显式给出的消息在前（深拷贝）， +// 任务本身作为最后一条 user 消息注入。裁决 2026-10-04：goal 正文 + +// attach_info/data 作为同一条 user 消息内的独立段落；system 只承载角色与 +// 安全约束（循环前置的 agent 指令），绝不把任务塞进 system——否则子代理 +// 的模型上下文里没有任何一轮用户消息，任务一个字都到不了模型。 +func taskContext(spec AgentSpec) []model.Message { + return append(cloneMessages(spec.Context), taskUserMessage(spec)) +} + +// taskUserMessage 把任务渲染为一条 user 消息（裁决 2026-10-04）： +// [任务目标] 段固定存在（Spawn 已保证 goal 非空），[附加信息] 与 +// [任务数据] 段仅在对应字段非空时出现；Data 以 JSON 编码，编码失败时 +// 整段跳过——宁可少一段噪声，不注入一段坏文本。 +func taskUserMessage(spec AgentSpec) model.Message { + var b strings.Builder + b.WriteString("[任务目标]\n") + b.WriteString(spec.Goal) + if spec.AttachInfo != "" { + b.WriteString("\n\n[附加信息]\n") + b.WriteString(spec.AttachInfo) + } + if spec.Data != nil { + if data, err := json.Marshal(spec.Data); err == nil { + b.WriteString("\n\n[任务数据]\n") + b.Write(data) + } + } + return model.TextMessage(model.RoleUser, b.String()) +} + +// cloneStrings 返回 names 的拷贝，使子代理持有的白名单与调用方的切片 +// 不共享内存。nil 进 nil 出。 +func cloneStrings(names []string) []string { + if names == nil { + return nil + } + return append([]string(nil), names...) +} + +// stringListArg 读取字符串数组参数：容忍缺失/非数组/非字符串元素， +// 逐项去空白并丢弃空项，使名单恒为「显式列出的工具名」。缺失返回 nil（全开）。 +func stringListArg(m map[string]any, key string) []string { + raw, ok := m[key].([]any) + if !ok { + return nil + } + out := make([]string, 0, len(raw)) + for _, v := range raw { + s, _ := v.(string) + if s = strings.TrimSpace(s); s != "" { + out = append(out, s) + } + } + return out +} + +// timeoutArg 读取 timeout_seconds 按调用覆盖（JSON 解码为 float64）： +// 缺失/非数/越界（<1s 或 >24h）视为未提供并沿用 dft；越界值不静默夹取—— +// 回退到默认而不是替模型猜一个邻近值。 +func timeoutArg(m map[string]any, key string, dft time.Duration) time.Duration { + f, ok := m[key].(float64) + if !ok { + return dft + } + sec := int64(f) + if sec < minSpawnTimeoutSeconds || sec > maxSpawnTimeoutSeconds { + return dft + } + return time.Duration(sec) * time.Second +} diff --git a/internal/runtime/orchestration/subagent.go b/internal/runtime/orchestration/subagent.go index c9bf51e..c7d1c19 100644 --- a/internal/runtime/orchestration/subagent.go +++ b/internal/runtime/orchestration/subagent.go @@ -11,8 +11,8 @@ import ( ) // SubAgent is a single spawned sub-agent instance. The identity fields (ID, Goal, -// RoleID, PersonaID, Mode, AttachInfo, ParentSession, Data, Context, WorkingMemory) -// are set once at Spawn and never mutated. The lifecycle fields (Status, Summary, +// RoleID, PersonaID, Mode, AttachInfo, ParentSession, Data, ToolWhitelist, +// 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 @@ -40,12 +40,19 @@ type SubAgent struct { ParentSession string // Data is the optional structured payload supplied at spawn. Data any + // ToolWhitelist 是该子代理的工具白名单（design/05 §7 裁决 02.4.1）： + // nil/空＝继承主 agent 全量工具集；非空＝只放行所列工具。Spawn 时深拷贝。 + ToolWhitelist []string // Context is the isolated conversation context. It is a deep copy of the - // AgentSpec.Context and is never shared with the front session. + // AgentSpec.Context plus the task user message, 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 + // timeout 是该子代理生效的运行时限（按调用覆盖优先于调度器默认， + // design/05 §7 裁决 02.4.4）；非正表示除取消外无期限。Spawn 时落定，只读。 + timeout time.Duration mu sync.RWMutex status Status