DBT/clawgo
1
0
Fork 0
mirror of https://github.com/YspCoder/clawgo.git synced 2026-04-14 22:09:37 +08:00
Code Issues Packages Projects Releases Wiki Activity

Release v0.1.0 agent topology and runtime refresh

Browse Source
This commit is contained in:
lpf
2026-03-06 17:44:13 +08:00
parent ac5a1bfcb2
commit 7d9ca89476
34 changed files with 1216 additions and 1462 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
pkg/agent/loop.go
View File

@@ -61,7 +61,6 @@ type AgentLoop struct {
streamMu sync.Mutex
sessionStreamed map[string]bool
subagentManager *tools.SubagentManager
orchestrator *tools.Orchestrator
subagentRouter *tools.SubagentRouter
subagentConfigTool *tools.SubagentConfigTool
nodeRouter *nodes.Router
@@ -184,8 +183,7 @@ func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, provider providers
toolsRegistry.Register(messageTool)
// Register spawn tool
orchestrator := tools.NewOrchestrator()
subagentManager := tools.NewSubagentManager(provider, workspace, msgBus, orchestrator)
subagentManager := tools.NewSubagentManager(provider, workspace, msgBus)
subagentRouter := tools.NewSubagentRouter(subagentManager)
subagentConfigTool := tools.NewSubagentConfigTool("")
spawnTool := tools.NewSpawnTool(subagentManager)
@@ -195,10 +193,6 @@ func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, provider providers
if store := subagentManager.ProfileStore(); store != nil {
toolsRegistry.Register(tools.NewSubagentProfileTool(store))
}
toolsRegistry.Register(tools.NewPipelineCreateTool(orchestrator))
toolsRegistry.Register(tools.NewPipelineStatusTool(orchestrator))
toolsRegistry.Register(tools.NewPipelineStateSetTool(orchestrator))
toolsRegistry.Register(tools.NewPipelineDispatchTool(orchestrator, subagentManager))
toolsRegistry.Register(tools.NewSessionsTool(
func(limit int) []tools.SessionInfo {
sessions := alSessionListForTool(sessionsManager, limit)
@@ -257,7 +251,6 @@ func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, provider providers
providerResponses: map[string]config.ProviderResponsesConfig{},
telegramStreaming: cfg.Channels.Telegram.Streaming,
subagentManager: subagentManager,
orchestrator: orchestrator,
subagentRouter: subagentRouter,
subagentConfigTool: subagentConfigTool,
nodeRouter: nodesRouter,
@@ -1748,7 +1741,7 @@ func withToolContextArgs(toolName string, args map[string]interface{}, channel,
return args
}
switch toolName {
case "message", "spawn", "remind", "pipeline_create", "pipeline_dispatch":
case "message", "spawn", "remind":
default:
return args
}

10
pkg/agent/loop_allowlist_test.go
View File

@@ -52,13 +52,3 @@ func TestEnsureToolAllowedByContext_GroupAllowlist(t *testing.T) {
t.Fatalf("expected files_read group to block write_file")
}
}
func TestEnsureToolAllowedByContext_GroupAliasToken(t *testing.T) {
ctx := withToolAllowlistContext(context.Background(), []string{"@pipeline"})
if err := ensureToolAllowedByContext(ctx, "pipeline_status", map[string]interface{}{}); err != nil {
t.Fatalf("expected @pipeline to allow pipeline_status, got: %v", err)
}
if err := ensureToolAllowedByContext(ctx, "memory_search", map[string]interface{}{}); err == nil {
t.Fatalf("expected @pipeline to block memory_search")
}
}

4
pkg/agent/router_dispatch_test.go
View File

@@ -44,7 +44,7 @@ func TestMaybeAutoRouteDispatchesExplicitAgentMention(t *testing.T) {
t.Cleanup(func() { runtimecfg.Set(config.DefaultConfig()) })
workspace := t.TempDir()
manager := tools.NewSubagentManager(nil, workspace, nil, nil)
manager := tools.NewSubagentManager(nil, workspace, nil)
manager.SetRunFunc(func(ctx context.Context, task *tools.SubagentTask) (string, error) {
return "auto-routed", nil
})
@@ -101,7 +101,7 @@ func TestMaybeAutoRouteDispatchesRulesFirstMatch(t *testing.T) {
t.Cleanup(func() { runtimecfg.Set(config.DefaultConfig()) })
workspace := t.TempDir()
manager := tools.NewSubagentManager(nil, workspace, nil, nil)
manager := tools.NewSubagentManager(nil, workspace, nil)
manager.SetRunFunc(func(ctx context.Context, task *tools.SubagentTask) (string, error) {
return "tested", nil
})

125
pkg/agent/runtime_admin.go
View File

@@ -65,8 +65,6 @@ func (al *AgentLoop) HandleSubagentRuntime(ctx context.Context, action string, a
MaxResultChars: runtimeIntArg(args, "max_result_chars", 0),
OriginChannel: fallbackString(runtimeStringArg(args, "channel"), "webui"),
OriginChatID: fallbackString(runtimeStringArg(args, "chat_id"), "webui"),
PipelineID: runtimeStringArg(args, "pipeline_id"),
PipelineTask: runtimeStringArg(args, "task_id"),
})
if err != nil {
return nil, err
@@ -388,89 +386,6 @@ func (al *AgentLoop) HandleSubagentRuntime(ctx context.Context, action string, a
}
}
func (al *AgentLoop) HandlePipelineRuntime(ctx context.Context, action string, args map[string]interface{}) (interface{}, error) {
if al == nil || al.orchestrator == nil {
return nil, fmt.Errorf("pipeline runtime is not configured")
}
action = strings.ToLower(strings.TrimSpace(action))
if action == "" {
action = "list"
}
switch action {
case "list":
return map[string]interface{}{"items": al.orchestrator.ListPipelines()}, nil
case "get", "status":
pipelineID := fallbackString(runtimeStringArg(args, "pipeline_id"), runtimeStringArg(args, "id"))
if strings.TrimSpace(pipelineID) == "" {
return nil, fmt.Errorf("pipeline_id is required")
}
p, ok := al.orchestrator.GetPipeline(strings.TrimSpace(pipelineID))
if !ok {
return map[string]interface{}{"found": false}, nil
}
return map[string]interface{}{"found": true, "pipeline": p}, nil
case "ready":
pipelineID := fallbackString(runtimeStringArg(args, "pipeline_id"), runtimeStringArg(args, "id"))
if strings.TrimSpace(pipelineID) == "" {
return nil, fmt.Errorf("pipeline_id is required")
}
items, err := al.orchestrator.ReadyTasks(strings.TrimSpace(pipelineID))
if err != nil {
return nil, err
}
return map[string]interface{}{"items": items}, nil
case "create":
objective := runtimeStringArg(args, "objective")
if objective == "" {
return nil, fmt.Errorf("objective is required")
}
specs, err := parsePipelineSpecsForRuntime(args["tasks"])
if err != nil {
return nil, err
}
label := runtimeStringArg(args, "label")
p, err := al.orchestrator.CreatePipeline(label, objective, "webui", "webui", specs)
if err != nil {
return nil, err
}
return map[string]interface{}{"pipeline": p}, nil
case "state_set":
pipelineID := fallbackString(runtimeStringArg(args, "pipeline_id"), runtimeStringArg(args, "id"))
key := runtimeStringArg(args, "key")
if strings.TrimSpace(pipelineID) == "" || strings.TrimSpace(key) == "" {
return nil, fmt.Errorf("pipeline_id and key are required")
}
value, ok := args["value"]
if !ok {
return nil, fmt.Errorf("value is required")
}
if err := al.orchestrator.SetSharedState(strings.TrimSpace(pipelineID), strings.TrimSpace(key), value); err != nil {
return nil, err
}
p, _ := al.orchestrator.GetPipeline(strings.TrimSpace(pipelineID))
return map[string]interface{}{"ok": true, "pipeline": p}, nil
case "dispatch":
pipelineID := fallbackString(runtimeStringArg(args, "pipeline_id"), runtimeStringArg(args, "id"))
if strings.TrimSpace(pipelineID) == "" {
return nil, fmt.Errorf("pipeline_id is required")
}
maxDispatch := runtimeIntArg(args, "max_dispatch", 3)
dispatchTool := tools.NewPipelineDispatchTool(al.orchestrator, al.subagentManager)
result, err := dispatchTool.Execute(ctx, map[string]interface{}{
"pipeline_id": strings.TrimSpace(pipelineID),
"max_dispatch": float64(maxDispatch),
})
if err != nil {
return nil, err
}
p, _ := al.orchestrator.GetPipeline(strings.TrimSpace(pipelineID))
return map[string]interface{}{"message": result, "pipeline": p}, nil
default:
return nil, fmt.Errorf("unsupported action: %s", action)
}
}
func cloneSubagentTask(in *tools.SubagentTask) *tools.SubagentTask {
if in == nil {
return nil
@@ -514,46 +429,6 @@ func resolveSubagentTaskIDForRuntime(sm *tools.SubagentManager, raw string) (str
return tasks[idx-1].ID, nil
}
func parsePipelineSpecsForRuntime(raw interface{}) ([]tools.PipelineSpec, error) {
items, ok := raw.([]interface{})
if !ok || len(items) == 0 {
return nil, fmt.Errorf("tasks is required")
}
specs := make([]tools.PipelineSpec, 0, len(items))
for i, item := range items {
m, ok := item.(map[string]interface{})
if !ok {
return nil, fmt.Errorf("tasks[%d] must be object", i)
}
id := runtimeStringArg(m, "id")
if id == "" {
return nil, fmt.Errorf("tasks[%d].id is required", i)
}
goal := runtimeStringArg(m, "goal")
if goal == "" {
return nil, fmt.Errorf("tasks[%d].goal is required", i)
}
spec := tools.PipelineSpec{
ID: id,
Role: runtimeStringArg(m, "role"),
Goal: goal,
}
if deps, ok := m["depends_on"].([]interface{}); ok {
spec.DependsOn = make([]string, 0, len(deps))
for _, dep := range deps {
d, _ := dep.(string)
d = strings.TrimSpace(d)
if d == "" {
continue
}
spec.DependsOn = append(spec.DependsOn, d)
}
}
specs = append(specs, spec)
}
return specs, nil
}
func runtimeStringArg(args map[string]interface{}, key string) string {
if args == nil {
return ""

12
pkg/agent/runtime_admin_test.go
View File

@@ -13,7 +13,7 @@ import (
func TestHandleSubagentRuntimeDispatchAndWait(t *testing.T) {
workspace := t.TempDir()
manager := tools.NewSubagentManager(nil, workspace, nil, nil)
manager := tools.NewSubagentManager(nil, workspace, nil)
manager.SetRunFunc(func(ctx context.Context, task *tools.SubagentTask) (string, error) {
return "runtime-admin-result", nil
})
@@ -66,7 +66,7 @@ func TestHandleSubagentRuntimeUpsertConfigSubagent(t *testing.T) {
runtimecfg.Set(cfg)
t.Cleanup(func() { runtimecfg.Set(config.DefaultConfig()) })
manager := tools.NewSubagentManager(nil, workspace, nil, nil)
manager := tools.NewSubagentManager(nil, workspace, nil)
loop := &AgentLoop{
configPath: configPath,
subagentManager: manager,
@@ -138,7 +138,7 @@ func TestHandleSubagentRuntimeRegistryAndToggleEnabled(t *testing.T) {
runtimecfg.Set(cfg)
t.Cleanup(func() { runtimecfg.Set(config.DefaultConfig()) })
manager := tools.NewSubagentManager(nil, workspace, nil, nil)
manager := tools.NewSubagentManager(nil, workspace, nil)
loop := &AgentLoop{
configPath: configPath,
subagentManager: manager,
@@ -197,7 +197,7 @@ func TestHandleSubagentRuntimeDeleteConfigSubagent(t *testing.T) {
runtimecfg.Set(cfg)
t.Cleanup(func() { runtimecfg.Set(config.DefaultConfig()) })
manager := tools.NewSubagentManager(nil, workspace, nil, nil)
manager := tools.NewSubagentManager(nil, workspace, nil)
loop := &AgentLoop{
configPath: configPath,
subagentManager: manager,
@@ -225,7 +225,7 @@ func TestHandleSubagentRuntimeDeleteConfigSubagent(t *testing.T) {
func TestHandleSubagentRuntimePromptFileGetSetBootstrap(t *testing.T) {
workspace := t.TempDir()
manager := tools.NewSubagentManager(nil, workspace, nil, nil)
manager := tools.NewSubagentManager(nil, workspace, nil)
loop := &AgentLoop{
workspace: workspace,
subagentManager: manager,
@@ -297,7 +297,7 @@ func TestHandleSubagentRuntimeProtectsMainAgent(t *testing.T) {
runtimecfg.Set(cfg)
t.Cleanup(func() { runtimecfg.Set(config.DefaultConfig()) })
manager := tools.NewSubagentManager(nil, workspace, nil, nil)
manager := tools.NewSubagentManager(nil, workspace, nil)
loop := &AgentLoop{
configPath: configPath,
workspace: workspace,