6 Commits

Author SHA1 Message Date
lpf
823f96be5a Stop planned task spam after cancellation 2026-03-06 21:31:21 +08:00
lpf
9d0ab54a97 Recover running subagent tasks after restart 2026-03-06 20:05:03 +08:00
lpf
ee9326b2f2 Reduce planned task progress noise 2026-03-06 19:57:54 +08:00
lpf
b4cf4a123b Fix planned task splitting and progress summaries 2026-03-06 19:22:57 +08:00
lpf
623b401850 Fix config reload and subagent config feedback 2026-03-06 18:45:43 +08:00
lpf
61c46fd710 Generate gateway token during onboard 2026-03-06 17:53:01 +08:00
14 changed files with 556 additions and 157 deletions

View File

@@ -154,8 +154,12 @@ func gatewayCmd() {
}
return out
})
reloadReqCh := make(chan struct{}, 1)
registryServer.SetConfigAfterHook(func() {
_ = requestGatewayReloadSignal()
select {
case reloadReqCh <- struct{}{}:
default:
}
})
registryServer.SetSubagentHandler(func(cctx context.Context, action string, args map[string]interface{}) (interface{}, error) {
return agentLoop.HandleSubagentRuntime(cctx, action, args)
@@ -309,77 +313,35 @@ func gatewayCmd() {
sigChan := make(chan os.Signal, 1)
signal.Notify(sigChan, gatewayNotifySignals()...)
for {
sig := <-sigChan
switch {
case isGatewayReloadSignal(sig):
fmt.Println("\n↻ Reloading config...")
newCfg, err := config.LoadConfig(getConfigPath())
if err != nil {
fmt.Printf("✗ Reload failed (load config): %v\n", err)
continue
}
if strings.EqualFold(strings.TrimSpace(os.Getenv(envRootGranted)), "1") || strings.EqualFold(strings.TrimSpace(os.Getenv(envRootGranted)), "true") {
applyMaximumPermissionPolicy(newCfg)
}
configureCronServiceRuntime(cronService, newCfg)
heartbeatService.Stop()
heartbeatService = buildHeartbeatService(newCfg, msgBus)
if err := heartbeatService.Start(); err != nil {
fmt.Printf("Error starting heartbeat service: %v\n", err)
}
applyReload := func() {
fmt.Println("\n↻ Reloading config...")
newCfg, err := config.LoadConfig(getConfigPath())
if err != nil {
fmt.Printf(" Reload failed (load config): %v\n", err)
return
}
if strings.EqualFold(strings.TrimSpace(os.Getenv(envRootGranted)), "1") || strings.EqualFold(strings.TrimSpace(os.Getenv(envRootGranted)), "true") {
applyMaximumPermissionPolicy(newCfg)
}
configureCronServiceRuntime(cronService, newCfg)
heartbeatService.Stop()
heartbeatService = buildHeartbeatService(newCfg, msgBus)
if err := heartbeatService.Start(); err != nil {
fmt.Printf("Error starting heartbeat service: %v\n", err)
}
if reflect.DeepEqual(cfg, newCfg) {
fmt.Println("✓ Config unchanged, skip reload")
continue
}
if reflect.DeepEqual(cfg, newCfg) {
fmt.Println("✓ Config unchanged, skip reload")
return
}
runtimeSame := reflect.DeepEqual(cfg.Agents, newCfg.Agents) &&
reflect.DeepEqual(cfg.Providers, newCfg.Providers) &&
reflect.DeepEqual(cfg.Tools, newCfg.Tools) &&
reflect.DeepEqual(cfg.Channels, newCfg.Channels)
runtimeSame := reflect.DeepEqual(cfg.Agents, newCfg.Agents) &&
reflect.DeepEqual(cfg.Providers, newCfg.Providers) &&
reflect.DeepEqual(cfg.Tools, newCfg.Tools) &&
reflect.DeepEqual(cfg.Channels, newCfg.Channels)
if runtimeSame {
configureLogging(newCfg)
sentinelService.Stop()
sentinelService = sentinel.NewService(
getConfigPath(),
newCfg.WorkspacePath(),
newCfg.Sentinel.IntervalSec,
newCfg.Sentinel.AutoHeal,
func(message string) {
if newCfg.Sentinel.NotifyChannel != "" && newCfg.Sentinel.NotifyChatID != "" {
msgBus.PublishOutbound(bus.OutboundMessage{
Channel: newCfg.Sentinel.NotifyChannel,
ChatID: newCfg.Sentinel.NotifyChatID,
Content: "[Sentinel] " + message,
})
}
},
)
if newCfg.Sentinel.Enabled {
sentinelService.SetManager(channelManager)
sentinelService.Start()
}
cfg = newCfg
runtimecfg.Set(cfg)
fmt.Println("✓ Config hot-reload applied (logging/metadata only)")
continue
}
newAgentLoop, newChannelManager, err := buildGatewayRuntime(ctx, newCfg, msgBus, cronService)
if err != nil {
fmt.Printf("✗ Reload failed (init runtime): %v\n", err)
continue
}
channelManager.StopAll(ctx)
agentLoop.Stop()
channelManager = newChannelManager
agentLoop = newAgentLoop
cfg = newCfg
runtimecfg.Set(cfg)
if runtimeSame {
configureLogging(newCfg)
sentinelService.Stop()
sentinelService = sentinel.NewService(
getConfigPath(),
@@ -397,27 +359,77 @@ func gatewayCmd() {
},
)
if newCfg.Sentinel.Enabled {
sentinelService.SetManager(channelManager)
sentinelService.Start()
}
sentinelService.SetManager(channelManager)
if err := channelManager.StartAll(ctx); err != nil {
fmt.Printf("✗ Reload failed (start channels): %v\n", err)
continue
}
go agentLoop.Run(ctx)
fmt.Println("✓ Config hot-reload applied")
default:
fmt.Println("\nShutting down...")
cancel()
heartbeatService.Stop()
sentinelService.Stop()
cronService.Stop()
agentLoop.Stop()
channelManager.StopAll(ctx)
fmt.Println("✓ Gateway stopped")
cfg = newCfg
runtimecfg.Set(cfg)
fmt.Println("✓ Config hot-reload applied (logging/metadata only)")
return
}
newAgentLoop, newChannelManager, err := buildGatewayRuntime(ctx, newCfg, msgBus, cronService)
if err != nil {
fmt.Printf("✗ Reload failed (init runtime): %v\n", err)
return
}
channelManager.StopAll(ctx)
agentLoop.Stop()
channelManager = newChannelManager
agentLoop = newAgentLoop
cfg = newCfg
runtimecfg.Set(cfg)
sentinelService.Stop()
sentinelService = sentinel.NewService(
getConfigPath(),
newCfg.WorkspacePath(),
newCfg.Sentinel.IntervalSec,
newCfg.Sentinel.AutoHeal,
func(message string) {
if newCfg.Sentinel.NotifyChannel != "" && newCfg.Sentinel.NotifyChatID != "" {
msgBus.PublishOutbound(bus.OutboundMessage{
Channel: newCfg.Sentinel.NotifyChannel,
ChatID: newCfg.Sentinel.NotifyChatID,
Content: "[Sentinel] " + message,
})
}
},
)
if newCfg.Sentinel.Enabled {
sentinelService.Start()
}
sentinelService.SetManager(channelManager)
if err := channelManager.StartAll(ctx); err != nil {
fmt.Printf("✗ Reload failed (start channels): %v\n", err)
return
}
go agentLoop.Run(ctx)
fmt.Println("✓ Config hot-reload applied")
}
for {
select {
case <-reloadReqCh:
applyReload()
case sig := <-sigChan:
switch {
case isGatewayReloadSignal(sig):
applyReload()
default:
fmt.Println("\nShutting down...")
cancel()
heartbeatService.Stop()
sentinelService.Stop()
cronService.Stop()
agentLoop.Stop()
channelManager.StopAll(ctx)
fmt.Println("✓ Gateway stopped")
return
}
}
}
}

View File

@@ -44,6 +44,9 @@ func ensureConfigOnboard(configPath string, defaults *config.Config) (string, er
if defaults == nil {
return "", fmt.Errorf("defaults is nil")
}
if defaults.Gateway.Token == "" {
defaults.Gateway.Token = config.DefaultConfig().Gateway.Token
}
exists := true
if _, err := os.Stat(configPath); os.IsNotExist(err) {

View File

@@ -0,0 +1,31 @@
package main
import (
"testing"
"clawgo/pkg/config"
)
func TestEnsureConfigOnboardGeneratesGatewayToken(t *testing.T) {
t.Parallel()
configPath := t.TempDir() + "/config.json"
cfg := config.DefaultConfig()
cfg.Gateway.Token = ""
state, err := ensureConfigOnboard(configPath, cfg)
if err != nil {
t.Fatalf("ensureConfigOnboard failed: %v", err)
}
if state != "created" {
t.Fatalf("unexpected state: %s", state)
}
loaded, err := config.LoadConfig(configPath)
if err != nil {
t.Fatalf("load config failed: %v", err)
}
if loaded.Gateway.Token == "" {
t.Fatalf("expected gateway token to be generated")
}
}

View File

@@ -9,6 +9,7 @@ package agent
import (
"context"
"encoding/json"
"errors"
"fmt"
"hash/fnv"
"math"
@@ -544,6 +545,11 @@ func (al *AgentLoop) processInbound(ctx context.Context, msg bus.InboundMessage)
response, err := al.processPlannedMessage(ctx, msg)
if err != nil {
if errors.Is(err, context.Canceled) {
al.audit.Record(al.getTrigger(msg), msg.Channel, msg.SessionKey, true, err)
al.appendTaskAudit(taskID, msg, started, err, true)
return
}
response = fmt.Sprintf("Error processing message: %v", err)
}

View File

@@ -4,7 +4,9 @@ import (
"bufio"
"context"
"encoding/json"
"errors"
"fmt"
"math"
"os"
"path/filepath"
"regexp"
@@ -26,6 +28,7 @@ type plannedTaskResult struct {
Index int
Task plannedTask
Output string
Err error
ErrText string
}
@@ -103,7 +106,10 @@ func splitPlannedSegments(content string) []string {
return bullet
}
replaced := strings.NewReplacer("", ";", "\n", ";", "。然后", ";", " 然后 ", ";", " and then ", ";")
// Only split implicit plans on strong separators. Plain newlines are often
// just formatting inside a single request, and "然后/and then" frequently
// describes execution order inside one task rather than separate tasks.
replaced := strings.NewReplacer("", ";")
norm := replaced.Replace(content)
parts := strings.Split(norm, ";")
out := make([]string, 0, len(parts))
@@ -120,6 +126,11 @@ func splitPlannedSegments(content string) []string {
func (al *AgentLoop) runPlannedTasks(ctx context.Context, msg bus.InboundMessage, tasks []plannedTask) (string, error) {
results := make([]plannedTaskResult, len(tasks))
var wg sync.WaitGroup
var progressMu sync.Mutex
completed := 0
failed := 0
milestones := plannedProgressMilestones(len(tasks))
notified := make(map[int]struct{}, len(milestones))
for i, task := range tasks {
wg.Add(1)
go func(index int, t plannedTask) {
@@ -134,15 +145,33 @@ func (al *AgentLoop) runPlannedTasks(ctx context.Context, msg bus.InboundMessage
subMsg.Metadata["planned_task_index"] = fmt.Sprintf("%d", t.Index)
subMsg.Metadata["planned_task_total"] = fmt.Sprintf("%d", len(tasks))
out, err := al.processMessage(ctx, subMsg)
res := plannedTaskResult{Index: index, Task: t, Output: strings.TrimSpace(out)}
res := plannedTaskResult{Index: index, Task: t, Output: strings.TrimSpace(out), Err: err}
if err != nil {
res.ErrText = err.Error()
}
results[index] = res
al.publishPlannedTaskProgress(msg, len(tasks), res)
progressMu.Lock()
completed++
if res.ErrText != "" && !isPlannedTaskCancellation(ctx, res) {
failed++
}
snapshotCompleted := completed
snapshotFailed := failed
shouldNotify := shouldPublishPlannedTaskProgress(ctx, len(tasks), snapshotCompleted, res, milestones, notified)
if shouldNotify && res.ErrText == "" {
notified[snapshotCompleted] = struct{}{}
}
progressMu.Unlock()
if shouldNotify {
al.publishPlannedTaskProgress(msg, len(tasks), snapshotCompleted, snapshotFailed, res)
}
}(i, task)
}
wg.Wait()
if err := ctx.Err(); err != nil {
return "", err
}
var b strings.Builder
b.WriteString(fmt.Sprintf("已自动拆解为 %d 个任务并执行:\n\n", len(results)))
for _, r := range results {
@@ -160,7 +189,63 @@ func (al *AgentLoop) runPlannedTasks(ctx context.Context, msg bus.InboundMessage
return strings.TrimSpace(b.String()), nil
}
func (al *AgentLoop) publishPlannedTaskProgress(msg bus.InboundMessage, total int, res plannedTaskResult) {
func plannedProgressMilestones(total int) []int {
if total <= 3 {
return nil
}
points := []float64{0.33, 0.66}
out := make([]int, 0, len(points))
seen := map[int]struct{}{}
for _, p := range points {
step := int(math.Round(float64(total) * p))
if step <= 0 || step >= total {
continue
}
if _, ok := seen[step]; ok {
continue
}
seen[step] = struct{}{}
out = append(out, step)
}
return out
}
func shouldPublishPlannedTaskProgress(ctx context.Context, total, completed int, res plannedTaskResult, milestones []int, notified map[int]struct{}) bool {
if total <= 1 {
return false
}
if isPlannedTaskCancellation(ctx, res) {
return false
}
if strings.TrimSpace(res.ErrText) != "" {
return true
}
if completed >= total {
return false
}
for _, step := range milestones {
if completed != step {
continue
}
if _, ok := notified[step]; ok {
return false
}
return true
}
return false
}
func isPlannedTaskCancellation(ctx context.Context, res plannedTaskResult) bool {
if res.Err != nil && errors.Is(res.Err, context.Canceled) {
return true
}
if strings.EqualFold(strings.TrimSpace(res.ErrText), context.Canceled.Error()) {
return true
}
return ctx != nil && errors.Is(ctx.Err(), context.Canceled)
}
func (al *AgentLoop) publishPlannedTaskProgress(msg bus.InboundMessage, total, completed, failed int, res plannedTaskResult) {
if al == nil || al.bus == nil || total <= 1 {
return
}
@@ -180,8 +265,8 @@ func (al *AgentLoop) publishPlannedTaskProgress(msg bus.InboundMessage, total in
if body == "" {
body = "(无输出)"
}
body = truncate(strings.ReplaceAll(body, "\n", " "), 280)
content := fmt.Sprintf("进度 %d/%d任务%d已%s\n%s", idx, total, idx, status, body)
body = summarizePlannedTaskProgressBody(body, 6, 320)
content := fmt.Sprintf("阶段进度 %d/%d(失败 %d\n最近任务%d 已%s\n%s", completed, total, failed, idx, status, body)
al.bus.PublishOutbound(bus.OutboundMessage{
Channel: msg.Channel,
ChatID: msg.ChatID,
@@ -189,6 +274,37 @@ func (al *AgentLoop) publishPlannedTaskProgress(msg bus.InboundMessage, total in
})
}
func summarizePlannedTaskProgressBody(body string, maxLines, maxChars int) string {
body = strings.ReplaceAll(body, "\r\n", "\n")
body = strings.TrimSpace(body)
if body == "" {
return "(无输出)"
}
lines := strings.Split(body, "\n")
out := make([]string, 0, len(lines))
for _, line := range lines {
line = strings.TrimSpace(line)
if line == "" {
continue
}
out = append(out, line)
if maxLines > 0 && len(out) >= maxLines {
break
}
}
if len(out) == 0 {
return "(无输出)"
}
joined := strings.Join(out, "\n")
if maxChars > 0 && len(joined) > maxChars {
joined = truncate(joined, maxChars)
}
if len(lines) > len(out) && !strings.HasSuffix(joined, "...") {
joined += "\n..."
}
return joined
}
func (al *AgentLoop) enrichTaskContentWithMemoryAndEKG(ctx context.Context, task plannedTask) string {
base := strings.TrimSpace(task.Content)
if base == "" {

View File

@@ -0,0 +1,63 @@
package agent
import (
"context"
"errors"
"testing"
)
func TestPlannedProgressMilestones(t *testing.T) {
t.Parallel()
got := plannedProgressMilestones(12)
if len(got) != 2 || got[0] != 4 || got[1] != 8 {
t.Fatalf("unexpected milestones: %#v", got)
}
}
func TestShouldPublishPlannedTaskProgress(t *testing.T) {
t.Parallel()
milestones := plannedProgressMilestones(12)
notified := map[int]struct{}{}
if shouldPublishPlannedTaskProgress(context.Background(), 12, 1, plannedTaskResult{}, milestones, notified) {
t.Fatalf("did not expect early success notification")
}
if !shouldPublishPlannedTaskProgress(context.Background(), 12, 4, plannedTaskResult{}, milestones, notified) {
t.Fatalf("expected milestone notification")
}
notified[4] = struct{}{}
if shouldPublishPlannedTaskProgress(context.Background(), 12, 4, plannedTaskResult{}, milestones, notified) {
t.Fatalf("did not expect duplicate milestone notification")
}
if !shouldPublishPlannedTaskProgress(context.Background(), 12, 5, plannedTaskResult{ErrText: "boom"}, milestones, notified) {
t.Fatalf("expected failure notification")
}
if shouldPublishPlannedTaskProgress(context.Background(), 3, 3, plannedTaskResult{}, plannedProgressMilestones(3), map[int]struct{}{}) {
t.Fatalf("did not expect final success notification")
}
if shouldPublishPlannedTaskProgress(context.Background(), 12, 5, plannedTaskResult{Err: context.Canceled, ErrText: context.Canceled.Error()}, milestones, notified) {
t.Fatalf("did not expect cancellation notification")
}
ctx, cancel := context.WithCancel(context.Background())
cancel()
if shouldPublishPlannedTaskProgress(ctx, 12, 5, plannedTaskResult{Err: errors.New("worker exited after parent stop"), ErrText: "worker exited after parent stop"}, milestones, notified) {
t.Fatalf("did not expect notification after parent cancellation")
}
}
func TestIsPlannedTaskCancellation(t *testing.T) {
t.Parallel()
if !isPlannedTaskCancellation(context.Background(), plannedTaskResult{Err: context.Canceled, ErrText: context.Canceled.Error()}) {
t.Fatalf("expected direct context cancellation to be detected")
}
ctx, cancel := context.WithCancel(context.Background())
cancel()
if !isPlannedTaskCancellation(ctx, plannedTaskResult{Err: errors.New("worker exited after parent stop"), ErrText: "worker exited after parent stop"}) {
t.Fatalf("expected canceled parent context to suppress planned task result")
}
if isPlannedTaskCancellation(context.Background(), plannedTaskResult{Err: errors.New("boom"), ErrText: "boom"}) {
t.Fatalf("did not expect non-cancellation error to be suppressed")
}
}

View File

@@ -0,0 +1,33 @@
package agent
import "testing"
func TestSplitPlannedSegmentsDoesNotSplitPlainNewlines(t *testing.T) {
t.Parallel()
content := "编写ai漫画创作平台demo\n让产品出方案方案出完让前端后端开始编写写完后交个测试过一下"
got := splitPlannedSegments(content)
if len(got) != 1 {
t.Fatalf("expected 1 segment, got %d: %#v", len(got), got)
}
}
func TestSplitPlannedSegmentsStillSplitsBullets(t *testing.T) {
t.Parallel()
content := "1. 先实现前端\n2. 再补测试"
got := splitPlannedSegments(content)
if len(got) != 2 {
t.Fatalf("expected 2 segments, got %d: %#v", len(got), got)
}
}
func TestSplitPlannedSegmentsStillSplitsSemicolons(t *testing.T) {
t.Parallel()
content := "先实现前端;再补测试"
got := splitPlannedSegments(content)
if len(got) != 2 {
t.Fatalf("expected 2 segments, got %d: %#v", len(got), got)
}
}

View File

@@ -0,0 +1,23 @@
package agent
import (
"strings"
"testing"
)
func TestSummarizePlannedTaskProgressBodyPreservesUsefulLines(t *testing.T) {
t.Parallel()
body := "subagent 已写入 config.json。\npath: /root/.clawgo/config.json\nagent_id: tester\nrole: testing\ndisplay_name: Test Agent\ntool_allowlist: [filesystem shell]\nrouting_keywords: [test qa]\nsystem_prompt_file: agents/tester/AGENT.md"
out := summarizePlannedTaskProgressBody(body, 6, 320)
if !strings.Contains(out, "subagent 已写入 config.json。") {
t.Fatalf("expected title line, got:\n%s", out)
}
if !strings.Contains(out, "agent_id: tester") {
t.Fatalf("expected agent id line, got:\n%s", out)
}
if strings.Contains(out, "subagent 已写入 config.json。 path:") {
t.Fatalf("expected multi-line formatting, got:\n%s", out)
}
}

View File

@@ -88,15 +88,49 @@ func extractSubagentDescription(content string) string {
}
func formatCreatedSubagentForUser(result map[string]interface{}, configPath string) string {
subagent, _ := result["subagent"].(map[string]interface{})
role := ""
displayName := ""
toolAllowlist := interface{}(nil)
systemPromptFile := ""
if subagent != nil {
if v, _ := subagent["role"].(string); v != "" {
role = v
}
if v, _ := subagent["display_name"].(string); v != "" {
displayName = v
}
if tools, ok := subagent["tools"].(map[string]interface{}); ok {
toolAllowlist = tools["allowlist"]
}
if v, _ := subagent["system_prompt_file"].(string); v != "" {
systemPromptFile = v
}
}
routingKeywords := interface{}(nil)
if rules, ok := result["rules"].([]interface{}); ok {
agentID, _ := result["agent_id"].(string)
for _, raw := range rules {
rule, ok := raw.(map[string]interface{})
if !ok {
continue
}
if strings.TrimSpace(fmt.Sprint(rule["agent_id"])) != agentID {
continue
}
routingKeywords = rule["keywords"]
break
}
}
return fmt.Sprintf(
"subagent 已写入 config.json。\npath: %s\nagent_id: %v\nrole: %v\ndisplay_name: %v\ntool_allowlist: %v\nrouting_keywords: %v\nsystem_prompt_file: %v",
configPath,
result["agent_id"],
result["role"],
result["display_name"],
result["tool_allowlist"],
result["routing_keywords"],
result["system_prompt_file"],
role,
displayName,
toolAllowlist,
routingKeywords,
systemPromptFile,
)
}

View File

@@ -1,71 +1,43 @@
package agent
import (
"context"
"path/filepath"
"strings"
"testing"
"clawgo/pkg/bus"
"clawgo/pkg/config"
"clawgo/pkg/runtimecfg"
)
func TestMaybeHandleSubagentConfigIntentCreatePersistsImmediately(t *testing.T) {
workspace := t.TempDir()
configPath := filepath.Join(workspace, "config.json")
cfg := config.DefaultConfig()
cfg.Agents.Router.Enabled = true
cfg.Agents.Subagents["main"] = config.SubagentConfig{
Enabled: true,
Type: "router",
Role: "orchestrator",
SystemPromptFile: "agents/main/AGENT.md",
}
if err := config.SaveConfig(configPath, cfg); err != nil {
t.Fatalf("save config failed: %v", err)
}
runtimecfg.Set(cfg)
t.Cleanup(func() { runtimecfg.Set(config.DefaultConfig()) })
func TestFormatCreatedSubagentForUserReadsNestedFields(t *testing.T) {
t.Parallel()
loop := &AgentLoop{configPath: configPath}
out, handled, err := loop.maybeHandleSubagentConfigIntent(context.Background(), bus.InboundMessage{
SessionKey: "main",
Channel: "cli",
Content: "创建一个负责回归测试和验证修复结果的子代理",
})
if err != nil {
t.Fatalf("create subagent failed: %v", err)
}
if !handled || !strings.Contains(out, "已写入 config.json") {
t.Fatalf("expected immediate persist response, got handled=%v out=%q", handled, out)
}
if !strings.Contains(out, configPath) {
t.Fatalf("expected response to include config path, got %q", out)
}
out := formatCreatedSubagentForUser(map[string]interface{}{
"agent_id": "coder",
"subagent": map[string]interface{}{
"role": "coding",
"display_name": "Code Agent",
"system_prompt_file": "agents/coder/AGENT.md",
"tools": map[string]interface{}{
"allowlist": []interface{}{"filesystem", "shell"},
},
},
"rules": []interface{}{
map[string]interface{}{
"agent_id": "coder",
"keywords": []interface{}{"code", "fix"},
},
},
}, "/tmp/config.json")
reloaded, err := config.LoadConfig(configPath)
if err != nil {
t.Fatalf("reload config failed: %v", err)
}
if _, ok := reloaded.Agents.Subagents["tester"]; !ok {
t.Fatalf("expected tester subagent to persist, got %+v", reloaded.Agents.Subagents)
}
}
func TestMaybeHandleSubagentConfigIntentConfirmCancelNoLongerHandled(t *testing.T) {
loop := &AgentLoop{}
for _, content := range []string{"确认创建", "取消创建"} {
out, handled, err := loop.maybeHandleSubagentConfigIntent(context.Background(), bus.InboundMessage{
SessionKey: "main",
Channel: "cli",
Content: content,
})
if err != nil {
t.Fatalf("unexpected error for %q: %v", content, err)
}
if handled || out != "" {
t.Fatalf("expected %q to pass through, got handled=%v out=%q", content, handled, out)
for _, want := range []string{
"agent_id: coder",
"role: coding",
"display_name: Code Agent",
"system_prompt_file: agents/coder/AGENT.md",
"routing_keywords: [code fix]",
} {
if !strings.Contains(out, want) {
t.Fatalf("expected output to contain %q, got:\n%s", want, out)
}
}
if strings.Contains(out, "<nil>") {
t.Fatalf("did not expect nil placeholders, got:\n%s", out)
}
}

View File

@@ -2,6 +2,8 @@ package config
import (
"bytes"
"crypto/rand"
"encoding/hex"
"encoding/json"
"fmt"
"io"
@@ -510,7 +512,7 @@ func DefaultConfig() *Config {
Gateway: GatewayConfig{
Host: "0.0.0.0",
Port: 18790,
Token: "",
Token: generateGatewayToken(),
},
Cron: CronConfig{
MinSleepSec: 1,
@@ -568,6 +570,14 @@ func DefaultConfig() *Config {
}
}
func generateGatewayToken() string {
var buf [16]byte
if _, err := rand.Read(buf[:]); err != nil {
return ""
}
return "cg_" + hex.EncodeToString(buf[:])
}
func LoadConfig(path string) (*Config, error) {
cfg := DefaultConfig()

View File

@@ -2,6 +2,15 @@ package config
import "testing"
func TestDefaultConfigGeneratesGatewayToken(t *testing.T) {
t.Parallel()
cfg := DefaultConfig()
if cfg.Gateway.Token == "" {
t.Fatalf("expected default gateway token")
}
}
func TestValidateSubagentsAllowsKnownPeers(t *testing.T) {
t.Parallel()

View File

@@ -51,6 +51,7 @@ type SubagentTask struct {
type SubagentManager struct {
tasks map[string]*SubagentTask
cancelFuncs map[string]context.CancelFunc
recoverableTaskIDs []string
archiveAfterMinute int64
mu sync.RWMutex
provider providers.LLMProvider
@@ -99,9 +100,13 @@ func NewSubagentManager(provider providers.LLMProvider, workspace string, bus *b
if runStore != nil {
for _, task := range runStore.List() {
mgr.tasks[task.ID] = task
if task.Status == "running" {
mgr.recoverableTaskIDs = append(mgr.recoverableTaskIDs, task.ID)
}
}
mgr.nextID = runStore.NextIDSeed()
}
go mgr.resumeRecoveredTasks()
return mgr
}
@@ -534,6 +539,7 @@ func (sm *SubagentManager) SetRunFunc(f SubagentRunFunc) {
sm.mu.Lock()
defer sm.mu.Unlock()
sm.runFunc = f
go sm.resumeRecoveredTasks()
}
func (sm *SubagentManager) ProfileStore() *SubagentProfileStore {
@@ -542,6 +548,38 @@ func (sm *SubagentManager) ProfileStore() *SubagentProfileStore {
return sm.profileStore
}
func (sm *SubagentManager) resumeRecoveredTasks() {
if sm == nil {
return
}
sm.mu.Lock()
if sm.runFunc == nil && sm.provider == nil {
sm.mu.Unlock()
return
}
taskIDs := append([]string(nil), sm.recoverableTaskIDs...)
sm.recoverableTaskIDs = nil
toResume := make([]*SubagentTask, 0, len(taskIDs))
for _, taskID := range taskIDs {
task, ok := sm.tasks[taskID]
if !ok || task == nil || task.Status != "running" {
continue
}
task.Updated = time.Now().UnixMilli()
sm.persistTaskLocked(task, "recovered", "auto-resumed after restart")
toResume = append(toResume, task)
}
sm.mu.Unlock()
for _, task := range toResume {
taskCtx, cancel := context.WithCancel(context.Background())
sm.mu.Lock()
sm.cancelFuncs[task.ID] = cancel
sm.mu.Unlock()
go sm.runTask(taskCtx, task)
}
}
func (sm *SubagentManager) NextTaskSequence() int {
sm.mu.RLock()
defer sm.mu.RUnlock()

View File

@@ -205,6 +205,55 @@ func TestSubagentManagerRestoresPersistedRuns(t *testing.T) {
time.Sleep(100 * time.Millisecond)
}
func TestSubagentManagerAutoRecoversRunningTaskAfterRestart(t *testing.T) {
workspace := t.TempDir()
block := make(chan struct{})
manager := NewSubagentManager(nil, workspace, nil)
manager.SetRunFunc(func(ctx context.Context, task *SubagentTask) (string, error) {
<-block
return "should-not-complete-here", nil
})
_, err := manager.Spawn(context.Background(), SubagentSpawnOptions{
Task: "recover me",
AgentID: "coder",
OriginChannel: "cli",
OriginChatID: "direct",
})
if err != nil {
t.Fatalf("spawn failed: %v", err)
}
time.Sleep(80 * time.Millisecond)
recovered := make(chan string, 1)
reloaded := NewSubagentManager(nil, workspace, nil)
reloaded.SetRunFunc(func(ctx context.Context, task *SubagentTask) (string, error) {
recovered <- task.ID
return "recovered-ok", nil
})
select {
case taskID := <-recovered:
if taskID != "subagent-1" {
t.Fatalf("expected recovered task id subagent-1, got %s", taskID)
}
case <-time.After(2 * time.Second):
t.Fatalf("expected running task to auto-recover after restart")
}
got, ok := reloaded.GetTask("subagent-1")
if !ok {
t.Fatalf("expected recovered task to exist")
}
if got.Status != "completed" || got.Result != "recovered-ok" {
t.Fatalf("unexpected recovered task: %+v", got)
}
close(block)
_ = waitSubagentDone(t, manager, 4*time.Second)
time.Sleep(100 * time.Millisecond)
}
func TestSubagentManagerPersistsEvents(t *testing.T) {
workspace := t.TempDir()
manager := NewSubagentManager(nil, workspace, nil)