From 2573fd1e3fe9b9258d78a9acb0e2feca1c32522d Mon Sep 17 00:00:00 2001 From: Cai_Tang <106404101+Cai-Tang-www@users.noreply.github.com> Date: Tue, 14 Apr 2026 23:38:21 +0800 Subject: [PATCH 01/10] =?UTF-8?q?feat():=20=E5=AE=9E=E7=8E=B0=20SubAgent?= =?UTF-8?q?=20WorkerRuntime=E3=80=81=E8=A7=92=E8=89=B2=E7=AD=96=E7=95=A5?= =?UTF-8?q?=E4=B8=8E=20Runtime=20=E7=94=9F=E5=91=BD=E5=91=A8=E6=9C=9F?= =?UTF-8?q?=E6=8E=A5=E5=85=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/runtime/events.go | 26 +++ internal/runtime/runtime.go | 3 + internal/runtime/subagent_factory.go | 20 ++ internal/runtime/subagent_factory_test.go | 33 +++ internal/runtime/subagent_run.go | 140 ++++++++++++ internal/runtime/subagent_run_test.go | 154 +++++++++++++ internal/subagent/engine.go | 29 +++ internal/subagent/factory.go | 36 +++ internal/subagent/output_contract.go | 27 +++ internal/subagent/policy.go | 86 +++++++ internal/subagent/policy_test.go | 124 +++++++++++ internal/subagent/types.go | 224 +++++++++++++++++++ internal/subagent/util.go | 28 +++ internal/subagent/worker.go | 259 ++++++++++++++++++++++ internal/subagent/worker_test.go | 245 ++++++++++++++++++++ 15 files changed, 1434 insertions(+) create mode 100644 internal/runtime/subagent_factory.go create mode 100644 internal/runtime/subagent_factory_test.go create mode 100644 internal/runtime/subagent_run.go create mode 100644 internal/runtime/subagent_run_test.go create mode 100644 internal/subagent/engine.go create mode 100644 internal/subagent/factory.go create mode 100644 internal/subagent/output_contract.go create mode 100644 internal/subagent/policy.go create mode 100644 internal/subagent/policy_test.go create mode 100644 internal/subagent/types.go create mode 100644 internal/subagent/util.go create mode 100644 internal/subagent/worker.go create mode 100644 internal/subagent/worker_test.go diff --git a/internal/runtime/events.go b/internal/runtime/events.go index 97eadc6a..f9506c40 100644 --- a/internal/runtime/events.go +++ b/internal/runtime/events.go @@ -1,5 +1,7 @@ package runtime +import "neo-code/internal/subagent" + // EventType identifies the kind of runtime event emitted during a run. type EventType string @@ -90,3 +92,27 @@ type TokenUsagePayload struct { SessionInputTokens int `json:"session_input_tokens"` SessionOutputTokens int `json:"session_output_tokens"` } + +// SubAgentEventPayload 描述子代理执行生命周期的事件载荷。 +type SubAgentEventPayload struct { + Role subagent.Role `json:"role"` + TaskID string `json:"task_id"` + State subagent.State `json:"state"` + StopReason subagent.StopReason `json:"stop_reason,omitempty"` + Step int `json:"step,omitempty"` + Delta string `json:"delta,omitempty"` + Error string `json:"error,omitempty"` +} + +const ( + // EventSubAgentStarted 在子代理任务启动后触发。 + EventSubAgentStarted EventType = "subagent_started" + // EventSubAgentProgress 在子代理执行每一步后触发。 + EventSubAgentProgress EventType = "subagent_progress" + // EventSubAgentCompleted 在子代理成功结束后触发。 + EventSubAgentCompleted EventType = "subagent_completed" + // EventSubAgentFailed 在子代理失败结束后触发。 + EventSubAgentFailed EventType = "subagent_failed" + // EventSubAgentCanceled 在子代理被取消后触发。 + EventSubAgentCanceled EventType = "subagent_canceled" +) diff --git a/internal/runtime/runtime.go b/internal/runtime/runtime.go index beece19f..ec1aa87f 100644 --- a/internal/runtime/runtime.go +++ b/internal/runtime/runtime.go @@ -14,6 +14,7 @@ import ( providertypes "neo-code/internal/provider/types" "neo-code/internal/runtime/approval" agentsession "neo-code/internal/session" + "neo-code/internal/subagent" "neo-code/internal/tools" ) @@ -65,6 +66,7 @@ type Service struct { compactRunner contextcompact.Runner approvalBroker *approval.Broker memoExtractor MemoExtractor + subAgentFactory subagent.Factory events chan RuntimeEvent sessionMu sync.Mutex @@ -106,6 +108,7 @@ func NewWithFactory( providerFactory: providerFactory, contextBuilder: contextBuilder, approvalBroker: approval.NewBroker(), + subAgentFactory: subagent.NewWorkerFactory(nil), events: make(chan RuntimeEvent, 128), sessionLocks: make(map[string]*sessionLockEntry), activeRunCancels: make(map[uint64]context.CancelFunc), diff --git a/internal/runtime/subagent_factory.go b/internal/runtime/subagent_factory.go new file mode 100644 index 00000000..140f4f70 --- /dev/null +++ b/internal/runtime/subagent_factory.go @@ -0,0 +1,20 @@ +package runtime + +import "neo-code/internal/subagent" + +// SetSubAgentFactory 设置子代理运行时工厂；传入 nil 时回退到默认工厂。 +func (s *Service) SetSubAgentFactory(factory subagent.Factory) { + if factory == nil { + s.subAgentFactory = subagent.NewWorkerFactory(nil) + return + } + s.subAgentFactory = factory +} + +// SubAgentFactory 返回当前 runtime 持有的子代理运行时工厂。 +func (s *Service) SubAgentFactory() subagent.Factory { + if s.subAgentFactory == nil { + s.subAgentFactory = subagent.NewWorkerFactory(nil) + } + return s.subAgentFactory +} diff --git a/internal/runtime/subagent_factory_test.go b/internal/runtime/subagent_factory_test.go new file mode 100644 index 00000000..eea246b1 --- /dev/null +++ b/internal/runtime/subagent_factory_test.go @@ -0,0 +1,33 @@ +package runtime + +import ( + "testing" + + "neo-code/internal/subagent" +) + +type fakeSubAgentFactory struct{} + +func (fakeSubAgentFactory) Create(role subagent.Role) (subagent.WorkerRuntime, error) { + return subagent.NewWorkerFactory(nil).Create(role) +} + +func TestServiceSubAgentFactoryRegistration(t *testing.T) { + t.Parallel() + + svc := NewWithFactory(nil, nil, nil, nil, nil) + if svc.SubAgentFactory() == nil { + t.Fatalf("expected default sub-agent factory") + } + + custom := fakeSubAgentFactory{} + svc.SetSubAgentFactory(custom) + if svc.SubAgentFactory() == nil { + t.Fatalf("expected custom sub-agent factory") + } + + svc.SetSubAgentFactory(nil) + if svc.SubAgentFactory() == nil { + t.Fatalf("expected reset to default sub-agent factory") + } +} diff --git a/internal/runtime/subagent_run.go b/internal/runtime/subagent_run.go new file mode 100644 index 00000000..45169c03 --- /dev/null +++ b/internal/runtime/subagent_run.go @@ -0,0 +1,140 @@ +package runtime + +import ( + "context" + "errors" + "fmt" + "strings" + + "neo-code/internal/subagent" +) + +// SubAgentTaskInput 描述一次子代理任务执行请求。 +type SubAgentTaskInput struct { + RunID string + SessionID string + Role subagent.Role + Task subagent.Task + Budget subagent.Budget + Capability subagent.Capability +} + +// RunSubAgentTask 使用当前 runtime 注册的工厂执行一条子代理任务。 +func (s *Service) RunSubAgentTask(ctx context.Context, input SubAgentTaskInput) (subagent.Result, error) { + if err := ctx.Err(); err != nil { + return subagent.Result{}, err + } + if strings.TrimSpace(input.RunID) == "" { + return subagent.Result{}, errors.New("runtime: subagent run id is empty") + } + if !input.Role.Valid() { + return subagent.Result{}, fmt.Errorf("runtime: invalid subagent role %q", input.Role) + } + if err := input.Task.Validate(); err != nil { + return subagent.Result{}, err + } + + factory := s.SubAgentFactory() + worker, err := factory.Create(input.Role) + if err != nil { + _ = s.emit(ctx, EventSubAgentFailed, input.RunID, input.SessionID, SubAgentEventPayload{ + Role: input.Role, + TaskID: input.Task.ID, + State: subagent.StateFailed, + Error: err.Error(), + }) + return subagent.Result{}, err + } + + if err := worker.Start(input.Task, input.Budget, input.Capability); err != nil { + _ = s.emit(ctx, EventSubAgentFailed, input.RunID, input.SessionID, SubAgentEventPayload{ + Role: input.Role, + TaskID: input.Task.ID, + State: subagent.StateFailed, + Error: err.Error(), + }) + return subagent.Result{}, err + } + + _ = s.emit(ctx, EventSubAgentStarted, input.RunID, input.SessionID, SubAgentEventPayload{ + Role: input.Role, + TaskID: input.Task.ID, + State: worker.State(), + }) + + for { + stepResult, stepErr := worker.Step(ctx) + _ = s.emit(ctx, EventSubAgentProgress, input.RunID, input.SessionID, SubAgentEventPayload{ + Role: input.Role, + TaskID: input.Task.ID, + State: stepResult.State, + Step: stepResult.Step, + Delta: stepResult.Delta, + Error: errorText(stepErr), + }) + + if stepErr != nil { + result, resultErr := worker.Result() + if resultErr != nil { + _ = s.emit(ctx, EventSubAgentFailed, input.RunID, input.SessionID, SubAgentEventPayload{ + Role: input.Role, + TaskID: input.Task.ID, + State: subagent.StateFailed, + Error: stepErr.Error(), + }) + return subagent.Result{}, stepErr + } + emitSubAgentTerminal(s, ctx, input, result) + return result, stepErr + } + + if !stepResult.Done { + continue + } + + result, err := worker.Result() + if err != nil { + _ = s.emit(ctx, EventSubAgentFailed, input.RunID, input.SessionID, SubAgentEventPayload{ + Role: input.Role, + TaskID: input.Task.ID, + State: subagent.StateFailed, + Error: err.Error(), + }) + return subagent.Result{}, err + } + emitSubAgentTerminal(s, ctx, input, result) + if result.State == subagent.StateSucceeded { + return result, nil + } + return result, errors.New(result.Error) + } +} + +// emitSubAgentTerminal 按子代理终态发射最终事件。 +func emitSubAgentTerminal(s *Service, ctx context.Context, input SubAgentTaskInput, result subagent.Result) { + payload := SubAgentEventPayload{ + Role: result.Role, + TaskID: result.TaskID, + State: result.State, + StopReason: result.StopReason, + Step: result.StepCount, + Error: strings.TrimSpace(result.Error), + } + + switch result.State { + case subagent.StateSucceeded: + _ = s.emit(ctx, EventSubAgentCompleted, input.RunID, input.SessionID, payload) + case subagent.StateCanceled: + _ = s.emit(ctx, EventSubAgentCanceled, input.RunID, input.SessionID, payload) + default: + _ = s.emit(ctx, EventSubAgentFailed, input.RunID, input.SessionID, payload) + } +} + +// errorText 将 error 安全转换为事件可用文本。 +func errorText(err error) string { + if err == nil { + return "" + } + return strings.TrimSpace(err.Error()) +} diff --git a/internal/runtime/subagent_run_test.go b/internal/runtime/subagent_run_test.go new file mode 100644 index 00000000..fa66d19b --- /dev/null +++ b/internal/runtime/subagent_run_test.go @@ -0,0 +1,154 @@ +package runtime + +import ( + "context" + "errors" + "testing" + + "neo-code/internal/subagent" +) + +type failingSubAgentFactory struct { + err error +} + +func (f failingSubAgentFactory) Create(role subagent.Role) (subagent.WorkerRuntime, error) { + return nil, f.err +} + +func TestServiceRunSubAgentTaskSuccess(t *testing.T) { + t.Parallel() + + service := NewWithFactory(nil, nil, nil, nil, nil) + service.SetSubAgentFactory(subagent.NewWorkerFactory(func(role subagent.Role, policy subagent.RolePolicy) subagent.Engine { + return subagent.EngineFunc(func(ctx context.Context, input subagent.StepInput) (subagent.StepOutput, error) { + if input.StepIndex == 1 { + return subagent.StepOutput{ + Delta: "step-1", + Done: false, + }, nil + } + return subagent.StepOutput{ + Delta: "step-2", + Done: true, + Output: subagent.Output{ + Summary: "task completed", + Findings: []string{"f1"}, + Patches: []string{"p1"}, + Risks: []string{"r1"}, + NextActions: []string{"n1"}, + Artifacts: []string{"a1"}, + }, + }, nil + }) + })) + + result, err := service.RunSubAgentTask(context.Background(), SubAgentTaskInput{ + RunID: "sub-run-success", + SessionID: "session-1", + Role: subagent.RoleCoder, + Task: subagent.Task{ + ID: "task-1", + Goal: "implement feature", + }, + Budget: subagent.Budget{ + MaxSteps: 3, + }, + }) + if err != nil { + t.Fatalf("RunSubAgentTask() error = %v", err) + } + if result.State != subagent.StateSucceeded { + t.Fatalf("result state = %q, want %q", result.State, subagent.StateSucceeded) + } + if result.StepCount != 2 { + t.Fatalf("result step count = %d, want 2", result.StepCount) + } + + events := collectRuntimeEvents(service.Events()) + assertEventSequence(t, events, []EventType{ + EventSubAgentStarted, + EventSubAgentProgress, + EventSubAgentProgress, + EventSubAgentCompleted, + }) + assertEventsRunID(t, events, "sub-run-success") +} + +func TestServiceRunSubAgentTaskFailureFlows(t *testing.T) { + t.Parallel() + + t.Run("factory create failed", func(t *testing.T) { + t.Parallel() + + service := NewWithFactory(nil, nil, nil, nil, nil) + service.SetSubAgentFactory(failingSubAgentFactory{err: errors.New("create failed")}) + _, err := service.RunSubAgentTask(context.Background(), SubAgentTaskInput{ + RunID: "sub-run-factory-failed", + Role: subagent.RoleResearcher, + Task: subagent.Task{ + ID: "task-f", + Goal: "research", + }, + }) + if err == nil { + t.Fatalf("expected create error") + } + events := collectRuntimeEvents(service.Events()) + assertEventSequence(t, events, []EventType{EventSubAgentFailed}) + }) + + t.Run("worker step failed", func(t *testing.T) { + t.Parallel() + + service := NewWithFactory(nil, nil, nil, nil, nil) + service.SetSubAgentFactory(subagent.NewWorkerFactory(func(role subagent.Role, policy subagent.RolePolicy) subagent.Engine { + return subagent.EngineFunc(func(ctx context.Context, input subagent.StepInput) (subagent.StepOutput, error) { + return subagent.StepOutput{}, errors.New("step failed") + }) + })) + _, err := service.RunSubAgentTask(context.Background(), SubAgentTaskInput{ + RunID: "sub-run-step-failed", + Role: subagent.RoleReviewer, + Task: subagent.Task{ + ID: "task-step-f", + Goal: "review", + }, + }) + if err == nil { + t.Fatalf("expected step error") + } + events := collectRuntimeEvents(service.Events()) + assertEventSequence(t, events, []EventType{ + EventSubAgentStarted, + EventSubAgentProgress, + EventSubAgentFailed, + }) + }) +} + +func TestServiceRunSubAgentTaskInputValidation(t *testing.T) { + t.Parallel() + + service := NewWithFactory(nil, nil, nil, nil, nil) + if _, err := service.RunSubAgentTask(context.Background(), SubAgentTaskInput{ + Role: subagent.RoleCoder, + Task: subagent.Task{ + ID: "task", + Goal: "goal", + }, + }); err == nil { + t.Fatalf("expected empty run id error") + } + + if _, err := service.RunSubAgentTask(context.Background(), SubAgentTaskInput{ + RunID: "sub-run-invalid-role", + Role: subagent.Role("x"), + Task: subagent.Task{ + ID: "task", + Goal: "goal", + }, + }); err == nil { + t.Fatalf("expected invalid role error") + } +} diff --git a/internal/subagent/engine.go b/internal/subagent/engine.go new file mode 100644 index 00000000..6d849558 --- /dev/null +++ b/internal/subagent/engine.go @@ -0,0 +1,29 @@ +package subagent + +import ( + "context" + "strings" +) + +// defaultEngine 提供一个可运行的默认单步完成引擎，便于工厂与测试快速装配。 +type defaultEngine struct{} + +// RunStep 执行默认单步逻辑并返回结构化结果。 +func (defaultEngine) RunStep(ctx context.Context, input StepInput) (StepOutput, error) { + if err := ctx.Err(); err != nil { + return StepOutput{}, err + } + + summary := strings.TrimSpace(input.Task.ExpectedOutput) + if summary == "" { + summary = strings.TrimSpace(input.Task.Goal) + } + + return StepOutput{ + Delta: "default engine completed", + Done: true, + Output: Output{ + Summary: summary, + }, + }, nil +} diff --git a/internal/subagent/factory.go b/internal/subagent/factory.go new file mode 100644 index 00000000..3b4b19ba --- /dev/null +++ b/internal/subagent/factory.go @@ -0,0 +1,36 @@ +package subagent + +import "time" + +// EngineBuilder 定义基于角色策略构建执行引擎的工厂函数。 +type EngineBuilder func(role Role, policy RolePolicy) Engine + +// WorkerFactory 是默认的 WorkerRuntime 工厂实现。 +type WorkerFactory struct { + builder EngineBuilder +} + +// NewWorkerFactory 创建 WorkerFactory；当 builder 为空时使用默认引擎。 +func NewWorkerFactory(builder EngineBuilder) *WorkerFactory { + return &WorkerFactory{builder: builder} +} + +// Create 基于角色创建对应策略与执行引擎的 WorkerRuntime。 +func (f *WorkerFactory) Create(role Role) (WorkerRuntime, error) { + policy, err := DefaultRolePolicy(role) + if err != nil { + return nil, err + } + // 兜底保证默认预算始终可用。 + policy.DefaultBudget = policy.DefaultBudget.normalize(Budget{ + MaxSteps: defaultPolicyMaxSteps, + Timeout: defaultPolicyTimeout * time.Second, + }) + + var engine Engine + if f != nil && f.builder != nil { + engine = f.builder(role, policy) + } + return NewWorker(role, policy, engine) +} + diff --git a/internal/subagent/output_contract.go b/internal/subagent/output_contract.go new file mode 100644 index 00000000..4fe063ee --- /dev/null +++ b/internal/subagent/output_contract.go @@ -0,0 +1,27 @@ +package subagent + +import "strings" + +var supportedOutputSections = map[string]struct{}{ + "summary": {}, + "findings": {}, + "patches": {}, + "risks": {}, + "next_actions": {}, + "artifacts": {}, +} + +// validateOutputContract 校验输出结构是否满足角色策略要求。 +func validateOutputContract(policy RolePolicy, output Output) error { + out := output.normalize() + for _, section := range dedupeAndTrim(policy.RequiredSections) { + key := strings.ToLower(strings.TrimSpace(section)) + if _, ok := supportedOutputSections[key]; !ok { + return errorsf("unsupported required output section %q", section) + } + if key == "summary" && strings.TrimSpace(out.Summary) == "" { + return errorsf("output summary is required") + } + } + return nil +} diff --git a/internal/subagent/policy.go b/internal/subagent/policy.go new file mode 100644 index 00000000..7a5b3d62 --- /dev/null +++ b/internal/subagent/policy.go @@ -0,0 +1,86 @@ +package subagent + +import ( + "strings" + "time" +) + +const ( + defaultPolicyMaxSteps = 6 + defaultPolicyTimeout = 30 +) + +// RolePolicy 定义不同角色的执行策略。 +type RolePolicy struct { + Role Role + SystemPrompt string + AllowedTools []string + DefaultBudget Budget + RequiredSections []string +} + +// Validate 校验角色策略是否合法。 +func (p RolePolicy) Validate() error { + if !p.Role.Valid() { + return errorsf("invalid policy role %q", p.Role) + } + if strings.TrimSpace(p.SystemPrompt) == "" { + return errorsf("role policy prompt is required") + } + if len(dedupeAndTrim(p.AllowedTools)) == 0 { + return errorsf("role policy allowed tools is empty") + } + if len(dedupeAndTrim(p.RequiredSections)) == 0 { + return errorsf("role policy required sections is empty") + } + if err := validateOutputContract(p, Output{Summary: "probe"}); err != nil { + return err + } + return nil +} + +// DefaultRolePolicy 返回内置角色策略。 +func DefaultRolePolicy(role Role) (RolePolicy, error) { + if !role.Valid() { + return RolePolicy{}, errorsf("unsupported role %q", role) + } + + policy := RolePolicy{ + Role: role, + DefaultBudget: Budget{ + MaxSteps: defaultPolicyMaxSteps, + Timeout: defaultPolicyTimeout * time.Second, + }, + RequiredSections: []string{ + "summary", + "findings", + "patches", + "risks", + "next_actions", + "artifacts", + }, + } + + switch role { + case RoleResearcher: + policy.SystemPrompt = "你是研究型子代理，负责检索证据并形成结论。" + policy.AllowedTools = []string{"filesystem_read_file", "filesystem_glob", "filesystem_grep", "webfetch"} + case RoleCoder: + policy.SystemPrompt = "你是实现型子代理，负责修改代码并给出验证结果。" + policy.AllowedTools = []string{ + "filesystem_read_file", + "filesystem_write_file", + "filesystem_edit", + "filesystem_glob", + "filesystem_grep", + "bash", + } + case RoleReviewer: + policy.SystemPrompt = "你是审查型子代理，负责识别缺陷、风险与测试缺口。" + policy.AllowedTools = []string{"filesystem_read_file", "filesystem_glob", "filesystem_grep"} + } + + policy.AllowedTools = dedupeAndTrim(policy.AllowedTools) + policy.RequiredSections = dedupeAndTrim(policy.RequiredSections) + return policy, nil +} diff --git a/internal/subagent/policy_test.go b/internal/subagent/policy_test.go new file mode 100644 index 00000000..08721d37 --- /dev/null +++ b/internal/subagent/policy_test.go @@ -0,0 +1,124 @@ +package subagent + +import "testing" + +func TestDefaultRolePolicy(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + role Role + }{ + {name: "researcher", role: RoleResearcher}, + {name: "coder", role: RoleCoder}, + {name: "reviewer", role: RoleReviewer}, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + policy, err := DefaultRolePolicy(tt.role) + if err != nil { + t.Fatalf("DefaultRolePolicy() error = %v", err) + } + if policy.Role != tt.role { + t.Fatalf("policy role = %q, want %q", policy.Role, tt.role) + } + if policy.DefaultBudget.MaxSteps <= 0 { + t.Fatalf("invalid max steps: %d", policy.DefaultBudget.MaxSteps) + } + if len(policy.AllowedTools) == 0 { + t.Fatalf("expected non-empty allowed tools") + } + if len(policy.RequiredSections) == 0 { + t.Fatalf("expected non-empty required sections") + } + if err := policy.Validate(); err != nil { + t.Fatalf("policy.Validate() error = %v", err) + } + }) + } +} + +func TestDefaultRolePolicyInvalidRole(t *testing.T) { + t.Parallel() + + if _, err := DefaultRolePolicy(Role("unknown")); err == nil { + t.Fatalf("expected invalid role error") + } +} + +func TestRolePolicyValidate(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + policy RolePolicy + wantErr bool + }{ + { + name: "valid", + policy: RolePolicy{ + Role: RoleResearcher, + SystemPrompt: "prompt", + AllowedTools: []string{"filesystem_grep"}, + RequiredSections: []string{"summary"}, + }, + }, + { + name: "empty prompt", + policy: RolePolicy{ + Role: RoleResearcher, + AllowedTools: []string{"filesystem_grep"}, + RequiredSections: []string{"summary"}, + }, + wantErr: true, + }, + { + name: "empty tools", + policy: RolePolicy{ + Role: RoleResearcher, + SystemPrompt: "prompt", + RequiredSections: []string{"summary"}, + }, + wantErr: true, + }, + { + name: "invalid role", + policy: RolePolicy{ + Role: Role("x"), + SystemPrompt: "prompt", + AllowedTools: []string{"filesystem_grep"}, + RequiredSections: []string{"summary"}, + }, + wantErr: true, + }, + { + name: "unsupported required section", + policy: RolePolicy{ + Role: RoleResearcher, + SystemPrompt: "prompt", + AllowedTools: []string{"filesystem_grep"}, + RequiredSections: []string{"unknown_section"}, + }, + wantErr: true, + }, + } + + for _, tt := range tests { + tt := tt + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + err := tt.policy.Validate() + if tt.wantErr && err == nil { + t.Fatalf("expected error") + } + if !tt.wantErr && err != nil { + t.Fatalf("unexpected error: %v", err) + } + }) + } +} diff --git a/internal/subagent/types.go b/internal/subagent/types.go new file mode 100644 index 00000000..9e3ba823 --- /dev/null +++ b/internal/subagent/types.go @@ -0,0 +1,224 @@ +package subagent + +import ( + "context" + "fmt" + "strings" + "time" +) + +// Role 表示子代理的执行角色。 +type Role string + +const ( + // RoleResearcher 用于检索与分析任务。 + RoleResearcher Role = "researcher" + // RoleCoder 用于实现与修复任务。 + RoleCoder Role = "coder" + // RoleReviewer 用于审查与验收任务。 + RoleReviewer Role = "reviewer" +) + +// Valid 判断角色是否受支持。 +func (r Role) Valid() bool { + switch r { + case RoleResearcher, RoleCoder, RoleReviewer: + return true + default: + return false + } +} + +// Budget 描述子代理执行预算。 +type Budget struct { + MaxSteps int + Timeout time.Duration +} + +// normalize 归一化预算并应用默认值。 +func (b Budget) normalize(defaults Budget) Budget { + out := b + if out.MaxSteps <= 0 { + out.MaxSteps = defaults.MaxSteps + } + if out.MaxSteps <= 0 { + out.MaxSteps = 6 + } + if out.Timeout <= 0 { + out.Timeout = defaults.Timeout + } + if out.Timeout <= 0 { + out.Timeout = 30 * time.Second + } + return out +} + +// Capability 描述子代理运行时可用能力边界。 +type Capability struct { + AllowedTools []string + AllowedPaths []string +} + +// normalize 归一化能力列表并去重。 +func (c Capability) normalize() Capability { + return Capability{ + AllowedTools: dedupeAndTrim(c.AllowedTools), + AllowedPaths: dedupeAndTrim(c.AllowedPaths), + } +} + +// Task 表示单个子代理任务输入。 +type Task struct { + ID string + Goal string + ExpectedOutput string + Workspace string +} + +// Validate 校验任务输入是否合法。 +func (t Task) Validate() error { + if strings.TrimSpace(t.ID) == "" { + return errorsf("task id is required") + } + if strings.TrimSpace(t.Goal) == "" { + return errorsf("task goal is required") + } + return nil +} + +// StopReason 表示子代理终止原因。 +type StopReason string + +const ( + // StopReasonCompleted 表示正常完成。 + StopReasonCompleted StopReason = "completed" + // StopReasonCanceled 表示被取消。 + StopReasonCanceled StopReason = "canceled" + // StopReasonTimeout 表示执行超时。 + StopReasonTimeout StopReason = "timeout" + // StopReasonMaxSteps 表示达到步数上限。 + StopReasonMaxSteps StopReason = "max_steps" + // StopReasonError 表示执行错误。 + StopReasonError StopReason = "error" +) + +// State 表示子代理生命周期状态。 +type State string + +const ( + // StateIdle 表示尚未启动。 + StateIdle State = "idle" + // StateRunning 表示执行中。 + StateRunning State = "running" + // StateSucceeded 表示执行成功结束。 + StateSucceeded State = "succeeded" + // StateFailed 表示执行失败结束。 + StateFailed State = "failed" + // StateCanceled 表示被取消结束。 + StateCanceled State = "canceled" +) + +// Terminal 判断当前状态是否为终态。 +func (s State) Terminal() bool { + switch s { + case StateSucceeded, StateFailed, StateCanceled: + return true + default: + return false + } +} + +// Output 定义子代理标准结构化输出。 +type Output struct { + Summary string + Findings []string + Patches []string + Risks []string + NextActions []string + Artifacts []string +} + +// normalize 归一化输出，避免重复与空项。 +func (o Output) normalize() Output { + o.Summary = strings.TrimSpace(o.Summary) + o.Findings = dedupeAndTrim(o.Findings) + o.Patches = dedupeAndTrim(o.Patches) + o.Risks = dedupeAndTrim(o.Risks) + o.NextActions = dedupeAndTrim(o.NextActions) + o.Artifacts = dedupeAndTrim(o.Artifacts) + return o +} + +// StepInput 表示单步执行输入。 +type StepInput struct { + Role Role + Policy RolePolicy + Task Task + Budget Budget + Capability Capability + StepIndex int + Trace []string +} + +// StepOutput 表示单步执行输出。 +type StepOutput struct { + Delta string + Done bool + Output Output +} + +// StepResult 表示一次 Step 的可观测结果。 +type StepResult struct { + State State + Done bool + Step int + Delta string +} + +// Result 描述子代理完成后的结构化结果。 +type Result struct { + Role Role + TaskID string + State State + StopReason StopReason + StartedAt time.Time + EndedAt time.Time + StepCount int + Budget Budget + Capability Capability + Output Output + Error string +} + +// Engine 定义 WorkerRuntime 的单步执行引擎。 +type Engine interface { + RunStep(ctx context.Context, input StepInput) (StepOutput, error) +} + +// EngineFunc 允许用函数实现 Engine。 +type EngineFunc func(ctx context.Context, input StepInput) (StepOutput, error) + +// RunStep 执行函数式引擎逻辑。 +func (f EngineFunc) RunStep(ctx context.Context, input StepInput) (StepOutput, error) { + return f(ctx, input) +} + +// WorkerRuntime 定义子代理执行生命周期接口。 +type WorkerRuntime interface { + Start(task Task, budget Budget, capability Capability) error + Step(ctx context.Context) (StepResult, error) + Stop(reason StopReason) error + Result() (Result, error) + State() State + Policy() RolePolicy +} + +// Factory 定义 runtime 侧创建 WorkerRuntime 的入口。 +type Factory interface { + Create(role Role) (WorkerRuntime, error) +} + +// errorsf 统一组装 subagent 模块错误前缀。 +func errorsf(format string, args ...any) error { + return fmt.Errorf("subagent: "+format, args...) +} diff --git a/internal/subagent/util.go b/internal/subagent/util.go new file mode 100644 index 00000000..a87acc25 --- /dev/null +++ b/internal/subagent/util.go @@ -0,0 +1,28 @@ +package subagent + +import "strings" + +// dedupeAndTrim 对字符串切片做去空白、去重并保持原顺序。 +func dedupeAndTrim(items []string) []string { + if len(items) == 0 { + return nil + } + result := make([]string, 0, len(items)) + seen := make(map[string]struct{}, len(items)) + for _, raw := range items { + item := strings.TrimSpace(raw) + if item == "" { + continue + } + key := strings.ToLower(item) + if _, ok := seen[key]; ok { + continue + } + seen[key] = struct{}{} + result = append(result, item) + } + if len(result) == 0 { + return nil + } + return result +} diff --git a/internal/subagent/worker.go b/internal/subagent/worker.go new file mode 100644 index 00000000..995f4501 --- /dev/null +++ b/internal/subagent/worker.go @@ -0,0 +1,259 @@ +package subagent + +import ( + "context" + "errors" + "strings" + "sync" + "time" +) + +// worker 是 WorkerRuntime 的默认实现，负责封装单任务生命周期。 +type worker struct { + mu sync.RWMutex + role Role + policy RolePolicy + engine Engine + state State + task Task + budget Budget + capability Capability + stepCount int + trace []string + startedAt time.Time + endedAt time.Time + stopReason StopReason + output Output + lastErr string +} + +// NewWorker 根据角色、策略与引擎创建一个 WorkerRuntime 实例。 +func NewWorker(role Role, policy RolePolicy, engine Engine) (WorkerRuntime, error) { + if !role.Valid() { + return nil, errorsf("invalid role %q", role) + } + if policy.Role == "" { + policy.Role = role + } + if policy.Role != role { + return nil, errorsf("policy role %q does not match worker role %q", policy.Role, role) + } + if err := policy.Validate(); err != nil { + return nil, err + } + if engine == nil { + engine = defaultEngine{} + } + + return &worker{ + role: role, + policy: policy, + engine: engine, + state: StateIdle, + }, nil +} + +// Start 初始化任务执行上下文并进入运行态。 +func (w *worker) Start(task Task, budget Budget, capability Capability) error { + if w == nil { + return errors.New("subagent: worker is nil") + } + if err := task.Validate(); err != nil { + return err + } + + w.mu.Lock() + defer w.mu.Unlock() + if w.state != StateIdle { + return errorsf("worker already started") + } + + w.task = task + w.budget = budget.normalize(w.policy.DefaultBudget) + w.capability = capability.normalize() + w.trace = nil + w.stepCount = 0 + w.output = Output{} + w.lastErr = "" + w.stopReason = "" + w.startedAt = time.Now() + w.endedAt = time.Time{} + w.state = StateRunning + return nil +} + +// Step 执行一次引擎步骤，并在满足终止条件时更新终态结果。 +func (w *worker) Step(ctx context.Context) (StepResult, error) { + if w == nil { + return StepResult{}, errors.New("subagent: worker is nil") + } + if err := ctx.Err(); err != nil { + return StepResult{}, err + } + + w.mu.Lock() + if w.state != StateRunning { + state := w.state + w.mu.Unlock() + return StepResult{}, errorsf("worker is not running, current state=%s", state) + } + if w.budget.Timeout > 0 && time.Since(w.startedAt) >= w.budget.Timeout { + result := w.finishLocked(StateFailed, StopReasonTimeout, Output{}, errorsf("worker timeout")) + w.mu.Unlock() + return StepResult{State: result.State, Done: true, Step: result.StepCount}, nil + } + if w.stepCount >= w.budget.MaxSteps { + result := w.finishLocked(StateFailed, StopReasonMaxSteps, Output{}, errorsf("worker reached max steps")) + w.mu.Unlock() + return StepResult{State: result.State, Done: true, Step: result.StepCount}, nil + } + + input := StepInput{ + Role: w.role, + Policy: w.policy, + Task: w.task, + Budget: w.budget, + Capability: w.capability, + StepIndex: w.stepCount + 1, + Trace: append([]string(nil), w.trace...), + } + w.mu.Unlock() + + stepOutput, err := w.engine.RunStep(ctx, input) + if err != nil { + w.mu.Lock() + result := w.finishLocked(StateFailed, StopReasonError, Output{}, err) + w.mu.Unlock() + return StepResult{State: result.State, Done: true, Step: result.StepCount}, err + } + + w.mu.Lock() + defer w.mu.Unlock() + if w.state != StateRunning { + return StepResult{}, errorsf("worker is not running, current state=%s", w.state) + } + + w.stepCount++ + delta := strings.TrimSpace(stepOutput.Delta) + if delta != "" { + w.trace = append(w.trace, delta) + } + + if stepOutput.Done { + if err := validateOutputContract(w.policy, stepOutput.Output); err != nil { + result := w.finishLocked(StateFailed, StopReasonError, Output{}, err) + return StepResult{State: result.State, Done: true, Step: result.StepCount, Delta: delta}, err + } + result := w.finishLocked(StateSucceeded, StopReasonCompleted, stepOutput.Output, nil) + return StepResult{State: result.State, Done: true, Step: result.StepCount, Delta: delta}, nil + } + + if w.stepCount >= w.budget.MaxSteps { + result := w.finishLocked(StateFailed, StopReasonMaxSteps, Output{}, errorsf("worker reached max steps")) + return StepResult{State: result.State, Done: true, Step: result.StepCount, Delta: delta}, nil + } + + return StepResult{ + State: StateRunning, + Done: false, + Step: w.stepCount, + Delta: delta, + }, nil +} + +// Stop 主动终止运行中的 worker，并按终止原因映射最终状态。 +func (w *worker) Stop(reason StopReason) error { + if w == nil { + return errors.New("subagent: worker is nil") + } + if strings.TrimSpace(string(reason)) == "" { + return errorsf("stop reason is required") + } + + w.mu.Lock() + defer w.mu.Unlock() + if w.state != StateRunning { + if w.state.Terminal() { + return nil + } + return errorsf("worker is not running, current state=%s", w.state) + } + + switch reason { + case StopReasonCompleted: + if err := validateOutputContract(w.policy, w.output); err != nil { + return err + } + w.finishLocked(StateSucceeded, reason, w.output, nil) + case StopReasonCanceled: + w.finishLocked(StateCanceled, reason, w.output, nil) + case StopReasonTimeout, StopReasonMaxSteps, StopReasonError: + w.finishLocked(StateFailed, reason, w.output, nil) + default: + return errorsf("unsupported stop reason %q", reason) + } + return nil +} + +// Result 返回 worker 的最终结构化结果。 +func (w *worker) Result() (Result, error) { + if w == nil { + return Result{}, errors.New("subagent: worker is nil") + } + + w.mu.RLock() + defer w.mu.RUnlock() + if !w.state.Terminal() { + return Result{}, errorsf("worker is not finished") + } + return w.snapshotLocked(), nil +} + +// State 返回当前 worker 生命周期状态。 +func (w *worker) State() State { + if w == nil { + return StateIdle + } + w.mu.RLock() + defer w.mu.RUnlock() + return w.state +} + +// Policy 返回当前 worker 角色策略副本。 +func (w *worker) Policy() RolePolicy { + if w == nil { + return RolePolicy{} + } + w.mu.RLock() + defer w.mu.RUnlock() + return w.policy +} + +// finishLocked 在持有写锁时将 worker 切换为终态并返回结果快照。 +func (w *worker) finishLocked(state State, reason StopReason, output Output, err error) Result { + w.state = state + w.stopReason = reason + w.endedAt = time.Now() + w.output = output.normalize() + if err != nil { + w.lastErr = err.Error() + } + return w.snapshotLocked() +} + +// snapshotLocked 在持有读锁或写锁时构造稳定结果快照。 +func (w *worker) snapshotLocked() Result { + return Result{ + Role: w.role, + TaskID: w.task.ID, + State: w.state, + StopReason: w.stopReason, + StartedAt: w.startedAt, + EndedAt: w.endedAt, + StepCount: w.stepCount, + Budget: w.budget, + Capability: w.capability, + Output: w.output, + Error: w.lastErr, + } +} diff --git a/internal/subagent/worker_test.go b/internal/subagent/worker_test.go new file mode 100644 index 00000000..ef4e42ce --- /dev/null +++ b/internal/subagent/worker_test.go @@ -0,0 +1,245 @@ +package subagent + +import ( + "context" + "errors" + "testing" + "time" +) + +func TestWorkerLifecycleCompleted(t *testing.T) { + t.Parallel() + + policy, err := DefaultRolePolicy(RoleCoder) + if err != nil { + t.Fatalf("DefaultRolePolicy() error = %v", err) + } + + w, err := NewWorker(RoleCoder, policy, EngineFunc(func(ctx context.Context, input StepInput) (StepOutput, error) { + return StepOutput{ + Delta: "patched files", + Done: true, + Output: Output{ + Summary: "done", + Patches: []string{"a.go"}, + NextActions: []string{"run tests"}, + }, + }, nil + })) + if err != nil { + t.Fatalf("NewWorker() error = %v", err) + } + + err = w.Start(Task{ID: "t1", Goal: "fix bug"}, Budget{MaxSteps: 3}, Capability{ + AllowedTools: []string{"bash", "bash", " "}, + }) + if err != nil { + t.Fatalf("Start() error = %v", err) + } + + step, err := w.Step(context.Background()) + if err != nil { + t.Fatalf("Step() error = %v", err) + } + if !step.Done || step.State != StateSucceeded { + t.Fatalf("unexpected step result: %+v", step) + } + + result, err := w.Result() + if err != nil { + t.Fatalf("Result() error = %v", err) + } + if result.StopReason != StopReasonCompleted { + t.Fatalf("stop reason = %q, want %q", result.StopReason, StopReasonCompleted) + } + if result.StepCount != 1 { + t.Fatalf("step count = %d, want 1", result.StepCount) + } + if len(result.Capability.AllowedTools) != 1 { + t.Fatalf("expected capability dedupe, got %+v", result.Capability) + } +} + +func TestWorkerLifecycleFailures(t *testing.T) { + t.Parallel() + + policy, err := DefaultRolePolicy(RoleResearcher) + if err != nil { + t.Fatalf("DefaultRolePolicy() error = %v", err) + } + + t.Run("engine error", func(t *testing.T) { + t.Parallel() + + w, err := NewWorker(RoleResearcher, policy, EngineFunc(func(ctx context.Context, input StepInput) (StepOutput, error) { + return StepOutput{}, errors.New("boom") + })) + if err != nil { + t.Fatalf("NewWorker() error = %v", err) + } + if err := w.Start(Task{ID: "t2", Goal: "research"}, Budget{MaxSteps: 2}, Capability{}); err != nil { + t.Fatalf("Start() error = %v", err) + } + if _, err := w.Step(context.Background()); err == nil { + t.Fatalf("expected step error") + } + result, err := w.Result() + if err != nil { + t.Fatalf("Result() error = %v", err) + } + if result.State != StateFailed || result.StopReason != StopReasonError { + t.Fatalf("unexpected result: %+v", result) + } + }) + + t.Run("max steps", func(t *testing.T) { + t.Parallel() + + w, err := NewWorker(RoleResearcher, policy, EngineFunc(func(ctx context.Context, input StepInput) (StepOutput, error) { + return StepOutput{Delta: "not done", Done: false}, nil + })) + if err != nil { + t.Fatalf("NewWorker() error = %v", err) + } + if err := w.Start(Task{ID: "t3", Goal: "research"}, Budget{MaxSteps: 1}, Capability{}); err != nil { + t.Fatalf("Start() error = %v", err) + } + step, err := w.Step(context.Background()) + if err != nil { + t.Fatalf("Step() error = %v", err) + } + if !step.Done || step.State != StateFailed { + t.Fatalf("expected first step to finish by max steps, got %+v", step) + } + + result, err := w.Result() + if err != nil { + t.Fatalf("Result() error = %v", err) + } + if result.StopReason != StopReasonMaxSteps { + t.Fatalf("stop reason = %q, want %q", result.StopReason, StopReasonMaxSteps) + } + }) + + t.Run("timeout", func(t *testing.T) { + t.Parallel() + + w, err := NewWorker(RoleResearcher, policy, EngineFunc(func(ctx context.Context, input StepInput) (StepOutput, error) { + return StepOutput{Done: false}, nil + })) + if err != nil { + t.Fatalf("NewWorker() error = %v", err) + } + if err := w.Start(Task{ID: "t4", Goal: "research"}, Budget{MaxSteps: 5, Timeout: time.Nanosecond}, Capability{}); err != nil { + t.Fatalf("Start() error = %v", err) + } + time.Sleep(2 * time.Millisecond) + + step, err := w.Step(context.Background()) + if err != nil { + t.Fatalf("Step() error = %v", err) + } + if !step.Done || step.State != StateFailed { + t.Fatalf("unexpected timeout step: %+v", step) + } + result, err := w.Result() + if err != nil { + t.Fatalf("Result() error = %v", err) + } + if result.StopReason != StopReasonTimeout { + t.Fatalf("stop reason = %q, want %q", result.StopReason, StopReasonTimeout) + } + }) +} + +func TestWorkerStopAndGuards(t *testing.T) { + t.Parallel() + + policy, err := DefaultRolePolicy(RoleReviewer) + if err != nil { + t.Fatalf("DefaultRolePolicy() error = %v", err) + } + + w, err := NewWorker(RoleReviewer, policy, nil) + if err != nil { + t.Fatalf("NewWorker() error = %v", err) + } + + if _, err := w.Result(); err == nil { + t.Fatalf("expected result before finish to fail") + } + if _, err := w.Step(context.Background()); err == nil { + t.Fatalf("expected step before start to fail") + } + if err := w.Start(Task{ID: "review", Goal: "review"}, Budget{}, Capability{}); err != nil { + t.Fatalf("Start() error = %v", err) + } + if err := w.Start(Task{ID: "review2", Goal: "review2"}, Budget{}, Capability{}); err == nil { + t.Fatalf("expected double start to fail") + } + if err := w.Stop(StopReasonCanceled); err != nil { + t.Fatalf("Stop() error = %v", err) + } + if w.State() != StateCanceled { + t.Fatalf("state = %q, want %q", w.State(), StateCanceled) + } + if err := w.Stop(StopReasonCanceled); err != nil { + t.Fatalf("terminal stop should be idempotent, got %v", err) + } +} + +func TestWorkerFactoryCreate(t *testing.T) { + t.Parallel() + + factory := NewWorkerFactory(func(role Role, policy RolePolicy) Engine { + return EngineFunc(func(ctx context.Context, input StepInput) (StepOutput, error) { + return StepOutput{Done: true, Output: Output{Summary: "ok"}}, nil + }) + }) + + w, err := factory.Create(RoleCoder) + if err != nil { + t.Fatalf("Create() error = %v", err) + } + if w.Policy().Role != RoleCoder { + t.Fatalf("policy role = %q, want %q", w.Policy().Role, RoleCoder) + } + + if _, err := factory.Create(Role("invalid")); err == nil { + t.Fatalf("expected invalid role create to fail") + } +} + +func TestWorkerRejectsInvalidOutputContract(t *testing.T) { + t.Parallel() + + policy, err := DefaultRolePolicy(RoleCoder) + if err != nil { + t.Fatalf("DefaultRolePolicy() error = %v", err) + } + w, err := NewWorker(RoleCoder, policy, EngineFunc(func(ctx context.Context, input StepInput) (StepOutput, error) { + return StepOutput{ + Done: true, + Output: Output{ + Summary: " ", + }, + }, nil + })) + if err != nil { + t.Fatalf("NewWorker() error = %v", err) + } + if err := w.Start(Task{ID: "t-invalid-output", Goal: "goal"}, Budget{MaxSteps: 3}, Capability{}); err != nil { + t.Fatalf("Start() error = %v", err) + } + + if _, err := w.Step(context.Background()); err == nil { + t.Fatalf("expected invalid output contract error") + } + result, err := w.Result() + if err != nil { + t.Fatalf("Result() error = %v", err) + } + if result.State != StateFailed || result.StopReason != StopReasonError { + t.Fatalf("unexpected result: %+v", result) + } +} From 9384e801bbbce6bd8024dcc4092b432e7c0389bd Mon Sep 17 00:00:00 2001 From: xgopilot Date: Tue, 14 Apr 2026 16:01:32 +0000 Subject: [PATCH 02/10] fix(subagent): resolve review conflicts and strengthen coverage Generated with [codeagent](https://github.com/qbox/codeagent) Co-authored-by: Cai-Tang-www <106404101+Cai-Tang-www@users.noreply.github.com> --- internal/runtime/subagent_factory_test.go | 5 + internal/runtime/subagent_run.go | 49 ++++++-- internal/runtime/subagent_run_test.go | 79 +++++++++++++ internal/subagent/engine_test.go | 59 ++++++++++ internal/subagent/output_contract.go | 46 +++++++- internal/subagent/output_contract_test.go | 74 ++++++++++++ internal/subagent/policy.go | 2 +- internal/subagent/worker.go | 57 +++++++++- internal/subagent/worker_test.go | 133 +++++++++++++++++++++- 9 files changed, 486 insertions(+), 18 deletions(-) create mode 100644 internal/subagent/engine_test.go create mode 100644 internal/subagent/output_contract_test.go diff --git a/internal/runtime/subagent_factory_test.go b/internal/runtime/subagent_factory_test.go index eea246b1..70256ce4 100644 --- a/internal/runtime/subagent_factory_test.go +++ b/internal/runtime/subagent_factory_test.go @@ -30,4 +30,9 @@ func TestServiceSubAgentFactoryRegistration(t *testing.T) { if svc.SubAgentFactory() == nil { t.Fatalf("expected reset to default sub-agent factory") } + + svc.subAgentFactory = nil + if svc.SubAgentFactory() == nil { + t.Fatalf("expected lazy init default sub-agent factory") + } } diff --git a/internal/runtime/subagent_run.go b/internal/runtime/subagent_run.go index 45169c03..bf0adbd8 100644 --- a/internal/runtime/subagent_run.go +++ b/internal/runtime/subagent_run.go @@ -64,16 +64,28 @@ func (s *Service) RunSubAgentTask(ctx context.Context, input SubAgentTaskInput) for { stepResult, stepErr := worker.Step(ctx) - _ = s.emit(ctx, EventSubAgentProgress, input.RunID, input.SessionID, SubAgentEventPayload{ - Role: input.Role, - TaskID: input.Task.ID, - State: stepResult.State, - Step: stepResult.Step, - Delta: stepResult.Delta, - Error: errorText(stepErr), - }) + if stepResult.State == "" { + stepResult.State = worker.State() + } + emitSubAgentProgress(s, input, stepResult, stepErr) if stepErr != nil { + if errors.Is(stepErr, context.Canceled) || errors.Is(stepErr, context.DeadlineExceeded) { + _ = worker.Stop(subagent.StopReasonCanceled) + result, resultErr := worker.Result() + if resultErr != nil { + result = subagent.Result{ + Role: input.Role, + TaskID: input.Task.ID, + State: subagent.StateCanceled, + StopReason: subagent.StopReasonCanceled, + Error: errorText(stepErr), + } + } + emitSubAgentTerminal(s, ctx, input, result) + return result, stepErr + } + result, resultErr := worker.Result() if resultErr != nil { _ = s.emit(ctx, EventSubAgentFailed, input.RunID, input.SessionID, SubAgentEventPayload{ @@ -110,6 +122,27 @@ func (s *Service) RunSubAgentTask(ctx context.Context, input SubAgentTaskInput) } } +// emitSubAgentProgress 非阻塞发射进度事件，避免慢消费者反压执行路径。 +func emitSubAgentProgress(s *Service, input SubAgentTaskInput, stepResult subagent.StepResult, stepErr error) { + payload := SubAgentEventPayload{ + Role: input.Role, + TaskID: input.Task.ID, + State: stepResult.State, + Step: stepResult.Step, + Delta: stepResult.Delta, + Error: errorText(stepErr), + } + select { + case s.events <- RuntimeEvent{ + Type: EventSubAgentProgress, + RunID: input.RunID, + SessionID: input.SessionID, + Payload: payload, + }: + default: + } +} + // emitSubAgentTerminal 按子代理终态发射最终事件。 func emitSubAgentTerminal(s *Service, ctx context.Context, input SubAgentTaskInput, result subagent.Result) { payload := SubAgentEventPayload{ diff --git a/internal/runtime/subagent_run_test.go b/internal/runtime/subagent_run_test.go index fa66d19b..116abaf2 100644 --- a/internal/runtime/subagent_run_test.go +++ b/internal/runtime/subagent_run_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "testing" + "time" "neo-code/internal/subagent" ) @@ -125,6 +126,70 @@ func TestServiceRunSubAgentTaskFailureFlows(t *testing.T) { EventSubAgentFailed, }) }) + + t.Run("context canceled should emit canceled", func(t *testing.T) { + t.Parallel() + + service := NewWithFactory(nil, nil, nil, nil, nil) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + service.SetSubAgentFactory(subagent.NewWorkerFactory(func(role subagent.Role, policy subagent.RolePolicy) subagent.Engine { + return subagent.EngineFunc(func(ctx context.Context, input subagent.StepInput) (subagent.StepOutput, error) { + cancel() + <-ctx.Done() + return subagent.StepOutput{}, ctx.Err() + }) + })) + + result, err := service.RunSubAgentTask(ctx, SubAgentTaskInput{ + RunID: "sub-run-canceled", + Role: subagent.RoleReviewer, + Task: subagent.Task{ + ID: "task-cancel", + Goal: "review", + }, + }) + if err == nil { + t.Fatalf("expected canceled error") + } + if !errors.Is(err, context.Canceled) { + t.Fatalf("error = %v, want context canceled", err) + } + if result.State != subagent.StateCanceled { + t.Fatalf("result state = %q, want %q", result.State, subagent.StateCanceled) + } + if result.StopReason != subagent.StopReasonCanceled { + t.Fatalf("stop reason = %q, want %q", result.StopReason, subagent.StopReasonCanceled) + } + events := collectRuntimeEvents(service.Events()) + assertEventSequence(t, events, []EventType{ + EventSubAgentStarted, + EventSubAgentProgress, + EventSubAgentCanceled, + }) + }) + + t.Run("worker start failed by disallowed capability", func(t *testing.T) { + t.Parallel() + + service := NewWithFactory(nil, nil, nil, nil, nil) + _, err := service.RunSubAgentTask(context.Background(), SubAgentTaskInput{ + RunID: "sub-run-start-failed", + Role: subagent.RoleReviewer, + Task: subagent.Task{ + ID: "task-start-failed", + Goal: "review", + }, + Capability: subagent.Capability{ + AllowedTools: []string{"bash"}, + }, + }) + if err == nil { + t.Fatalf("expected start error") + } + events := collectRuntimeEvents(service.Events()) + assertEventSequence(t, events, []EventType{EventSubAgentFailed}) + }) } func TestServiceRunSubAgentTaskInputValidation(t *testing.T) { @@ -151,4 +216,18 @@ func TestServiceRunSubAgentTaskInputValidation(t *testing.T) { }); err == nil { t.Fatalf("expected invalid role error") } + + ctx, cancel := context.WithTimeout(context.Background(), time.Nanosecond) + defer cancel() + time.Sleep(2 * time.Millisecond) + if _, err := service.RunSubAgentTask(ctx, SubAgentTaskInput{ + RunID: "sub-run-timeout", + Role: subagent.RoleCoder, + Task: subagent.Task{ + ID: "task-timeout", + Goal: "goal", + }, + }); err == nil { + t.Fatalf("expected context error") + } } diff --git a/internal/subagent/engine_test.go b/internal/subagent/engine_test.go new file mode 100644 index 00000000..feeb449c --- /dev/null +++ b/internal/subagent/engine_test.go @@ -0,0 +1,59 @@ +package subagent + +import ( + "context" + "testing" +) + +func TestDefaultEngineRunStep(t *testing.T) { + t.Parallel() + + engine := defaultEngine{} + + t.Run("uses expected output as summary", func(t *testing.T) { + t.Parallel() + + out, err := engine.RunStep(context.Background(), StepInput{ + Task: Task{ + Goal: "goal", + ExpectedOutput: "expected", + }, + }) + if err != nil { + t.Fatalf("RunStep() error = %v", err) + } + if !out.Done { + t.Fatalf("expected done output") + } + if out.Output.Summary != "expected" { + t.Fatalf("summary = %q, want %q", out.Output.Summary, "expected") + } + }) + + t.Run("falls back to goal", func(t *testing.T) { + t.Parallel() + + out, err := engine.RunStep(context.Background(), StepInput{ + Task: Task{ + Goal: "goal-value", + ExpectedOutput: " ", + }, + }) + if err != nil { + t.Fatalf("RunStep() error = %v", err) + } + if out.Output.Summary != "goal-value" { + t.Fatalf("summary = %q, want %q", out.Output.Summary, "goal-value") + } + }) + + t.Run("context canceled", func(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := engine.RunStep(ctx, StepInput{Task: Task{Goal: "g"}}); err == nil { + t.Fatalf("expected context error") + } + }) +} diff --git a/internal/subagent/output_contract.go b/internal/subagent/output_contract.go index 4fe063ee..5b7514ea 100644 --- a/internal/subagent/output_contract.go +++ b/internal/subagent/output_contract.go @@ -14,14 +14,48 @@ var supportedOutputSections = map[string]struct{}{ // validateOutputContract 校验输出结构是否满足角色策略要求。 func validateOutputContract(policy RolePolicy, output Output) error { out := output.normalize() - for _, section := range dedupeAndTrim(policy.RequiredSections) { + requiredSections, err := normalizeRequiredSections(policy.RequiredSections) + if err != nil { + return err + } + for _, key := range requiredSections { + if !hasOutputSectionContent(out, key) { + return errorsf("output section %q is required", key) + } + } + return nil +} + +// normalizeRequiredSections 归一化并校验 required section 名称集合。 +func normalizeRequiredSections(sections []string) ([]string, error) { + items := dedupeAndTrim(sections) + keys := make([]string, 0, len(items)) + for _, section := range items { key := strings.ToLower(strings.TrimSpace(section)) if _, ok := supportedOutputSections[key]; !ok { - return errorsf("unsupported required output section %q", section) - } - if key == "summary" && strings.TrimSpace(out.Summary) == "" { - return errorsf("output summary is required") + return nil, errorsf("unsupported required output section %q", section) } + keys = append(keys, key) + } + return keys, nil +} + +// hasOutputSectionContent 判断指定 section 在输出中是否包含有效内容。 +func hasOutputSectionContent(output Output, section string) bool { + switch section { + case "summary": + return strings.TrimSpace(output.Summary) != "" + case "findings": + return len(output.Findings) > 0 + case "patches": + return len(output.Patches) > 0 + case "risks": + return len(output.Risks) > 0 + case "next_actions": + return len(output.NextActions) > 0 + case "artifacts": + return len(output.Artifacts) > 0 + default: + return false } - return nil } diff --git a/internal/subagent/output_contract_test.go b/internal/subagent/output_contract_test.go new file mode 100644 index 00000000..3971c2d0 --- /dev/null +++ b/internal/subagent/output_contract_test.go @@ -0,0 +1,74 @@ +package subagent + +import "testing" + +func TestValidateOutputContractRequiresAllDeclaredSections(t *testing.T) { + t.Parallel() + + policy := RolePolicy{ + Role: RoleCoder, + SystemPrompt: "prompt", + AllowedTools: []string{"bash"}, + RequiredSections: []string{"summary", "findings", "patches", "risks", "next_actions", "artifacts"}, + } + + full := Output{ + Summary: "ok", + Findings: []string{"f"}, + Patches: []string{"p"}, + Risks: []string{"r"}, + NextActions: []string{"n"}, + Artifacts: []string{"a"}, + } + if err := validateOutputContract(policy, full); err != nil { + t.Fatalf("validateOutputContract() error = %v", err) + } + + cases := []struct { + name string + output Output + }{ + {name: "missing summary", output: Output{ + Findings: []string{"f"}, Patches: []string{"p"}, Risks: []string{"r"}, NextActions: []string{"n"}, Artifacts: []string{"a"}, + }}, + {name: "missing findings", output: Output{ + Summary: "ok", Patches: []string{"p"}, Risks: []string{"r"}, NextActions: []string{"n"}, Artifacts: []string{"a"}, + }}, + {name: "missing patches", output: Output{ + Summary: "ok", Findings: []string{"f"}, Risks: []string{"r"}, NextActions: []string{"n"}, Artifacts: []string{"a"}, + }}, + {name: "missing risks", output: Output{ + Summary: "ok", Findings: []string{"f"}, Patches: []string{"p"}, NextActions: []string{"n"}, Artifacts: []string{"a"}, + }}, + {name: "missing next actions", output: Output{ + Summary: "ok", Findings: []string{"f"}, Patches: []string{"p"}, Risks: []string{"r"}, Artifacts: []string{"a"}, + }}, + {name: "missing artifacts", output: Output{ + Summary: "ok", Findings: []string{"f"}, Patches: []string{"p"}, Risks: []string{"r"}, NextActions: []string{"n"}, + }}, + } + + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if err := validateOutputContract(policy, tc.output); err == nil { + t.Fatalf("expected contract validation error") + } + }) + } +} + +func TestValidateOutputContractUnsupportedSection(t *testing.T) { + t.Parallel() + + policy := RolePolicy{ + Role: RoleResearcher, + SystemPrompt: "prompt", + AllowedTools: []string{"filesystem_grep"}, + RequiredSections: []string{"summary", "x"}, + } + if err := validateOutputContract(policy, Output{Summary: "ok"}); err == nil { + t.Fatalf("expected unsupported section error") + } +} diff --git a/internal/subagent/policy.go b/internal/subagent/policy.go index 7a5b3d62..12c81544 100644 --- a/internal/subagent/policy.go +++ b/internal/subagent/policy.go @@ -33,7 +33,7 @@ func (p RolePolicy) Validate() error { if len(dedupeAndTrim(p.RequiredSections)) == 0 { return errorsf("role policy required sections is empty") } - if err := validateOutputContract(p, Output{Summary: "probe"}); err != nil { + if _, err := normalizeRequiredSections(p.RequiredSections); err != nil { return err } return nil diff --git a/internal/subagent/worker.go b/internal/subagent/worker.go index 995f4501..6e6c791a 100644 --- a/internal/subagent/worker.go +++ b/internal/subagent/worker.go @@ -27,6 +27,8 @@ type worker struct { lastErr string } +const traceWindowSize = 16 + // NewWorker 根据角色、策略与引擎创建一个 WorkerRuntime 实例。 func NewWorker(role Role, policy RolePolicy, engine Engine) (WorkerRuntime, error) { if !role.Valid() { @@ -70,7 +72,11 @@ func (w *worker) Start(task Task, budget Budget, capability Capability) error { w.task = task w.budget = budget.normalize(w.policy.DefaultBudget) - w.capability = capability.normalize() + effectiveCapability, err := bindCapabilityToPolicy(capability.normalize(), w.policy) + if err != nil { + return err + } + w.capability = effectiveCapability w.trace = nil w.stepCount = 0 w.output = Output{} @@ -115,13 +121,18 @@ func (w *worker) Step(ctx context.Context) (StepResult, error) { Budget: w.budget, Capability: w.capability, StepIndex: w.stepCount + 1, - Trace: append([]string(nil), w.trace...), + Trace: cloneRecentTrace(w.trace, traceWindowSize), } w.mu.Unlock() stepOutput, err := w.engine.RunStep(ctx, input) if err != nil { w.mu.Lock() + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + result := w.finishLocked(StateCanceled, StopReasonCanceled, Output{}, nil) + w.mu.Unlock() + return StepResult{State: result.State, Done: true, Step: result.StepCount}, err + } result := w.finishLocked(StateFailed, StopReasonError, Output{}, err) w.mu.Unlock() return StepResult{State: result.State, Done: true, Step: result.StepCount}, err @@ -161,6 +172,48 @@ func (w *worker) Step(ctx context.Context) (StepResult, error) { }, nil } +// bindCapabilityToPolicy 将 capability 约束在角色策略允许的工具集合内。 +func bindCapabilityToPolicy(capability Capability, policy RolePolicy) (Capability, error) { + allowedPolicyTools := dedupeAndTrim(policy.AllowedTools) + allowedSet := make(map[string]struct{}, len(allowedPolicyTools)) + for _, tool := range allowedPolicyTools { + allowedSet[strings.ToLower(strings.TrimSpace(tool))] = struct{}{} + } + + if len(capability.AllowedTools) == 0 { + capability.AllowedTools = append([]string(nil), allowedPolicyTools...) + return capability, nil + } + + effective := make([]string, 0, len(capability.AllowedTools)) + disallowed := make([]string, 0) + for _, tool := range capability.AllowedTools { + normalized := strings.ToLower(strings.TrimSpace(tool)) + if _, ok := allowedSet[normalized]; !ok { + disallowed = append(disallowed, tool) + continue + } + effective = append(effective, tool) + } + if len(disallowed) > 0 { + return Capability{}, errorsf("capability contains disallowed tools: %s", strings.Join(disallowed, ", ")) + } + capability.AllowedTools = effective + return capability, nil +} + +// cloneRecentTrace 复制最近 limit 条 trace，避免每步复制完整历史导致复杂度放大。 +func cloneRecentTrace(trace []string, limit int) []string { + if len(trace) == 0 { + return nil + } + if limit <= 0 || len(trace) <= limit { + return append([]string(nil), trace...) + } + start := len(trace) - limit + return append([]string(nil), trace[start:]...) +} + // Stop 主动终止运行中的 worker，并按终止原因映射最终状态。 func (w *worker) Stop(reason StopReason) error { if w == nil { diff --git a/internal/subagent/worker_test.go b/internal/subagent/worker_test.go index ef4e42ce..b323ad98 100644 --- a/internal/subagent/worker_test.go +++ b/internal/subagent/worker_test.go @@ -21,8 +21,11 @@ func TestWorkerLifecycleCompleted(t *testing.T) { Done: true, Output: Output{ Summary: "done", + Findings: []string{"root cause fixed"}, Patches: []string{"a.go"}, + Risks: []string{"need integration verify"}, NextActions: []string{"run tests"}, + Artifacts: []string{"test report"}, }, }, nil })) @@ -193,7 +196,17 @@ func TestWorkerFactoryCreate(t *testing.T) { factory := NewWorkerFactory(func(role Role, policy RolePolicy) Engine { return EngineFunc(func(ctx context.Context, input StepInput) (StepOutput, error) { - return StepOutput{Done: true, Output: Output{Summary: "ok"}}, nil + return StepOutput{ + Done: true, + Output: Output{ + Summary: "ok", + Findings: []string{"f1"}, + Patches: []string{"p1"}, + Risks: []string{"r1"}, + NextActions: []string{"n1"}, + Artifacts: []string{"a1"}, + }, + }, nil }) }) @@ -243,3 +256,121 @@ func TestWorkerRejectsInvalidOutputContract(t *testing.T) { t.Fatalf("unexpected result: %+v", result) } } + +func TestWorkerStartCapabilityPolicyGuard(t *testing.T) { + t.Parallel() + + policy, err := DefaultRolePolicy(RoleReviewer) + if err != nil { + t.Fatalf("DefaultRolePolicy() error = %v", err) + } + + w, err := NewWorker(RoleReviewer, policy, nil) + if err != nil { + t.Fatalf("NewWorker() error = %v", err) + } + + if err := w.Start(Task{ID: "t-cap", Goal: "goal"}, Budget{}, Capability{ + AllowedTools: []string{"filesystem_read_file", "bash"}, + }); err == nil { + t.Fatalf("expected disallowed capability tool to fail") + } + + w2, err := NewWorker(RoleReviewer, policy, nil) + if err != nil { + t.Fatalf("NewWorker() error = %v", err) + } + if err := w2.Start(Task{ID: "t-cap-ok", Goal: "goal"}, Budget{}, Capability{}); err != nil { + t.Fatalf("Start() error = %v", err) + } + if err := w2.Stop(StopReasonCanceled); err != nil { + t.Fatalf("Stop() error = %v", err) + } + result, err := w2.Result() + if err != nil { + t.Fatalf("Result() error = %v", err) + } + if len(result.Capability.AllowedTools) != len(policy.AllowedTools) { + t.Fatalf("capability tools = %v, want policy tools %v", result.Capability.AllowedTools, policy.AllowedTools) + } +} + +func TestWorkerTraceWindow(t *testing.T) { + t.Parallel() + + policy, err := DefaultRolePolicy(RoleResearcher) + if err != nil { + t.Fatalf("DefaultRolePolicy() error = %v", err) + } + + var observedTraceLen int + w, err := NewWorker(RoleResearcher, policy, EngineFunc(func(ctx context.Context, input StepInput) (StepOutput, error) { + observedTraceLen = len(input.Trace) + return StepOutput{ + Done: true, + Output: Output{ + Summary: "done", + Findings: []string{"f1"}, + Patches: []string{"p1"}, + Risks: []string{"r1"}, + NextActions: []string{"n1"}, + Artifacts: []string{"a1"}, + }, + }, nil + })) + if err != nil { + t.Fatalf("NewWorker() error = %v", err) + } + impl, ok := w.(*worker) + if !ok { + t.Fatalf("expected *worker implementation") + } + impl.trace = make([]string, traceWindowSize+4) + for i := range impl.trace { + impl.trace[i] = "trace" + } + impl.state = StateRunning + impl.task = Task{ID: "trace", Goal: "goal"} + impl.budget = Budget{MaxSteps: 5, Timeout: time.Second} + impl.capability = Capability{} + impl.startedAt = time.Now() + + if _, err := w.Step(context.Background()); err != nil { + t.Fatalf("Step() error = %v", err) + } + if observedTraceLen != traceWindowSize { + t.Fatalf("trace len = %d, want %d", observedTraceLen, traceWindowSize) + } +} + +func TestWorkerNilAndValidationBranches(t *testing.T) { + t.Parallel() + + if _, err := NewWorker(Role("bad"), RolePolicy{}, nil); err == nil { + t.Fatalf("expected invalid role error") + } + if _, err := NewWorker(RoleCoder, RolePolicy{Role: RoleReviewer, SystemPrompt: "p", AllowedTools: []string{"bash"}, RequiredSections: []string{"summary"}}, nil); err == nil { + t.Fatalf("expected role mismatch error") + } + + var nilWorker *worker + if err := nilWorker.Start(Task{}, Budget{}, Capability{}); err == nil { + t.Fatalf("expected nil worker start error") + } + if _, err := nilWorker.Step(context.Background()); err == nil { + t.Fatalf("expected nil worker step error") + } + if err := nilWorker.Stop(StopReasonCanceled); err == nil { + t.Fatalf("expected nil worker stop error") + } + if _, err := nilWorker.Result(); err == nil { + t.Fatalf("expected nil worker result error") + } + if nilWorker.State() != StateIdle { + t.Fatalf("nil worker state = %q, want %q", nilWorker.State(), StateIdle) + } + nilPolicy := nilWorker.Policy() + if nilPolicy.Role != "" || nilPolicy.SystemPrompt != "" || len(nilPolicy.AllowedTools) != 0 || len(nilPolicy.RequiredSections) != 0 { + t.Fatalf("nil worker policy should be empty, got %+v", nilPolicy) + } +} From cc74bd41131e958db9d54c1cd0080c0a05e4e1b1 Mon Sep 17 00:00:00 2001 From: xgopilot Date: Tue, 14 Apr 2026 16:18:05 +0000 Subject: [PATCH 03/10] fix(runtime): resolve merge conflicts with main event/runtime changes Generated with [codeagent](https://github.com/qbox/codeagent) Co-authored-by: Cai-Tang-www <106404101+Cai-Tang-www@users.noreply.github.com> --- internal/runtime/events.go | 63 +++++++++++++++++++++++++++++++------ internal/runtime/runtime.go | 47 +++++++++++++++++---------- 2 files changed, 84 insertions(+), 26 deletions(-) diff --git a/internal/runtime/events.go b/internal/runtime/events.go index f9506c40..f8608473 100644 --- a/internal/runtime/events.go +++ b/internal/runtime/events.go @@ -1,6 +1,10 @@ package runtime -import "neo-code/internal/subagent" +import ( + "time" + + "neo-code/internal/subagent" +) // EventType identifies the kind of runtime event emitted during a run. type EventType string @@ -9,10 +13,41 @@ type EventType string // for a specific run. RunID is provided by the caller and is echoed back on all // events so upper layers can ignore stale events from older runs. type RuntimeEvent struct { - Type EventType - RunID string - SessionID string - Payload any + Type EventType + RunID string + SessionID string + Turn int + Phase string + Timestamp time.Time + PayloadVersion int + Payload any +} + +// PhaseChangedPayload 描述 phase 迁移。 +type PhaseChangedPayload struct { + From string `json:"from"` + To string `json:"to"` +} + +// BudgetCheckedPayload 为预算检查壳事件负载（1A 仅占位，1B阶段使用）。 +type BudgetCheckedPayload struct { + Note string `json:"note,omitempty"` +} + +// ProgressEvaluatedPayload 汇总 progress 控制面评估结果。 +type ProgressEvaluatedPayload struct { + Score string `json:"score"` +} + +// StopReasonDecidedPayload 承载唯一停止原因决议结果。 +type StopReasonDecidedPayload struct { + Reason string `json:"reason"` + Detail string `json:"detail,omitempty"` +} + +// LedgerReconciledPayload 为账本对账壳事件负载（1A 仅占位）。 +type LedgerReconciledPayload struct { + Note string `json:"note,omitempty"` } // PermissionRequestPayload 描述一次需要审批的权限请求上下文。 @@ -71,18 +106,28 @@ const ( // EventProviderRetry is emitted when runtime retries a provider call due to // a retryable error (e.g. 429, 5xx). Payload is a human-readable message. EventProviderRetry EventType = "provider_retry" - // EventPermissionRequest is emitted when a tool call hits an ask decision. - EventPermissionRequest EventType = "permission_request" + // EventPermissionRequested 是 1A 权限请求事件名。 + EventPermissionRequested EventType = "permission_requested" + // EventPermissionRequest 为兼容旧事件名保留，语义等同 EventPermissionRequested。 + EventPermissionRequest = EventPermissionRequested // EventPermissionResolved is emitted when runtime resolves a permission request or denial. EventPermissionResolved EventType = "permission_resolved" // EventCompactStart is emitted when a compact cycle starts. EventCompactStart EventType = "compact_start" - // EventCompactDone is emitted when a compact cycle completes. - EventCompactDone EventType = "compact_done" + // EventCompactApplied 表示一次 compact 已成功应用或校验完成（1A 主事件）。 + EventCompactApplied EventType = "compact_applied" + // EventCompactDone 为兼容旧事件名保留，语义等同 EventCompactApplied。 + EventCompactDone = EventCompactApplied // EventCompactError is emitted when compact fails. EventCompactError EventType = "compact_error" // EventTokenUsage is emitted after each provider response with token statistics. EventTokenUsage EventType = "token_usage" + // EventPhaseChanged 表示显式 phase 迁移。 + EventPhaseChanged EventType = "phase_changed" + // EventProgressEvaluated 表示 progress 评估结果。 + EventProgressEvaluated EventType = "progress_evaluated" + // EventStopReasonDecided 表示唯一停止原因已决议。 + EventStopReasonDecided EventType = "stop_reason_decided" ) // TokenUsagePayload carries token usage statistics for a single provider turn. diff --git a/internal/runtime/runtime.go b/internal/runtime/runtime.go index ec1aa87f..ad4e28a6 100644 --- a/internal/runtime/runtime.go +++ b/internal/runtime/runtime.go @@ -23,6 +23,10 @@ const ( providerRetryBaseWait = 1 * time.Second providerRetryMaxWait = 5 * time.Second defaultMaxLoops = 8 + defaultToolParallelism = 4 + noProgressStreakLimit = 3 + + terminationEventEmitTimeout = 500 * time.Millisecond ) // Runtime 定义 runtime 对外暴露的运行、压缩与审批接口。 @@ -68,13 +72,15 @@ type Service struct { memoExtractor MemoExtractor subAgentFactory subagent.Factory - events chan RuntimeEvent - sessionMu sync.Mutex - sessionLocks map[string]*sessionLockEntry - runMu sync.Mutex - activeRunToken uint64 - nextRunToken uint64 - activeRunCancels map[uint64]context.CancelFunc + events chan RuntimeEvent + sessionMu sync.Mutex + sessionLocks map[string]*sessionLockEntry + runMu sync.Mutex + activeRunToken uint64 + nextRunToken uint64 + activeRunCancels map[uint64]context.CancelFunc + permissionAskMapMu sync.Mutex + permissionAskLocks map[string]*permissionAskLockEntry } // sessionLockEntry 维护单个会话锁及其当前引用计数，用于在无引用时回收 map 项。 @@ -83,6 +89,12 @@ type sessionLockEntry struct { refs int } +// permissionAskLockEntry 维护单个运行的审批串行锁与引用计数。 +type permissionAskLockEntry struct { + mu sync.Mutex + refs int +} + // NewWithFactory 使用注入依赖构建默认 runtime Service。 func NewWithFactory( configManager *config.Manager, @@ -102,16 +114,17 @@ func NewWithFactory( } return &Service{ - configManager: configManager, - sessionStore: sessionStore, - toolManager: toolManager, - providerFactory: providerFactory, - contextBuilder: contextBuilder, - approvalBroker: approval.NewBroker(), - subAgentFactory: subagent.NewWorkerFactory(nil), - events: make(chan RuntimeEvent, 128), - sessionLocks: make(map[string]*sessionLockEntry), - activeRunCancels: make(map[uint64]context.CancelFunc), + configManager: configManager, + sessionStore: sessionStore, + toolManager: toolManager, + providerFactory: providerFactory, + contextBuilder: contextBuilder, + approvalBroker: approval.NewBroker(), + subAgentFactory: subagent.NewWorkerFactory(nil), + events: make(chan RuntimeEvent, 128), + sessionLocks: make(map[string]*sessionLockEntry), + activeRunCancels: make(map[uint64]context.CancelFunc), + permissionAskLocks: make(map[string]*permissionAskLockEntry), } } From 730a61b129c5ccb04d04b837699ce8e6b2d2fb66 Mon Sep 17 00:00:00 2001 From: xgopilot Date: Tue, 14 Apr 2026 16:21:23 +0000 Subject: [PATCH 04/10] fix(runtime): align runtime core with main to resolve merge conflicts Generated with [codeagent](https://github.com/qbox/codeagent) Co-authored-by: Cai-Tang-www <106404101+Cai-Tang-www@users.noreply.github.com> --- internal/runtime/controlplane/decider.go | 38 ++++++++ internal/runtime/controlplane/decider_test.go | 88 +++++++++++++++++++ internal/runtime/controlplane/envelope.go | 4 + internal/runtime/controlplane/phase.go | 13 +++ internal/runtime/controlplane/progress.go | 39 ++++++++ .../runtime/controlplane/progress_test.go | 42 +++++++++ internal/runtime/controlplane/stop_reason.go | 15 ++++ internal/runtime/events.go | 36 +------- internal/runtime/events_subagent.go | 33 +++++++ internal/runtime/runtime.go | 7 +- internal/runtime/runtime_limits.go | 4 + 11 files changed, 283 insertions(+), 36 deletions(-) create mode 100644 internal/runtime/controlplane/decider.go create mode 100644 internal/runtime/controlplane/decider_test.go create mode 100644 internal/runtime/controlplane/envelope.go create mode 100644 internal/runtime/controlplane/phase.go create mode 100644 internal/runtime/controlplane/progress.go create mode 100644 internal/runtime/controlplane/progress_test.go create mode 100644 internal/runtime/controlplane/stop_reason.go create mode 100644 internal/runtime/events_subagent.go create mode 100644 internal/runtime/runtime_limits.go diff --git a/internal/runtime/controlplane/decider.go b/internal/runtime/controlplane/decider.go new file mode 100644 index 00000000..641bc6cb --- /dev/null +++ b/internal/runtime/controlplane/decider.go @@ -0,0 +1,38 @@ +package controlplane + +import ( + "context" + "errors" + "strings" +) + +// StopInput 汇总停止决议所需的信号（可多信号并存，由 DecideStopReason 按优先级表决）。 +type StopInput struct { + ContextCanceled bool + MaxLoopsReached bool + RunError error + Success bool +} + +// DecideStopReason 按固定优先级返回唯一 StopReason：取消 > 达到轮数上限 > 错误 > 成功。 +func DecideStopReason(in StopInput) (StopReason, string) { + if in.ContextCanceled { + return StopReasonCanceled, "" + } + if in.MaxLoopsReached { + if in.RunError != nil { + return StopReasonMaxLoops, strings.TrimSpace(in.RunError.Error()) + } + return StopReasonMaxLoops, "runtime: max loop reached" + } + if in.RunError != nil { + if errors.Is(in.RunError, context.Canceled) { + return StopReasonCanceled, "" + } + return StopReasonError, strings.TrimSpace(in.RunError.Error()) + } + if in.Success { + return StopReasonSuccess, "" + } + return StopReasonError, "runtime: stop reason undetermined" +} diff --git a/internal/runtime/controlplane/decider_test.go b/internal/runtime/controlplane/decider_test.go new file mode 100644 index 00000000..1f41b8e4 --- /dev/null +++ b/internal/runtime/controlplane/decider_test.go @@ -0,0 +1,88 @@ +package controlplane + +import ( + "context" + "errors" + "testing" +) + +func TestDecideStopReasonPriority(t *testing.T) { + t.Parallel() + + errSample := errors.New("boom") + cases := []struct { + name string + in StopInput + reason StopReason + }{ + { + name: "canceled_wins_over_max_loops", + in: StopInput{ + ContextCanceled: true, + MaxLoopsReached: true, + RunError: errSample, + }, + reason: StopReasonCanceled, + }, + { + name: "max_loops_wins_over_error", + in: StopInput{ + MaxLoopsReached: true, + RunError: errSample, + }, + reason: StopReasonMaxLoops, + }, + { + name: "error_when_no_max_loop_flag", + in: StopInput{ + RunError: errSample, + }, + reason: StopReasonError, + }, + { + name: "success", + in: StopInput{ + Success: true, + }, + reason: StopReasonSuccess, + }, + { + name: "context_canceled_on_error_field", + in: StopInput{ + RunError: context.Canceled, + }, + reason: StopReasonCanceled, + }, + } + + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + got, _ := DecideStopReason(tc.in) + if got != tc.reason { + t.Fatalf("DecideStopReason() = %q, want %q", got, tc.reason) + } + }) + } +} + +func TestDecideStopReasonDetails(t *testing.T) { + t.Parallel() + + reason, detail := DecideStopReason(StopInput{MaxLoopsReached: true}) + if reason != StopReasonMaxLoops { + t.Fatalf("reason = %q, want %q", reason, StopReasonMaxLoops) + } + if detail != "runtime: max loop reached" { + t.Fatalf("detail = %q, want default max-loop detail", detail) + } + + reason, detail = DecideStopReason(StopInput{}) + if reason != StopReasonError { + t.Fatalf("reason = %q, want %q", reason, StopReasonError) + } + if detail != "runtime: stop reason undetermined" { + t.Fatalf("detail = %q, want undetermined detail", detail) + } +} diff --git a/internal/runtime/controlplane/envelope.go b/internal/runtime/controlplane/envelope.go new file mode 100644 index 00000000..ec2006fd --- /dev/null +++ b/internal/runtime/controlplane/envelope.go @@ -0,0 +1,4 @@ +package controlplane + +// PayloadVersion 为 runtime 事件 envelope 的当前协议版本号。 +const PayloadVersion = 1 diff --git a/internal/runtime/controlplane/phase.go b/internal/runtime/controlplane/phase.go new file mode 100644 index 00000000..e43b583d --- /dev/null +++ b/internal/runtime/controlplane/phase.go @@ -0,0 +1,13 @@ +package controlplane + +// Phase 表示单轮 ReAct 内的显式阶段（plan -> execute -> verify）。 +type Phase string + +const ( + // PhasePlan 规划阶段：构建上下文、调用 provider 直至得到 assistant 消息（含工具调用决策）。 + PhasePlan Phase = "plan" + // PhaseExecute 执行阶段：执行本批次全部工具调用。 + PhaseExecute Phase = "execute" + // PhaseVerify 验证阶段：工具结果已回灌，等待下一轮 provider 校验或收尾。 + PhaseVerify Phase = "verify" +) diff --git a/internal/runtime/controlplane/progress.go b/internal/runtime/controlplane/progress.go new file mode 100644 index 00000000..a1d26875 --- /dev/null +++ b/internal/runtime/controlplane/progress.go @@ -0,0 +1,39 @@ +package controlplane + +// ProgressEvidenceKind 标识工具/适配器产出的证据类型，runtime 仅聚合不做语义推断。 +type ProgressEvidenceKind string + +const ( + // EvidenceNewInfoNonDup 表示本轮引入了非重复的新信息（用于 streak 回归约束）。 + EvidenceNewInfoNonDup ProgressEvidenceKind = "EVIDENCE_NEW_INFO_NON_DUP" +) + +// ProgressEvidenceRecord 描述一条可计分的进展证据。 +type ProgressEvidenceRecord struct { + Kind ProgressEvidenceKind `json:"kind"` + Detail string `json:"detail,omitempty"` +} + +// ProgressScore 表示一次评估后的分值增量与 streak 快照。 +type ProgressScore struct { + ScoreDelta int `json:"score_delta"` + NoProgressStreak int `json:"no_progress_streak"` + RepeatCycleStreak int `json:"repeat_cycle_streak"` +} + +// ProgressState 汇总当前运行期 progress 控制面状态。 +type ProgressState struct { + LastScore ProgressScore `json:"last_score"` +} + +// ApplyProgressEvidence 根据证据更新分值与 streak。 +func ApplyProgressEvidence(state ProgressState, records []ProgressEvidenceRecord) ProgressState { + next := state.LastScore + if len(records) == 0 { + next.NoProgressStreak++ + } else { + next.NoProgressStreak = 0 + next.ScoreDelta++ + } + return ProgressState{LastScore: next} +} diff --git a/internal/runtime/controlplane/progress_test.go b/internal/runtime/controlplane/progress_test.go new file mode 100644 index 00000000..cd25006e --- /dev/null +++ b/internal/runtime/controlplane/progress_test.go @@ -0,0 +1,42 @@ +package controlplane + +import "testing" + +func TestApplyProgressEvidenceNoEvidenceIncrementsNoProgress(t *testing.T) { + t.Parallel() + state := ProgressState{} + next := ApplyProgressEvidence(state, nil) + if next.LastScore.NoProgressStreak != 1 { + t.Fatalf("expected no_progress_streak 1, got %d", next.LastScore.NoProgressStreak) + } +} + +func TestApplyProgressEvidenceOnlyNonDupResetsNoProgressStreak(t *testing.T) { + t.Parallel() + state := ProgressState{ + LastScore: ProgressScore{NoProgressStreak: 3}, + } + next := ApplyProgressEvidence(state, []ProgressEvidenceRecord{ + {Kind: EvidenceNewInfoNonDup}, + }) + if next.LastScore.NoProgressStreak != 0 { + t.Fatalf("expected streak reset to 0, got %d", next.LastScore.NoProgressStreak) + } + if next.LastScore.ScoreDelta != 1 { + t.Fatalf("expected score_delta 1, got %d", next.LastScore.ScoreDelta) + } +} + +func TestApplyProgressEvidenceMixedResetsNoProgress(t *testing.T) { + t.Parallel() + state := ProgressState{ + LastScore: ProgressScore{NoProgressStreak: 2}, + } + next := ApplyProgressEvidence(state, []ProgressEvidenceRecord{ + {Kind: EvidenceNewInfoNonDup}, + {Kind: ProgressEvidenceKind("other_evidence")}, + }) + if next.LastScore.NoProgressStreak != 0 { + t.Fatalf("expected streak reset, got %d", next.LastScore.NoProgressStreak) + } +} diff --git a/internal/runtime/controlplane/stop_reason.go b/internal/runtime/controlplane/stop_reason.go new file mode 100644 index 00000000..d08b6cf6 --- /dev/null +++ b/internal/runtime/controlplane/stop_reason.go @@ -0,0 +1,15 @@ +package controlplane + +// StopReason 表示一次 Run 的最终停止原因，互斥且由决议器唯一确定。 +type StopReason string + +const ( + // StopReasonSuccess 表示助手正常结束（无待执行工具调用）。 + StopReasonSuccess StopReason = "success" + // StopReasonMaxLoops 表示达到配置的最大推理轮数。 + StopReasonMaxLoops StopReason = "max_loops" + // StopReasonError 表示不可恢复的运行时或 provider 错误。 + StopReasonError StopReason = "error" + // StopReasonCanceled 表示运行上下文被取消（含用户中断）。 + StopReasonCanceled StopReason = "canceled" +) diff --git a/internal/runtime/events.go b/internal/runtime/events.go index f8608473..a79eca47 100644 --- a/internal/runtime/events.go +++ b/internal/runtime/events.go @@ -3,7 +3,7 @@ package runtime import ( "time" - "neo-code/internal/subagent" + "neo-code/internal/runtime/controlplane" ) // EventType identifies the kind of runtime event emitted during a run. @@ -36,13 +36,13 @@ type BudgetCheckedPayload struct { // ProgressEvaluatedPayload 汇总 progress 控制面评估结果。 type ProgressEvaluatedPayload struct { - Score string `json:"score"` + Score controlplane.ProgressScore `json:"score"` } // StopReasonDecidedPayload 承载唯一停止原因决议结果。 type StopReasonDecidedPayload struct { - Reason string `json:"reason"` - Detail string `json:"detail,omitempty"` + Reason controlplane.StopReason `json:"reason"` + Detail string `json:"detail,omitempty"` } // LedgerReconciledPayload 为账本对账壳事件负载（1A 仅占位）。 @@ -108,16 +108,12 @@ const ( EventProviderRetry EventType = "provider_retry" // EventPermissionRequested 是 1A 权限请求事件名。 EventPermissionRequested EventType = "permission_requested" - // EventPermissionRequest 为兼容旧事件名保留，语义等同 EventPermissionRequested。 - EventPermissionRequest = EventPermissionRequested // EventPermissionResolved is emitted when runtime resolves a permission request or denial. EventPermissionResolved EventType = "permission_resolved" // EventCompactStart is emitted when a compact cycle starts. EventCompactStart EventType = "compact_start" // EventCompactApplied 表示一次 compact 已成功应用或校验完成（1A 主事件）。 EventCompactApplied EventType = "compact_applied" - // EventCompactDone 为兼容旧事件名保留，语义等同 EventCompactApplied。 - EventCompactDone = EventCompactApplied // EventCompactError is emitted when compact fails. EventCompactError EventType = "compact_error" // EventTokenUsage is emitted after each provider response with token statistics. @@ -137,27 +133,3 @@ type TokenUsagePayload struct { SessionInputTokens int `json:"session_input_tokens"` SessionOutputTokens int `json:"session_output_tokens"` } - -// SubAgentEventPayload 描述子代理执行生命周期的事件载荷。 -type SubAgentEventPayload struct { - Role subagent.Role `json:"role"` - TaskID string `json:"task_id"` - State subagent.State `json:"state"` - StopReason subagent.StopReason `json:"stop_reason,omitempty"` - Step int `json:"step,omitempty"` - Delta string `json:"delta,omitempty"` - Error string `json:"error,omitempty"` -} - -const ( - // EventSubAgentStarted 在子代理任务启动后触发。 - EventSubAgentStarted EventType = "subagent_started" - // EventSubAgentProgress 在子代理执行每一步后触发。 - EventSubAgentProgress EventType = "subagent_progress" - // EventSubAgentCompleted 在子代理成功结束后触发。 - EventSubAgentCompleted EventType = "subagent_completed" - // EventSubAgentFailed 在子代理失败结束后触发。 - EventSubAgentFailed EventType = "subagent_failed" - // EventSubAgentCanceled 在子代理被取消后触发。 - EventSubAgentCanceled EventType = "subagent_canceled" -) diff --git a/internal/runtime/events_subagent.go b/internal/runtime/events_subagent.go new file mode 100644 index 00000000..094564ab --- /dev/null +++ b/internal/runtime/events_subagent.go @@ -0,0 +1,33 @@ +package runtime + +import "neo-code/internal/subagent" + +// EventPermissionRequest 为兼容旧事件名保留，语义等同 EventPermissionRequested。 +const EventPermissionRequest EventType = EventPermissionRequested + +// EventCompactDone 为兼容旧事件名保留，语义等同 EventCompactApplied。 +const EventCompactDone EventType = EventCompactApplied + +// SubAgentEventPayload 描述子代理执行生命周期的事件载荷。 +type SubAgentEventPayload struct { + Role subagent.Role `json:"role"` + TaskID string `json:"task_id"` + State subagent.State `json:"state"` + StopReason subagent.StopReason `json:"stop_reason,omitempty"` + Step int `json:"step,omitempty"` + Delta string `json:"delta,omitempty"` + Error string `json:"error,omitempty"` +} + +const ( + // EventSubAgentStarted 在子代理任务启动后触发。 + EventSubAgentStarted EventType = "subagent_started" + // EventSubAgentProgress 在子代理执行每一步后触发。 + EventSubAgentProgress EventType = "subagent_progress" + // EventSubAgentCompleted 在子代理成功结束后触发。 + EventSubAgentCompleted EventType = "subagent_completed" + // EventSubAgentFailed 在子代理失败结束后触发。 + EventSubAgentFailed EventType = "subagent_failed" + // EventSubAgentCanceled 在子代理被取消后触发。 + EventSubAgentCanceled EventType = "subagent_canceled" +) diff --git a/internal/runtime/runtime.go b/internal/runtime/runtime.go index ad4e28a6..f2bb84e3 100644 --- a/internal/runtime/runtime.go +++ b/internal/runtime/runtime.go @@ -22,7 +22,6 @@ const ( defaultProviderRetryMax = 2 providerRetryBaseWait = 1 * time.Second providerRetryMaxWait = 5 * time.Second - defaultMaxLoops = 8 defaultToolParallelism = 4 noProgressStreakLimit = 3 @@ -70,7 +69,6 @@ type Service struct { compactRunner contextcompact.Runner approvalBroker *approval.Broker memoExtractor MemoExtractor - subAgentFactory subagent.Factory events chan RuntimeEvent sessionMu sync.Mutex @@ -81,6 +79,7 @@ type Service struct { activeRunCancels map[uint64]context.CancelFunc permissionAskMapMu sync.Mutex permissionAskLocks map[string]*permissionAskLockEntry + subAgentFactory subagent.Factory } // sessionLockEntry 维护单个会话锁及其当前引用计数，用于在无引用时回收 map 项。 @@ -120,11 +119,11 @@ func NewWithFactory( providerFactory: providerFactory, contextBuilder: contextBuilder, approvalBroker: approval.NewBroker(), - subAgentFactory: subagent.NewWorkerFactory(nil), events: make(chan RuntimeEvent, 128), sessionLocks: make(map[string]*sessionLockEntry), - activeRunCancels: make(map[uint64]context.CancelFunc), permissionAskLocks: make(map[string]*permissionAskLockEntry), + activeRunCancels: make(map[uint64]context.CancelFunc), + subAgentFactory: subagent.NewWorkerFactory(nil), } } diff --git a/internal/runtime/runtime_limits.go b/internal/runtime/runtime_limits.go new file mode 100644 index 00000000..3cab9213 --- /dev/null +++ b/internal/runtime/runtime_limits.go @@ -0,0 +1,4 @@ +package runtime + +// defaultMaxLoops 定义兼容旧运行循环逻辑时的默认最大轮数。 +const defaultMaxLoops = 8 From a5bf778be7c058a2e9349bde6542caa7e8a12b53 Mon Sep 17 00:00:00 2001 From: xgopilot Date: Tue, 14 Apr 2026 16:22:48 +0000 Subject: [PATCH 05/10] refactor(runtime): decouple subagent factory storage from service struct Generated with [codeagent](https://github.com/qbox/codeagent) Co-authored-by: Cai-Tang-www <106404101+Cai-Tang-www@users.noreply.github.com> --- internal/runtime/runtime.go | 3 --- internal/runtime/subagent_factory.go | 33 ++++++++++++++++++----- internal/runtime/subagent_factory_test.go | 5 ---- 3 files changed, 27 insertions(+), 14 deletions(-) diff --git a/internal/runtime/runtime.go b/internal/runtime/runtime.go index f2bb84e3..dd69c342 100644 --- a/internal/runtime/runtime.go +++ b/internal/runtime/runtime.go @@ -14,7 +14,6 @@ import ( providertypes "neo-code/internal/provider/types" "neo-code/internal/runtime/approval" agentsession "neo-code/internal/session" - "neo-code/internal/subagent" "neo-code/internal/tools" ) @@ -79,7 +78,6 @@ type Service struct { activeRunCancels map[uint64]context.CancelFunc permissionAskMapMu sync.Mutex permissionAskLocks map[string]*permissionAskLockEntry - subAgentFactory subagent.Factory } // sessionLockEntry 维护单个会话锁及其当前引用计数，用于在无引用时回收 map 项。 @@ -123,7 +121,6 @@ func NewWithFactory( sessionLocks: make(map[string]*sessionLockEntry), permissionAskLocks: make(map[string]*permissionAskLockEntry), activeRunCancels: make(map[uint64]context.CancelFunc), - subAgentFactory: subagent.NewWorkerFactory(nil), } } diff --git a/internal/runtime/subagent_factory.go b/internal/runtime/subagent_factory.go index 140f4f70..bc55651c 100644 --- a/internal/runtime/subagent_factory.go +++ b/internal/runtime/subagent_factory.go @@ -1,20 +1,41 @@ package runtime -import "neo-code/internal/subagent" +import ( + "sync" + + "neo-code/internal/subagent" +) + +var serviceSubAgentFactory sync.Map + +// defaultSubAgentFactory 返回默认的子代理工厂实例。 +func defaultSubAgentFactory() subagent.Factory { + return subagent.NewWorkerFactory(nil) +} // SetSubAgentFactory 设置子代理运行时工厂；传入 nil 时回退到默认工厂。 func (s *Service) SetSubAgentFactory(factory subagent.Factory) { + if s == nil { + return + } if factory == nil { - s.subAgentFactory = subagent.NewWorkerFactory(nil) + serviceSubAgentFactory.Store(s, defaultSubAgentFactory()) return } - s.subAgentFactory = factory + serviceSubAgentFactory.Store(s, factory) } // SubAgentFactory 返回当前 runtime 持有的子代理运行时工厂。 func (s *Service) SubAgentFactory() subagent.Factory { - if s.subAgentFactory == nil { - s.subAgentFactory = subagent.NewWorkerFactory(nil) + if s == nil { + return defaultSubAgentFactory() + } + if factory, ok := serviceSubAgentFactory.Load(s); ok { + if typed, valid := factory.(subagent.Factory); valid && typed != nil { + return typed + } } - return s.subAgentFactory + defaultFactory := defaultSubAgentFactory() + serviceSubAgentFactory.Store(s, defaultFactory) + return defaultFactory } diff --git a/internal/runtime/subagent_factory_test.go b/internal/runtime/subagent_factory_test.go index 70256ce4..eea246b1 100644 --- a/internal/runtime/subagent_factory_test.go +++ b/internal/runtime/subagent_factory_test.go @@ -30,9 +30,4 @@ func TestServiceSubAgentFactoryRegistration(t *testing.T) { if svc.SubAgentFactory() == nil { t.Fatalf("expected reset to default sub-agent factory") } - - svc.subAgentFactory = nil - if svc.SubAgentFactory() == nil { - t.Fatalf("expected lazy init default sub-agent factory") - } } From e43f70c63369cddeadb439643c3250c526ff906a Mon Sep 17 00:00:00 2001 From: xgopilot Date: Wed, 15 Apr 2026 01:33:56 +0000 Subject: [PATCH 06/10] fix(subagent): resolve review issues for factory lifecycle and default output contract Generated with [codeagent](https://github.com/qbox/codeagent) Co-authored-by: Cai-Tang-www <106404101+Cai-Tang-www@users.noreply.github.com> --- internal/runtime/runtime.go | 4 +++ internal/runtime/subagent_factory.go | 34 +++++++++++++---------- internal/runtime/subagent_factory_test.go | 21 ++++++++++++++ internal/subagent/engine.go | 7 ++++- internal/subagent/engine_test.go | 30 ++++++++++++++++++++ 5 files changed, 81 insertions(+), 15 deletions(-) diff --git a/internal/runtime/runtime.go b/internal/runtime/runtime.go index dd69c342..928dd6c7 100644 --- a/internal/runtime/runtime.go +++ b/internal/runtime/runtime.go @@ -14,6 +14,7 @@ import ( providertypes "neo-code/internal/provider/types" "neo-code/internal/runtime/approval" agentsession "neo-code/internal/session" + "neo-code/internal/subagent" "neo-code/internal/tools" ) @@ -68,6 +69,8 @@ type Service struct { compactRunner contextcompact.Runner approvalBroker *approval.Broker memoExtractor MemoExtractor + subAgentFactory subagent.Factory + subAgentMu sync.RWMutex events chan RuntimeEvent sessionMu sync.Mutex @@ -117,6 +120,7 @@ func NewWithFactory( providerFactory: providerFactory, contextBuilder: contextBuilder, approvalBroker: approval.NewBroker(), + subAgentFactory: defaultSubAgentFactory(), events: make(chan RuntimeEvent, 128), sessionLocks: make(map[string]*sessionLockEntry), permissionAskLocks: make(map[string]*permissionAskLockEntry), diff --git a/internal/runtime/subagent_factory.go b/internal/runtime/subagent_factory.go index bc55651c..e1889275 100644 --- a/internal/runtime/subagent_factory.go +++ b/internal/runtime/subagent_factory.go @@ -1,12 +1,6 @@ package runtime -import ( - "sync" - - "neo-code/internal/subagent" -) - -var serviceSubAgentFactory sync.Map +import "neo-code/internal/subagent" // defaultSubAgentFactory 返回默认的子代理工厂实例。 func defaultSubAgentFactory() subagent.Factory { @@ -18,11 +12,13 @@ func (s *Service) SetSubAgentFactory(factory subagent.Factory) { if s == nil { return } + s.subAgentMu.Lock() + defer s.subAgentMu.Unlock() if factory == nil { - serviceSubAgentFactory.Store(s, defaultSubAgentFactory()) + s.subAgentFactory = defaultSubAgentFactory() return } - serviceSubAgentFactory.Store(s, factory) + s.subAgentFactory = factory } // SubAgentFactory 返回当前 runtime 持有的子代理运行时工厂。 @@ -30,12 +26,22 @@ func (s *Service) SubAgentFactory() subagent.Factory { if s == nil { return defaultSubAgentFactory() } - if factory, ok := serviceSubAgentFactory.Load(s); ok { - if typed, valid := factory.(subagent.Factory); valid && typed != nil { - return typed - } + s.subAgentMu.RLock() + factory := s.subAgentFactory + s.subAgentMu.RUnlock() + if factory != nil { + return factory } + defaultFactory := defaultSubAgentFactory() - serviceSubAgentFactory.Store(s, defaultFactory) + s.subAgentMu.Lock() + if s.subAgentFactory == nil { + s.subAgentFactory = defaultFactory + } + factory = s.subAgentFactory + s.subAgentMu.Unlock() + if factory != nil { + return factory + } return defaultFactory } diff --git a/internal/runtime/subagent_factory_test.go b/internal/runtime/subagent_factory_test.go index eea246b1..9869ba45 100644 --- a/internal/runtime/subagent_factory_test.go +++ b/internal/runtime/subagent_factory_test.go @@ -31,3 +31,24 @@ func TestServiceSubAgentFactoryRegistration(t *testing.T) { t.Fatalf("expected reset to default sub-agent factory") } } + +func TestServiceSubAgentFactoryIsolationAcrossInstances(t *testing.T) { + t.Parallel() + + svcA := NewWithFactory(nil, nil, nil, nil, nil) + svcB := NewWithFactory(nil, nil, nil, nil, nil) + + custom := fakeSubAgentFactory{} + svcA.SetSubAgentFactory(custom) + + if svcA.SubAgentFactory() == nil { + t.Fatalf("expected service A factory to be set") + } + if svcB.SubAgentFactory() == nil { + t.Fatalf("expected service B default factory") + } + + if svcA.SubAgentFactory() == svcB.SubAgentFactory() { + t.Fatalf("expected per-service factory isolation") + } +} diff --git a/internal/subagent/engine.go b/internal/subagent/engine.go index 6d849558..b8e859d8 100644 --- a/internal/subagent/engine.go +++ b/internal/subagent/engine.go @@ -23,7 +23,12 @@ func (defaultEngine) RunStep(ctx context.Context, input StepInput) (StepOutput, Delta: "default engine completed", Done: true, Output: Output{ - Summary: summary, + Summary: summary, + Findings: []string{"default-engine: no extra findings"}, + Patches: []string{"default-engine: no code changes"}, + Risks: []string{"default-engine: output generated without external verification"}, + NextActions: []string{"replace default engine with role-specific implementation if deeper execution is required"}, + Artifacts: []string{"default-engine"}, }, }, nil } diff --git a/internal/subagent/engine_test.go b/internal/subagent/engine_test.go index feeb449c..9a11048f 100644 --- a/internal/subagent/engine_test.go +++ b/internal/subagent/engine_test.go @@ -28,6 +28,12 @@ func TestDefaultEngineRunStep(t *testing.T) { if out.Output.Summary != "expected" { t.Fatalf("summary = %q, want %q", out.Output.Summary, "expected") } + if len(out.Output.Findings) == 0 || len(out.Output.Patches) == 0 || len(out.Output.Risks) == 0 { + t.Fatalf("default engine should populate required list sections, got %+v", out.Output) + } + if len(out.Output.NextActions) == 0 || len(out.Output.Artifacts) == 0 { + t.Fatalf("default engine should populate required sections, got %+v", out.Output) + } }) t.Run("falls back to goal", func(t *testing.T) { @@ -56,4 +62,28 @@ func TestDefaultEngineRunStep(t *testing.T) { t.Fatalf("expected context error") } }) + + t.Run("satisfies default role contract", func(t *testing.T) { + t.Parallel() + + out, err := engine.RunStep(context.Background(), StepInput{ + Task: Task{ + Goal: "goal", + ExpectedOutput: "summary", + }, + }) + if err != nil { + t.Fatalf("RunStep() error = %v", err) + } + + for _, role := range []Role{RoleResearcher, RoleCoder, RoleReviewer} { + policy, err := DefaultRolePolicy(role) + if err != nil { + t.Fatalf("DefaultRolePolicy(%q) error = %v", role, err) + } + if err := validateOutputContract(policy, out.Output); err != nil { + t.Fatalf("validateOutputContract(%q) error = %v", role, err) + } + } + }) } From 062846d8db662bb3fbb2bf2c696997650d16ef8f Mon Sep 17 00:00:00 2001 From: xgopilot Date: Wed, 15 Apr 2026 02:41:22 +0000 Subject: [PATCH 07/10] fix(runtime): resolve main merge conflicts for subagent runtime Generated with [codeagent](https://github.com/qbox/codeagent) Co-authored-by: Cai-Tang-www <106404101+Cai-Tang-www@users.noreply.github.com> --- internal/runtime/event_emitter.go | 71 ++++++++++++ internal/runtime/runtime.go | 154 -------------------------- internal/runtime/session_scheduler.go | 138 +++++++++++++++++++++++ internal/runtime/state.go | 3 + internal/runtime/subagent_factory.go | 86 +++++++++++--- 5 files changed, 280 insertions(+), 172 deletions(-) create mode 100644 internal/runtime/event_emitter.go create mode 100644 internal/runtime/session_scheduler.go diff --git a/internal/runtime/event_emitter.go b/internal/runtime/event_emitter.go new file mode 100644 index 00000000..43080dbb --- /dev/null +++ b/internal/runtime/event_emitter.go @@ -0,0 +1,71 @@ +package runtime + +import ( + "context" + "time" + + "neo-code/internal/runtime/controlplane" +) + +const turnUnspecified = -1 + +// emit 将 runtime 事件投递到事件通道，并在通道阻塞且上下文取消时返回错误。 +func (s *Service) emit(ctx context.Context, kind EventType, runID string, sessionID string, payload any) error { + return s.emitWithEnvelope(ctx, RuntimeEvent{ + Type: kind, + RunID: runID, + SessionID: sessionID, + Turn: turnUnspecified, + Timestamp: time.Now(), + PayloadVersion: controlplane.PayloadVersion, + Payload: payload, + }) +} + +// emitRunScoped 携带当前 run 的 turn/phase 元数据发出事件。 +func (s *Service) emitRunScoped(ctx context.Context, kind EventType, state *runState, payload any) error { + if state == nil { + return s.emit(ctx, kind, "", "", payload) + } + phase := "" + if state.phase != "" { + phase = string(state.phase) + } + return s.emitWithEnvelope(ctx, RuntimeEvent{ + Type: kind, + RunID: state.runID, + SessionID: state.session.ID, + Turn: state.turn, + Phase: phase, + Timestamp: time.Now(), + PayloadVersion: controlplane.PayloadVersion, + Payload: payload, + }) +} + +func (s *Service) emitWithEnvelope(ctx context.Context, evt RuntimeEvent) error { + if evt.PayloadVersion == 0 { + evt.PayloadVersion = controlplane.PayloadVersion + } + if evt.Timestamp.IsZero() { + evt.Timestamp = time.Now() + } + if err := s.deliverEvent(ctx, evt); err != nil { + return err + } + return nil +} + +func (s *Service) deliverEvent(ctx context.Context, evt RuntimeEvent) error { + select { + case s.events <- evt: + return nil + default: + } + select { + case s.events <- evt: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} diff --git a/internal/runtime/runtime.go b/internal/runtime/runtime.go index 928dd6c7..1a345f5a 100644 --- a/internal/runtime/runtime.go +++ b/internal/runtime/runtime.go @@ -2,8 +2,6 @@ package runtime import ( "context" - "path/filepath" - "strings" "sync" "time" @@ -14,7 +12,6 @@ import ( providertypes "neo-code/internal/provider/types" "neo-code/internal/runtime/approval" agentsession "neo-code/internal/session" - "neo-code/internal/subagent" "neo-code/internal/tools" ) @@ -69,8 +66,6 @@ type Service struct { compactRunner contextcompact.Runner approvalBroker *approval.Broker memoExtractor MemoExtractor - subAgentFactory subagent.Factory - subAgentMu sync.RWMutex events chan RuntimeEvent sessionMu sync.Mutex @@ -120,7 +115,6 @@ func NewWithFactory( providerFactory: providerFactory, contextBuilder: contextBuilder, approvalBroker: approval.NewBroker(), - subAgentFactory: defaultSubAgentFactory(), events: make(chan RuntimeEvent, 128), sessionLocks: make(map[string]*sessionLockEntry), permissionAskLocks: make(map[string]*permissionAskLockEntry), @@ -168,151 +162,3 @@ func (s *Service) LoadSession(ctx context.Context, id string) (agentsession.Sess } return session, nil } - -// loadOrCreateSession 负责在运行开始时解析工作目录并加载或创建会话。 -func (s *Service) loadOrCreateSession( - ctx context.Context, - sessionID string, - title string, - defaultWorkdir string, - requestedWorkdir string, -) (agentsession.Session, error) { - if strings.TrimSpace(sessionID) == "" { - sessionWorkdir, err := resolveWorkdirForSession(defaultWorkdir, "", requestedWorkdir) - if err != nil { - return agentsession.Session{}, err - } - session := agentsession.NewWithWorkdir(title, sessionWorkdir) - if err := s.sessionStore.Save(ctx, &session); err != nil { - return agentsession.Session{}, err - } - return session, nil - } - - session, err := s.sessionStore.Load(ctx, sessionID) - if err != nil { - return agentsession.Session{}, err - } - if strings.TrimSpace(requestedWorkdir) == "" && strings.TrimSpace(session.Workdir) != "" { - return session, nil - } - - resolved, err := resolveWorkdirForSession(defaultWorkdir, session.Workdir, requestedWorkdir) - if err != nil { - return agentsession.Session{}, err - } - if session.Workdir == resolved { - return session, nil - } - - session.Workdir = resolved - session.UpdatedAt = time.Now() - if err := s.sessionStore.Save(ctx, &session); err != nil { - return agentsession.Session{}, err - } - return session, nil -} - -// emit 将 runtime 事件投递到事件通道，并在通道阻塞且上下文取消时返回错误。 -func (s *Service) emit(ctx context.Context, kind EventType, runID string, sessionID string, payload any) error { - evt := RuntimeEvent{ - Type: kind, - RunID: runID, - SessionID: sessionID, - Payload: payload, - } - select { - case s.events <- evt: - return nil - default: - } - select { - case s.events <- evt: - return nil - case <-ctx.Done(): - return ctx.Err() - } -} - -// startRun 记录当前激活的运行取消句柄，并分配一个新的运行令牌。 -func (s *Service) startRun(cancel context.CancelFunc) uint64 { - s.runMu.Lock() - defer s.runMu.Unlock() - if s.activeRunCancels == nil { - s.activeRunCancels = make(map[uint64]context.CancelFunc) - } - - s.nextRunToken++ - token := s.nextRunToken - s.activeRunToken = token - s.activeRunCancels[token] = cancel - return token -} - -// finishRun 在运行结束时释放指定运行的取消句柄，并回退到最新活跃运行。 -func (s *Service) finishRun(token uint64) { - s.runMu.Lock() - defer s.runMu.Unlock() - - delete(s.activeRunCancels, token) - if s.activeRunToken != token { - return - } - - s.activeRunToken = 0 - for activeToken := range s.activeRunCancels { - if activeToken > s.activeRunToken { - s.activeRunToken = activeToken - } - } -} - -// acquireSessionLock 获取指定会话锁并返回释放引用的函数。 -// 调用方在完成会话级串行操作后，必须调用 release 以允许锁条目回收。 -func (s *Service) acquireSessionLock(sessionID string) (*sync.Mutex, func()) { - s.sessionMu.Lock() - if s.sessionLocks == nil { - s.sessionLocks = make(map[string]*sessionLockEntry) - } - - entry, ok := s.sessionLocks[sessionID] - if !ok { - entry = &sessionLockEntry{} - s.sessionLocks[sessionID] = entry - } - entry.refs++ - s.sessionMu.Unlock() - - released := false - release := func() { - s.sessionMu.Lock() - defer s.sessionMu.Unlock() - if released { - return - } - released = true - - current, exists := s.sessionLocks[sessionID] - if !exists || current != entry { - return - } - current.refs-- - if current.refs <= 0 { - delete(s.sessionLocks, sessionID) - } - } - return &entry.mu, release -} - -func resolveWorkdirForSession(defaultWorkdir string, currentWorkdir string, requestedWorkdir string) (string, error) { - base := agentsession.EffectiveWorkdir(currentWorkdir, defaultWorkdir) - if strings.TrimSpace(requestedWorkdir) == "" { - return agentsession.ResolveExistingDir(base) - } - - target := strings.TrimSpace(requestedWorkdir) - if !filepath.IsAbs(target) { - target = filepath.Join(base, target) - } - return agentsession.ResolveExistingDir(target) -} diff --git a/internal/runtime/session_scheduler.go b/internal/runtime/session_scheduler.go new file mode 100644 index 00000000..ed783622 --- /dev/null +++ b/internal/runtime/session_scheduler.go @@ -0,0 +1,138 @@ +package runtime + +import ( + "context" + "path/filepath" + "strings" + "sync" + "time" + + agentsession "neo-code/internal/session" +) + +// loadOrCreateSession 负责在运行开始时解析工作目录并加载或创建会话。 +func (s *Service) loadOrCreateSession( + ctx context.Context, + sessionID string, + title string, + defaultWorkdir string, + requestedWorkdir string, +) (agentsession.Session, error) { + if strings.TrimSpace(sessionID) == "" { + sessionWorkdir, err := resolveWorkdirForSession(defaultWorkdir, "", requestedWorkdir) + if err != nil { + return agentsession.Session{}, err + } + session := agentsession.NewWithWorkdir(title, sessionWorkdir) + if err := s.sessionStore.Save(ctx, &session); err != nil { + return agentsession.Session{}, err + } + return session, nil + } + + session, err := s.sessionStore.Load(ctx, sessionID) + if err != nil { + return agentsession.Session{}, err + } + if strings.TrimSpace(requestedWorkdir) == "" && strings.TrimSpace(session.Workdir) != "" { + return session, nil + } + + resolved, err := resolveWorkdirForSession(defaultWorkdir, session.Workdir, requestedWorkdir) + if err != nil { + return agentsession.Session{}, err + } + if session.Workdir == resolved { + return session, nil + } + + session.Workdir = resolved + session.UpdatedAt = time.Now() + if err := s.sessionStore.Save(ctx, &session); err != nil { + return agentsession.Session{}, err + } + return session, nil +} + +// startRun 记录当前激活的运行取消句柄，并分配一个新的运行令牌。 +func (s *Service) startRun(cancel context.CancelFunc) uint64 { + s.runMu.Lock() + defer s.runMu.Unlock() + if s.activeRunCancels == nil { + s.activeRunCancels = make(map[uint64]context.CancelFunc) + } + + s.nextRunToken++ + token := s.nextRunToken + s.activeRunToken = token + s.activeRunCancels[token] = cancel + return token +} + +// finishRun 在运行结束时释放指定运行的取消句柄，并回退到最新活跃运行。 +func (s *Service) finishRun(token uint64) { + s.runMu.Lock() + defer s.runMu.Unlock() + + delete(s.activeRunCancels, token) + if s.activeRunToken != token { + return + } + + s.activeRunToken = 0 + for activeToken := range s.activeRunCancels { + if activeToken > s.activeRunToken { + s.activeRunToken = activeToken + } + } +} + +// acquireSessionLock 获取指定会话锁并返回释放引用的函数。 +// 调用方在完成会话级串行操作后，必须调用 release 以允许锁条目回收。 +func (s *Service) acquireSessionLock(sessionID string) (*sync.Mutex, func()) { + s.sessionMu.Lock() + if s.sessionLocks == nil { + s.sessionLocks = make(map[string]*sessionLockEntry) + } + + entry, ok := s.sessionLocks[sessionID] + if !ok { + entry = &sessionLockEntry{} + s.sessionLocks[sessionID] = entry + } + entry.refs++ + s.sessionMu.Unlock() + + released := false + release := func() { + s.sessionMu.Lock() + defer s.sessionMu.Unlock() + if released { + return + } + released = true + + current, exists := s.sessionLocks[sessionID] + if !exists || current != entry { + return + } + current.refs-- + if current.refs <= 0 { + delete(s.sessionLocks, sessionID) + } + } + return &entry.mu, release +} + +func resolveWorkdirForSession(defaultWorkdir string, currentWorkdir string, requestedWorkdir string) (string, error) { + base := agentsession.EffectiveWorkdir(currentWorkdir, defaultWorkdir) + if strings.TrimSpace(requestedWorkdir) == "" { + return agentsession.ResolveExistingDir(base) + } + + target := strings.TrimSpace(requestedWorkdir) + if !filepath.IsAbs(target) { + target = filepath.Join(base, target) + } + return agentsession.ResolveExistingDir(target) +} diff --git a/internal/runtime/state.go b/internal/runtime/state.go index 3b7d1c41..a5d24411 100644 --- a/internal/runtime/state.go +++ b/internal/runtime/state.go @@ -6,6 +6,7 @@ import ( "neo-code/internal/config" "neo-code/internal/provider" providertypes "neo-code/internal/provider/types" + "neo-code/internal/runtime/controlplane" agentsession "neo-code/internal/session" ) @@ -21,6 +22,8 @@ type runState struct { compactApplied bool reactiveCompactAttempts int rememberedThisRun bool + turn int + phase controlplane.Phase } // newRunState 基于持久化会话创建一次运行的内存状态镜像。 diff --git a/internal/runtime/subagent_factory.go b/internal/runtime/subagent_factory.go index e1889275..68662d0b 100644 --- a/internal/runtime/subagent_factory.go +++ b/internal/runtime/subagent_factory.go @@ -1,6 +1,69 @@ package runtime -import "neo-code/internal/subagent" +import ( + "runtime" + "sync" + + "neo-code/internal/subagent" +) + +type subAgentFactoryRegistry struct { + mu sync.RWMutex + factory map[*Service]subagent.Factory + tracked map[*Service]struct{} +} + +var globalSubAgentFactories = &subAgentFactoryRegistry{ + factory: make(map[*Service]subagent.Factory), + tracked: make(map[*Service]struct{}), +} + +// ensureTracked 为 Service 注册 GC 回调，避免全局注册表持有泄漏。 +func (r *subAgentFactoryRegistry) ensureTracked(s *Service) { + if s == nil { + return + } + + r.mu.Lock() + if _, ok := r.tracked[s]; ok { + r.mu.Unlock() + return + } + r.tracked[s] = struct{}{} + r.mu.Unlock() + + runtime.SetFinalizer(s, func(service *Service) { + globalSubAgentFactories.mu.Lock() + delete(globalSubAgentFactories.factory, service) + delete(globalSubAgentFactories.tracked, service) + globalSubAgentFactories.mu.Unlock() + }) +} + +// set 保存 Service 级工厂实例。 +func (r *subAgentFactoryRegistry) set(s *Service, f subagent.Factory) { + if s == nil { + return + } + r.ensureTracked(s) + + r.mu.Lock() + r.factory[s] = f + r.mu.Unlock() +} + +// get 读取 Service 级工厂实例。 +func (r *subAgentFactoryRegistry) get(s *Service) (subagent.Factory, bool) { + if s == nil { + return nil, false + } + r.ensureTracked(s) + + r.mu.RLock() + factory, ok := r.factory[s] + r.mu.RUnlock() + return factory, ok +} // defaultSubAgentFactory 返回默认的子代理工厂实例。 func defaultSubAgentFactory() subagent.Factory { @@ -12,13 +75,11 @@ func (s *Service) SetSubAgentFactory(factory subagent.Factory) { if s == nil { return } - s.subAgentMu.Lock() - defer s.subAgentMu.Unlock() if factory == nil { - s.subAgentFactory = defaultSubAgentFactory() + globalSubAgentFactories.set(s, defaultSubAgentFactory()) return } - s.subAgentFactory = factory + globalSubAgentFactories.set(s, factory) } // SubAgentFactory 返回当前 runtime 持有的子代理运行时工厂。 @@ -26,22 +87,11 @@ func (s *Service) SubAgentFactory() subagent.Factory { if s == nil { return defaultSubAgentFactory() } - s.subAgentMu.RLock() - factory := s.subAgentFactory - s.subAgentMu.RUnlock() - if factory != nil { + if factory, ok := globalSubAgentFactories.get(s); ok && factory != nil { return factory } defaultFactory := defaultSubAgentFactory() - s.subAgentMu.Lock() - if s.subAgentFactory == nil { - s.subAgentFactory = defaultFactory - } - factory = s.subAgentFactory - s.subAgentMu.Unlock() - if factory != nil { - return factory - } + globalSubAgentFactories.set(s, defaultFactory) return defaultFactory } From 631d8ab012eb0cf02a82cced6c8b1cff78a67320 Mon Sep 17 00:00:00 2001 From: xgopilot Date: Wed, 15 Apr 2026 02:44:23 +0000 Subject: [PATCH 08/10] refactor(runtime): align state model with main-compatible event envelope Generated with [codeagent](https://github.com/qbox/codeagent) Co-authored-by: Cai-Tang-www <106404101+Cai-Tang-www@users.noreply.github.com> --- internal/runtime/run.go | 8 ++--- .../runtime/runtime_internal_helpers_test.go | 13 +++----- internal/runtime/runtime_test.go | 28 ++++++++-------- internal/runtime/session_mutation.go | 1 - internal/runtime/state.go | 33 +++++++------------ 5 files changed, 33 insertions(+), 50 deletions(-) diff --git a/internal/runtime/run.go b/internal/runtime/run.go index b0caf20a..be606beb 100644 --- a/internal/runtime/run.go +++ b/internal/runtime/run.go @@ -139,8 +139,8 @@ func (s *Service) prepareTurnSnapshot(ctx context.Context, state *runState) (tur Shell: cfg.Shell, Provider: cfg.SelectedProvider, Model: cfg.CurrentModel, - SessionInputTokens: state.tokenInputTotal, - SessionOutputTokens: state.tokenOutputTotal, + SessionInputTokens: state.session.TokenInputTotal, + SessionOutputTokens: state.session.TokenOutputTotal, }, Compact: agentcontext.CompactOptions{ DisableMicroCompact: cfg.Context.Compact.MicroCompactDisabled, @@ -257,8 +257,8 @@ func (s *Service) emitTokenUsage(ctx context.Context, state *runState, result pr s.emit(ctx, EventTokenUsage, state.runID, state.session.ID, TokenUsagePayload{ InputTokens: result.inputTokens, OutputTokens: result.outputTokens, - SessionInputTokens: state.tokenInputTotal, - SessionOutputTokens: state.tokenOutputTotal, + SessionInputTokens: state.session.TokenInputTotal, + SessionOutputTokens: state.session.TokenOutputTotal, }) } diff --git a/internal/runtime/runtime_internal_helpers_test.go b/internal/runtime/runtime_internal_helpers_test.go index 3b5472f6..c1952b82 100644 --- a/internal/runtime/runtime_internal_helpers_test.go +++ b/internal/runtime/runtime_internal_helpers_test.go @@ -40,7 +40,6 @@ func TestRunStateNilReceiverNoops(t *testing.T) { t.Parallel() var state *runState - state.syncSessionTokenTotals() state.recordUsage(3, 5) state.resetTokenTotals() state.touchSession() @@ -53,18 +52,14 @@ func TestRunStateMutationsAndSync(t *testing.T) { state := newRunState("run-1", session) state.recordUsage(10, 20) - if state.tokenInputTotal != 11 || state.tokenOutputTotal != 22 { - t.Fatalf("unexpected token totals: in=%d out=%d", state.tokenInputTotal, state.tokenOutputTotal) - } - - state.syncSessionTokenTotals() if state.session.TokenInputTotal != 11 || state.session.TokenOutputTotal != 22 { t.Fatalf("session totals not synced: %+v", state.session) } state.resetTokenTotals() - if state.tokenInputTotal != 0 || state.tokenOutputTotal != 0 { - t.Fatalf("expected reset totals to be zero, got in=%d out=%d", state.tokenInputTotal, state.tokenOutputTotal) + if state.session.TokenInputTotal != 0 || state.session.TokenOutputTotal != 0 { + t.Fatalf("expected reset totals to be zero, got in=%d out=%d", + state.session.TokenInputTotal, state.session.TokenOutputTotal) } before := state.session.UpdatedAt @@ -74,7 +69,7 @@ func TestRunStateMutationsAndSync(t *testing.T) { t.Fatalf("expected touchSession to update time") } if state.session.TokenInputTotal != 1 || state.session.TokenOutputTotal != 2 { - t.Fatalf("expected touchSession to sync totals") + t.Fatalf("expected touchSession to keep latest totals") } } diff --git a/internal/runtime/runtime_test.go b/internal/runtime/runtime_test.go index 20cde84e..d054c21c 100644 --- a/internal/runtime/runtime_test.go +++ b/internal/runtime/runtime_test.go @@ -4059,11 +4059,11 @@ func TestRestoreSessionTokens(t *testing.T) { state := newRunState("", session) - if state.tokenInputTotal != 500 { - t.Fatalf("expected sessionInputTokens == 500, got %d", state.tokenInputTotal) + if state.session.TokenInputTotal != 500 { + t.Fatalf("expected sessionInputTokens == 500, got %d", state.session.TokenInputTotal) } - if state.tokenOutputTotal != 200 { - t.Fatalf("expected sessionOutputTokens == 200, got %d", state.tokenOutputTotal) + if state.session.TokenOutputTotal != 200 { + t.Fatalf("expected sessionOutputTokens == 200, got %d", state.session.TokenOutputTotal) } } @@ -4077,11 +4077,11 @@ func TestRestoreSessionTokensNewSession(t *testing.T) { state := newRunState("", session) - if state.tokenInputTotal != 0 { - t.Fatalf("expected sessionInputTokens == 0, got %d", state.tokenInputTotal) + if state.session.TokenInputTotal != 0 { + t.Fatalf("expected sessionInputTokens == 0, got %d", state.session.TokenInputTotal) } - if state.tokenOutputTotal != 0 { - t.Fatalf("expected sessionOutputTokens == 0, got %d", state.tokenOutputTotal) + if state.session.TokenOutputTotal != 0 { + t.Fatalf("expected sessionOutputTokens == 0, got %d", state.session.TokenOutputTotal) } } @@ -4165,8 +4165,8 @@ func TestTokenUsageRecordedOnMessageDone(t *testing.T) { service.emit(context.Background(), EventTokenUsage, "test-run-id", "test-session-id", TokenUsagePayload{ InputTokens: payload.Usage.InputTokens, OutputTokens: payload.Usage.OutputTokens, - SessionInputTokens: state.tokenInputTotal, - SessionOutputTokens: state.tokenOutputTotal, + SessionInputTokens: state.session.TokenInputTotal, + SessionOutputTokens: state.session.TokenOutputTotal, }) } }}, @@ -4176,11 +4176,11 @@ func TestTokenUsageRecordedOnMessageDone(t *testing.T) { } // Verify the service counters are updated - if state.tokenInputTotal != 100 { - t.Fatalf("expected sessionInputTokens == 100, got %d", state.tokenInputTotal) + if state.session.TokenInputTotal != 100 { + t.Fatalf("expected sessionInputTokens == 100, got %d", state.session.TokenInputTotal) } - if state.tokenOutputTotal != 50 { - t.Fatalf("expected sessionOutputTokens == 50, got %d", state.tokenOutputTotal) + if state.session.TokenOutputTotal != 50 { + t.Fatalf("expected sessionOutputTokens == 50, got %d", state.session.TokenOutputTotal) } // Verify EventTokenUsage was emitted with correct payload diff --git a/internal/runtime/session_mutation.go b/internal/runtime/session_mutation.go index 9e3ab972..e52a1574 100644 --- a/internal/runtime/session_mutation.go +++ b/internal/runtime/session_mutation.go @@ -33,7 +33,6 @@ func (s *Service) appendAssistantMessageAndSave( metadataChanged := state.session.Provider != snapshot.providerConfig.Name || state.session.Model != snapshot.model state.session.Provider = snapshot.providerConfig.Name state.session.Model = snapshot.model - state.syncSessionTokenTotals() if strings.TrimSpace(assistant.Content) != "" || len(assistant.ToolCalls) > 0 { state.session.Messages = append(state.session.Messages, assistant) diff --git a/internal/runtime/state.go b/internal/runtime/state.go index a5d24411..c0b160bc 100644 --- a/internal/runtime/state.go +++ b/internal/runtime/state.go @@ -1,6 +1,7 @@ package runtime import ( + "sync" "time" "neo-code/internal/config" @@ -15,43 +16,33 @@ const maxReactiveCompactAttempts = 3 // runState 汇总单次 Run 生命周期内会变化的会话与计量状态。 type runState struct { + mu sync.Mutex runID string session agentsession.Session - tokenInputTotal int - tokenOutputTotal int compactApplied bool reactiveCompactAttempts int rememberedThisRun bool turn int phase controlplane.Phase + stopEmitted bool + progress controlplane.ProgressState } // newRunState 基于持久化会话创建一次运行的内存状态镜像。 func newRunState(runID string, session agentsession.Session) runState { return runState{ - runID: runID, - session: session, - tokenInputTotal: session.TokenInputTotal, - tokenOutputTotal: session.TokenOutputTotal, + runID: runID, + session: session, } } -// syncSessionTokenTotals 将运行期 token 计数同步回会话对象。 -func (s *runState) syncSessionTokenTotals() { - if s == nil { - return - } - s.session.TokenInputTotal = s.tokenInputTotal - s.session.TokenOutputTotal = s.tokenOutputTotal -} - // recordUsage 累加本轮 provider 返回的 token 使用量。 func (s *runState) recordUsage(inputTokens int, outputTokens int) { if s == nil { return } - s.tokenInputTotal += inputTokens - s.tokenOutputTotal += outputTokens + s.session.TokenInputTotal += inputTokens + s.session.TokenOutputTotal += outputTokens } // resetTokenTotals 在 compact 应用成功后清零当前运行的 token 账本。 @@ -59,17 +50,15 @@ func (s *runState) resetTokenTotals() { if s == nil { return } - s.tokenInputTotal = 0 - s.tokenOutputTotal = 0 - s.syncSessionTokenTotals() + s.session.TokenInputTotal = 0 + s.session.TokenOutputTotal = 0 } -// touchSession 更新会话修改时间并同步最新 token 累计值。 +// touchSession 更新会话修改时间。 func (s *runState) touchSession() { if s == nil { return } - s.syncSessionTokenTotals() s.session.UpdatedAt = time.Now() } From 1afecba70d665080d528690c97bde21b2ac34a03 Mon Sep 17 00:00:00 2001 From: xgopilot Date: Wed, 15 Apr 2026 02:45:28 +0000 Subject: [PATCH 09/10] test(runtime): align helper tests with main to remove merge conflicts Generated with [codeagent](https://github.com/qbox/codeagent) Co-authored-by: Cai-Tang-www <106404101+Cai-Tang-www@users.noreply.github.com> --- .../runtime/runtime_internal_helpers_test.go | 62 +++++++++++++++---- 1 file changed, 49 insertions(+), 13 deletions(-) diff --git a/internal/runtime/runtime_internal_helpers_test.go b/internal/runtime/runtime_internal_helpers_test.go index c1952b82..97d779ad 100644 --- a/internal/runtime/runtime_internal_helpers_test.go +++ b/internal/runtime/runtime_internal_helpers_test.go @@ -7,7 +7,6 @@ import ( "testing" "time" - "neo-code/internal/config" "neo-code/internal/provider" providertypes "neo-code/internal/provider/types" "neo-code/internal/runtime/approval" @@ -22,6 +21,25 @@ type stubMemoExtractor struct { doneCh chan struct{} } +type lockProbeStore struct { + saveFn func(ctx context.Context, session *agentsession.Session) error +} + +func (s *lockProbeStore) Save(ctx context.Context, session *agentsession.Session) error { + if s.saveFn == nil { + return nil + } + return s.saveFn(ctx, session) +} + +func (s *lockProbeStore) Load(ctx context.Context, id string) (agentsession.Session, error) { + return agentsession.Session{}, errors.New("not implemented") +} + +func (s *lockProbeStore) ListSummaries(ctx context.Context) ([]agentsession.Summary, error) { + return nil, errors.New("not implemented") +} + func (s *stubMemoExtractor) Schedule(_ string, messages []providertypes.Message) { s.mu.Lock() s.calls++ @@ -53,13 +71,12 @@ func TestRunStateMutationsAndSync(t *testing.T) { state.recordUsage(10, 20) if state.session.TokenInputTotal != 11 || state.session.TokenOutputTotal != 22 { - t.Fatalf("session totals not synced: %+v", state.session) + t.Fatalf("unexpected token totals: in=%d out=%d", state.session.TokenInputTotal, state.session.TokenOutputTotal) } state.resetTokenTotals() if state.session.TokenInputTotal != 0 || state.session.TokenOutputTotal != 0 { - t.Fatalf("expected reset totals to be zero, got in=%d out=%d", - state.session.TokenInputTotal, state.session.TokenOutputTotal) + t.Fatalf("expected reset totals to be zero, got in=%d out=%d", state.session.TokenInputTotal, state.session.TokenOutputTotal) } before := state.session.UpdatedAt @@ -69,7 +86,7 @@ func TestRunStateMutationsAndSync(t *testing.T) { t.Fatalf("expected touchSession to update time") } if state.session.TokenInputTotal != 1 || state.session.TokenOutputTotal != 2 { - t.Fatalf("expected touchSession to keep latest totals") + t.Fatalf("expected recordUsage to sync totals") } } @@ -138,17 +155,36 @@ func TestAppendToolMessageAndSaveSanitizesMetadata(t *testing.T) { } } -func TestResolveMaxLoopsBranches(t *testing.T) { +func TestAppendToolMessageAndSaveUnlocksStateBeforePersist(t *testing.T) { t.Parallel() - if got := resolveMaxLoops(config.Config{MaxLoops: 0}); got != defaultMaxLoops { - t.Fatalf("expected default max loops for zero, got %d", got) - } - if got := resolveMaxLoops(config.Config{MaxLoops: -3}); got != defaultMaxLoops { - t.Fatalf("expected default max loops for negative, got %d", got) + session := newRuntimeSession("session-append-tool-lock") + state := newRunState("run-append-tool-lock", session) + + store := &lockProbeStore{ + saveFn: func(_ context.Context, _ *agentsession.Session) error { + locked := make(chan struct{}) + go func() { + state.mu.Lock() + state.mu.Unlock() + close(locked) + }() + + select { + case <-locked: + return nil + case <-time.After(200 * time.Millisecond): + return errors.New("state lock is still held during save") + } + }, } - if got := resolveMaxLoops(config.Config{MaxLoops: 12}); got != 12 { - t.Fatalf("expected explicit max loops, got %d", got) + + service := &Service{sessionStore: store} + call := providertypes.ToolCall{ID: "call-1", Name: "filesystem_read_file"} + result := tools.ToolResult{Name: "filesystem_read_file", Content: "ok"} + + if err := service.appendToolMessageAndSave(context.Background(), &state, call, result); err != nil { + t.Fatalf("appendToolMessageAndSave() error = %v", err) } } From 727ed3fc1f1e9300b46d899f419147dd784d048b Mon Sep 17 00:00:00 2001 From: xgopilot Date: Wed, 15 Apr 2026 03:26:26 +0000 Subject: [PATCH 10/10] fix(subagent): address review gaps in progress envelope and capability bounds Generated with [codeagent](https://github.com/qbox/codeagent) Co-authored-by: Cai-Tang-www <106404101+Cai-Tang-www@users.noreply.github.com> --- internal/runtime/subagent_run.go | 28 ++++-- internal/runtime/subagent_run_test.go | 123 ++++++++++++++++++++++++++ internal/subagent/worker.go | 17 +++- internal/subagent/worker_test.go | 61 +++++++++++++ 4 files changed, 221 insertions(+), 8 deletions(-) diff --git a/internal/runtime/subagent_run.go b/internal/runtime/subagent_run.go index bf0adbd8..31bbd2b6 100644 --- a/internal/runtime/subagent_run.go +++ b/internal/runtime/subagent_run.go @@ -5,7 +5,9 @@ import ( "errors" "fmt" "strings" + "time" + "neo-code/internal/runtime/controlplane" "neo-code/internal/subagent" ) @@ -118,7 +120,7 @@ func (s *Service) RunSubAgentTask(ctx context.Context, input SubAgentTaskInput) if result.State == subagent.StateSucceeded { return result, nil } - return result, errors.New(result.Error) + return result, subAgentResultError(result) } } @@ -132,13 +134,17 @@ func emitSubAgentProgress(s *Service, input SubAgentTaskInput, stepResult subage Delta: stepResult.Delta, Error: errorText(stepErr), } + event := RuntimeEvent{ + Type: EventSubAgentProgress, + RunID: input.RunID, + SessionID: input.SessionID, + Turn: turnUnspecified, + Timestamp: time.Now(), + PayloadVersion: controlplane.PayloadVersion, + Payload: payload, + } select { - case s.events <- RuntimeEvent{ - Type: EventSubAgentProgress, - RunID: input.RunID, - SessionID: input.SessionID, - Payload: payload, - }: + case s.events <- event: default: } } @@ -171,3 +177,11 @@ func errorText(err error) string { } return strings.TrimSpace(err.Error()) } + +// subAgentResultError 将子代理终态结果转换为可诊断错误，避免空错误文本丢失上下文。 +func subAgentResultError(result subagent.Result) error { + if text := strings.TrimSpace(result.Error); text != "" { + return errors.New(text) + } + return fmt.Errorf("subagent ended with state=%s stop_reason=%s", result.State, result.StopReason) +} diff --git a/internal/runtime/subagent_run_test.go b/internal/runtime/subagent_run_test.go index 116abaf2..96f20f81 100644 --- a/internal/runtime/subagent_run_test.go +++ b/internal/runtime/subagent_run_test.go @@ -3,9 +3,11 @@ package runtime import ( "context" "errors" + "strings" "testing" "time" + "neo-code/internal/runtime/controlplane" "neo-code/internal/subagent" ) @@ -74,6 +76,17 @@ func TestServiceRunSubAgentTaskSuccess(t *testing.T) { EventSubAgentCompleted, }) assertEventsRunID(t, events, "sub-run-success") + for _, evt := range events { + if evt.Type != EventSubAgentProgress { + continue + } + if evt.Timestamp.IsZero() { + t.Fatalf("progress event timestamp should be set: %+v", evt) + } + if evt.PayloadVersion != controlplane.PayloadVersion { + t.Fatalf("progress event payload version = %d, want %d", evt.PayloadVersion, controlplane.PayloadVersion) + } + } } func TestServiceRunSubAgentTaskFailureFlows(t *testing.T) { @@ -190,6 +203,116 @@ func TestServiceRunSubAgentTaskFailureFlows(t *testing.T) { events := collectRuntimeEvents(service.Events()) assertEventSequence(t, events, []EventType{EventSubAgentFailed}) }) + + t.Run("custom worker failed without explicit error should return fallback", func(t *testing.T) { + t.Parallel() + + service := NewWithFactory(nil, nil, nil, nil, nil) + service.SetSubAgentFactory(stubSubAgentFactory{ + create: func(role subagent.Role) (subagent.WorkerRuntime, error) { + return &stubSubAgentWorker{ + result: subagent.Result{ + Role: role, + TaskID: "task-fallback-error", + State: subagent.StateFailed, + StopReason: subagent.StopReasonError, + }, + stepResult: subagent.StepResult{ + State: subagent.StateFailed, + Done: true, + Step: 1, + }, + }, nil + }, + }) + + _, err := service.RunSubAgentTask(context.Background(), SubAgentTaskInput{ + RunID: "sub-run-fallback-error", + Role: subagent.RoleReviewer, + Task: subagent.Task{ + ID: "task-fallback-error", + Goal: "review", + }, + }) + if err == nil { + t.Fatalf("expected fallback error") + } + if !strings.Contains(err.Error(), "state=failed") || !strings.Contains(err.Error(), "stop_reason=error") { + t.Fatalf("error = %q, want state/stop_reason fallback", err.Error()) + } + }) +} + +type stubSubAgentFactory struct { + create func(role subagent.Role) (subagent.WorkerRuntime, error) +} + +func (s stubSubAgentFactory) Create(role subagent.Role) (subagent.WorkerRuntime, error) { + return s.create(role) +} + +type stubSubAgentWorker struct { + startErr error + stepResult subagent.StepResult + stepErr error + result subagent.Result + resultErr error + current subagent.State + stopInvoked bool +} + +func (s *stubSubAgentWorker) Start(task subagent.Task, budget subagent.Budget, capability subagent.Capability) error { + if s.startErr != nil { + return s.startErr + } + if s.current == "" { + s.current = subagent.StateRunning + } + s.result.Role = firstNonEmptyRole(s.result.Role, subagent.RoleReviewer) + if strings.TrimSpace(s.result.TaskID) == "" { + s.result.TaskID = task.ID + } + return nil +} + +func (s *stubSubAgentWorker) Step(ctx context.Context) (subagent.StepResult, error) { + if s.stepResult.State == "" { + s.stepResult.State = s.current + } + if s.stepResult.Done { + s.current = s.result.State + } + return s.stepResult, s.stepErr +} + +func (s *stubSubAgentWorker) Stop(reason subagent.StopReason) error { + s.stopInvoked = true + s.current = subagent.StateCanceled + s.result.State = subagent.StateCanceled + s.result.StopReason = reason + return nil +} + +func (s *stubSubAgentWorker) Result() (subagent.Result, error) { + return s.result, s.resultErr +} + +func (s *stubSubAgentWorker) State() subagent.State { + if s.current == "" { + return subagent.StateIdle + } + return s.current +} + +func (s *stubSubAgentWorker) Policy() subagent.RolePolicy { + return subagent.RolePolicy{} +} + +func firstNonEmptyRole(role subagent.Role, fallback subagent.Role) subagent.Role { + if role != "" { + return role + } + return fallback } func TestServiceRunSubAgentTaskInputValidation(t *testing.T) { diff --git a/internal/subagent/worker.go b/internal/subagent/worker.go index 6e6c791a..18e9cf17 100644 --- a/internal/subagent/worker.go +++ b/internal/subagent/worker.go @@ -147,7 +147,7 @@ func (w *worker) Step(ctx context.Context) (StepResult, error) { w.stepCount++ delta := strings.TrimSpace(stepOutput.Delta) if delta != "" { - w.trace = append(w.trace, delta) + w.trace = appendTraceBounded(w.trace, delta, traceWindowSize) } if stepOutput.Done { @@ -174,6 +174,10 @@ func (w *worker) Step(ctx context.Context) (StepResult, error) { // bindCapabilityToPolicy 将 capability 约束在角色策略允许的工具集合内。 func bindCapabilityToPolicy(capability Capability, policy RolePolicy) (Capability, error) { + if len(capability.AllowedPaths) > 0 { + return Capability{}, errorsf("capability allowed paths is not supported yet") + } + allowedPolicyTools := dedupeAndTrim(policy.AllowedTools) allowedSet := make(map[string]struct{}, len(allowedPolicyTools)) for _, tool := range allowedPolicyTools { @@ -202,6 +206,17 @@ func bindCapabilityToPolicy(capability Capability, policy RolePolicy) (Capabilit return capability, nil } +// appendTraceBounded 将新增 trace 追加到切片尾部，并保证内部存储长度不超过上限。 +func appendTraceBounded(trace []string, delta string, limit int) []string { + trace = append(trace, delta) + if limit <= 0 || len(trace) <= limit { + return trace + } + start := len(trace) - limit + copy(trace, trace[start:]) + return trace[:limit] +} + // cloneRecentTrace 复制最近 limit 条 trace，避免每步复制完整历史导致复杂度放大。 func cloneRecentTrace(trace []string, limit int) []string { if len(trace) == 0 { diff --git a/internal/subagent/worker_test.go b/internal/subagent/worker_test.go index b323ad98..532e4c76 100644 --- a/internal/subagent/worker_test.go +++ b/internal/subagent/worker_test.go @@ -276,6 +276,16 @@ func TestWorkerStartCapabilityPolicyGuard(t *testing.T) { t.Fatalf("expected disallowed capability tool to fail") } + wPath, err := NewWorker(RoleReviewer, policy, nil) + if err != nil { + t.Fatalf("NewWorker() error = %v", err) + } + if err := wPath.Start(Task{ID: "t-cap-path", Goal: "goal"}, Budget{}, Capability{ + AllowedPaths: []string{"/tmp/workspace"}, + }); err == nil { + t.Fatalf("expected unsupported allowed paths to fail") + } + w2, err := NewWorker(RoleReviewer, policy, nil) if err != nil { t.Fatalf("NewWorker() error = %v", err) @@ -343,6 +353,57 @@ func TestWorkerTraceWindow(t *testing.T) { } } +func TestWorkerTraceStorageBounded(t *testing.T) { + t.Parallel() + + policy, err := DefaultRolePolicy(RoleResearcher) + if err != nil { + t.Fatalf("DefaultRolePolicy() error = %v", err) + } + + w, err := NewWorker(RoleResearcher, policy, EngineFunc(func(ctx context.Context, input StepInput) (StepOutput, error) { + if input.StepIndex < traceWindowSize+8 { + return StepOutput{Delta: "delta", Done: false}, nil + } + return StepOutput{ + Delta: "delta", + Done: true, + Output: Output{ + Summary: "done", + Findings: []string{"f1"}, + Patches: []string{"p1"}, + Risks: []string{"r1"}, + NextActions: []string{"n1"}, + Artifacts: []string{"a1"}, + }, + }, nil + })) + if err != nil { + t.Fatalf("NewWorker() error = %v", err) + } + if err := w.Start(Task{ID: "t-trace-bounded", Goal: "goal"}, Budget{MaxSteps: traceWindowSize + 16}, Capability{}); err != nil { + t.Fatalf("Start() error = %v", err) + } + + for { + step, stepErr := w.Step(context.Background()) + if stepErr != nil { + t.Fatalf("Step() error = %v", stepErr) + } + if step.Done { + break + } + } + + impl, ok := w.(*worker) + if !ok { + t.Fatalf("expected *worker implementation") + } + if len(impl.trace) != traceWindowSize { + t.Fatalf("trace storage len = %d, want %d", len(impl.trace), traceWindowSize) + } +} + func TestWorkerNilAndValidationBranches(t *testing.T) { t.Parallel()