fix(agent): unblock stuck agent on /stop, auto-complete nil-result tasks, graceful shutdown (#527, #504, #39)

- loop.go: replace blocking range with select+ctx.Done() for parallel tool collection;
  document finalization trade-off on early cancel
- loop_cancel_test.go: tests for context cancellation and normal completion paths
- gateway_consumer_post_turn.go: always auto-complete team task in default case,
  fallback message when outcome.Result is nil
- gateway.go: call sched.Stop() with 5s drain before context cancel on shutdown
This commit is contained in:
viettranx committed 2026-03-30 14:12:00 +07:00
1 parent e1c52cbda8
commit ec060ad7e1
3 files changed
+115 -29

No files matched your search

+22 -18
View File
@@ -171,28 +171,32 @@ func resolveTeamTaskOutcome(
default:
// Agent turn ended without terminal action — auto-complete.
// Covers: Progressed, Commented, Claimed, or no flags at all.
result := ""
if outcome.Result != nil {
result := outcome.Result.Content
result = outcome.Result.Content
if len(outcome.Result.Deliverables) > 0 {
result = strings.Join(outcome.Result.Deliverables, "\n\n---\n\n")
}
if len(result) > 100_000 {
result = result[:100_000] + "\n[truncated]"
}
if err := deps.TeamStore.CompleteTask(ctx, meta.TaskID, meta.TeamID, result); err != nil {
slog.Warn("auto-complete: CompleteTask error", "task_id", meta.TaskID, "error", err)
} else {
bus.BroadcastForTenant(deps.MsgBus, protocol.EventTeamTaskCompleted, store.TenantIDFromContext(ctx), tools.BuildTaskEventPayload(
meta.TeamID.String(), meta.TaskID.String(),
store.TeamTaskStatusCompleted,
"agent", toAgent,
tools.WithTaskInfo(taskNumber, taskSubject),
tools.WithOwnerAgentKey(toAgent),
tools.WithChannel(taskChannel),
tools.WithChatID(taskChatID),
tools.WithTimestamp(now),
))
}
}
if result == "" {
result = "Agent run ended without explicit result"
}
if len(result) > 100_000 {
result = result[:100_000] + "\n[truncated]"
}
if err := deps.TeamStore.CompleteTask(ctx, meta.TaskID, meta.TeamID, result); err != nil {
slog.Warn("auto-complete: CompleteTask error", "task_id", meta.TaskID, "error", err)
} else {
bus.BroadcastForTenant(deps.MsgBus, protocol.EventTeamTaskCompleted, store.TenantIDFromContext(ctx), tools.BuildTaskEventPayload(
meta.TeamID.String(), meta.TaskID.String(),
store.TeamTaskStatusCompleted,
"agent", toAgent,
tools.WithTaskInfo(taskNumber, taskSubject),
tools.WithOwnerAgentKey(toAgent),
tools.WithChannel(taskChannel),
tools.WithChatID(taskChatID),
tools.WithTimestamp(now),
))
}
}
+25 -11
View File
@@ -20,6 +20,17 @@ import (
"github.com/nextlevelbuilder/goclaw/pkg/protocol"
)
// indexedResult holds the output of a single parallel tool execution, preserving
// the original call index so results can be sorted back into deterministic order.
type indexedResult struct {
idx int
tc providers.ToolCall
registryName string
result *tools.Result
argsJSON string
spanStart time.Time
}
func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) {
// Per-run emit wrapper: enriches every AgentEvent with delegation + routing context.
emitRun := func(event AgentEvent) {
@@ -551,14 +562,6 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error)
// Each goroutine performs lazy MCP activation + policy checking independently.
// Tool instances are immutable (context-based) so concurrent access is safe.
// Results are collected then processed sequentially for deterministic ordering.
type indexedResult struct {
idx int
tc providers.ToolCall
registryName string
result *tools.Result
argsJSON string
spanStart time.Time
}
// 1. Emit all tool.call events upfront (client sees all calls starting)
for _, tc := range resp.ToolCalls {
@@ -625,10 +628,21 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error)
// Close channel after all goroutines complete (run in separate goroutine to avoid deadlock)
go func() { wg.Wait(); close(resultCh) }()
// 3. Collect results
// 3. Collect results (respect context cancellation to allow /stop)
collected := make([]indexedResult, 0, len(resp.ToolCalls))
for r := range resultCh {
collected = append(collected, r)
collectLoop:
for range resp.ToolCalls {
select {
case r, ok := <-resultCh:
if !ok {
break collectLoop
}
collected = append(collected, r)
case <-ctx.Done():
// Trade-off: responsive /stop cancellation skips finalizeRun() cleanup
// (bootstrap cleanup, message flush). Stuck agent is worse than lost finalization.
return nil, ctx.Err()
}
}
// 4. Sort by original index → deterministic message ordering
+68
View File
@@ -0,0 +1,68 @@
package agent
import (
"context"
"testing"
)
func TestParallelToolCollection_ContextCancel(t *testing.T) {
// Simulate: 3 tool calls, but ctx cancelled before all arrive
ctx, cancel := context.WithCancel(context.Background())
resultCh := make(chan indexedResult, 3)
// Send only 1 result, then cancel
resultCh <- indexedResult{idx: 0}
cancel()
collected := make([]indexedResult, 0, 3)
var err error
collectLoop:
for i := 0; i < 3; i++ {
select {
case r, ok := <-resultCh:
if !ok {
break collectLoop
}
collected = append(collected, r)
case <-ctx.Done():
err = ctx.Err()
break collectLoop
}
}
if err == nil {
t.Fatal("expected context.Canceled error")
}
// We should have collected 1 result before cancellation (or 0 — timing dependent)
if len(collected) > 1 {
t.Errorf("expected at most 1 collected, got %d", len(collected))
}
}
func TestParallelToolCollection_AllComplete(t *testing.T) {
resultCh := make(chan indexedResult, 3)
resultCh <- indexedResult{idx: 0}
resultCh <- indexedResult{idx: 1}
resultCh <- indexedResult{idx: 2}
ctx := context.Background()
collected := make([]indexedResult, 0, 3)
collectLoop:
for i := 0; i < 3; i++ {
select {
case r, ok := <-resultCh:
if !ok {
break collectLoop
}
collected = append(collected, r)
case <-ctx.Done():
t.Fatal("context should not be cancelled")
}
}
if len(collected) != 3 {
t.Fatalf("expected 3, got %d", len(collected))
}
}