From e1674b0249375fc6ec77475532c2aa5ee0e3cb92 Mon Sep 17 00:00:00 2001 From: yatul Date: Tue, 1 Sep 2026 00:43:55 +0400 Subject: [PATCH] fix(teams): dispatch team tasks created by detached child runs (#1528) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Team tasks are not dispatched inline: the turn's PendingTeamDispatch collects them and the post-turn drain assigns and dispatches them once the turn ends. That same tracker also owns the per-(team, chat) create lock taken by team_tasks list/search. Async delegations and async spawns keep the caller's context values (context.WithoutCancel) but run detached — after the caller's turn has already ended and drained. A team lead reached through delegate therefore adds every task it creates to a tracker nobody will drain again, and takes a create lock nobody will release. Two symptoms follow: - tasks stay pending forever and are never dispatched, so the team never starts work. The task ticker does not cover this: it never dispatches pending tasks, it only marks them stale after 2h and asks the lead to retry by hand. - the next team_tasks list/search for the same (team, chat) blocks on the never-released mutex for the remaining lifetime of the process. Give each detached child run its own tracker through the existing InjectTeamDispatch helper, drained when that run ends. The synchronous delegate/spawn paths are deliberately left alone: they execute inside the caller's turn, where the caller's tracker is the correct owner. Co-authored-by: Claude Opus 5 --- cmd/gateway.go | 3 + cmd/gateway_managed.go | 8 ++ internal/tools/delegate_tool.go | 10 ++ internal/tools/subagent.go | 6 + internal/tools/subagent_spawn.go | 6 + internal/tools/subagent_team_dispatch_test.go | 128 ++++++++++++++++++ 6 files changed, 161 insertions(+) create mode 100644 internal/tools/subagent_team_dispatch_test.go diff --git a/cmd/gateway.go b/cmd/gateway.go index c8c77bd2..8cf530e1 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -785,6 +785,9 @@ func runGateway() { chatMethods.SetPostTurnProcessor(postTurn) server.SetPostTurnProcessor(postTurn) // HTTP: /v1/chat/completions, /v1/responses wakeH.SetPostTurnProcessor(postTurn) // HTTP: /v1/agents/{id}/wake + if subagentMgr != nil { + subagentMgr.SetPostTurnProcessor(postTurn) // async spawns: detached from the parent turn + } } // Wire pairing event broadcasts to all WS clients. diff --git a/cmd/gateway_managed.go b/cmd/gateway_managed.go index 50f3914f..917cee5f 100644 --- a/cmd/gateway_managed.go +++ b/cmd/gateway_managed.go @@ -647,6 +647,14 @@ func wireExtras( if stores.Teams != nil && stores.Agents != nil { teamMgr := tools.NewTeamToolManager(stores.Teams, stores.Agents, msgBus, workspace) postTurn = teamMgr + // Async delegations run detached from the caller's turn, so they need their + // own post-turn dispatch. The delegate tool is registered above, before the + // team manager exists — wire it now that postTurn is available. + if delegateTool, ok := toolsReg.Get("delegate"); ok { + if dt, ok := delegateTool.(*tools.DelegateTool); ok { + dt.SetPostTurnProcessor(postTurn) + } + } var teamPolicy tools.TeamActionPolicy = tools.FullTeamPolicy{} if !edition.Current().TeamFullMode { teamPolicy = tools.LiteTeamPolicy{} diff --git a/internal/tools/delegate_tool.go b/internal/tools/delegate_tool.go index c55fa5ec..a094fc65 100644 --- a/internal/tools/delegate_tool.go +++ b/internal/tools/delegate_tool.go @@ -74,6 +74,7 @@ type DelegateTool struct { msgBus *bus.MessageBus // for async announce back to parent taskStore store.SubagentTaskStore // durable async completion ledger hookDispatcher hooks.Dispatcher // optional; nil-safe + postTurn PostTurnProcessor // optional; dispatches team tasks a detached delegatee creates admission *orchestration.ChildRunAdmission workspace string dataDir string @@ -102,6 +103,9 @@ func (t *DelegateTool) SetMsgBus(mb *bus.MessageBus) { t.msgBus = mb } // SetTaskStore configures the durable ledger used by async delegations. func (t *DelegateTool) SetTaskStore(s store.SubagentTaskStore) { t.taskStore = s } +// SetPostTurnProcessor wires post-turn team-task dispatch for async delegations. +func (t *DelegateTool) SetPostTurnProcessor(p PostTurnProcessor) { t.postTurn = p } + // SetHookDispatcher sets the hook dispatcher for SubagentStart/Stop events. func (t *DelegateTool) SetHookDispatcher(d hooks.Dispatcher) { t.hookDispatcher = d } @@ -460,12 +464,18 @@ func (t *DelegateTool) executeAsyncMode(ctx context.Context, job *delegateArtifa }() // Detach from parent cancellation but keep a bounded admitted callback. bgCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Minute) + // The delegatee runs after the caller's turn has ended, so the caller's + // post-turn drain has already fired by the time it calls team_tasks. Give + // this run its own tracker, drained when the delegatee finishes, so tasks it + // creates are dispatched and the team create lock it takes is released. + bgCtx, drainTeamDispatch := InjectTeamDispatch(bgCtx, t.postTurn) announceCtx := context.WithoutCancel(ctx) var dr DelegateResult var runErr error runStarted := make(chan struct{}) ticket, err := t.admission.Enqueue(bgCtx, delegateAdmissionConstraints(ctx, req), func(runCtx context.Context, lease *orchestration.ChildRunLease) { defer job.closeCallerRoot() + defer drainTeamDispatch() close(runStarted) runCtx = withDelegatedAgentExecution(runCtx, lease) dr, runErr = t.runArtifactExchange(runCtx, job) diff --git a/internal/tools/subagent.go b/internal/tools/subagent.go index bab1bb70..f9b238dc 100644 --- a/internal/tools/subagent.go +++ b/internal/tools/subagent.go @@ -111,6 +111,7 @@ type SubagentManager struct { announceQueue *AnnounceQueue // optional: batches announces with debounce taskStore store.SubagentTaskStore // optional: durable async completion ledger usageCaps *usagecaps.Service + postTurn PostTurnProcessor // optional: dispatches team tasks a detached subagent creates admission *orchestration.ChildRunAdmission sweeperOnce sync.Once sweeperStop chan struct{} @@ -188,6 +189,11 @@ func (sm *SubagentManager) SetTaskStore(s store.SubagentTaskStore) { sm.taskStore = s } +// SetPostTurnProcessor wires post-turn team-task dispatch for async spawns. +func (sm *SubagentManager) SetPostTurnProcessor(p PostTurnProcessor) { + sm.postTurn = p +} + func (sm *SubagentManager) SetUsageCapService(s *usagecaps.Service) { sm.usageCaps = s } diff --git a/internal/tools/subagent_spawn.go b/internal/tools/subagent_spawn.go index cdb86eba..26bc4f26 100644 --- a/internal/tools/subagent_spawn.go +++ b/internal/tools/subagent_spawn.go @@ -71,6 +71,11 @@ func (sm *SubagentManager) SpawnWithReceipt( // WithoutCancel preserves all context values (agent ID, workspace, trace info, etc.) // but parent Done() no longer propagates. Manual cancel via taskCancel() still works. detached := context.WithoutCancel(ctx) + // The subagent runs after the parent's turn has ended, so the parent's + // post-turn drain has already fired by the time it calls team_tasks. Give + // this run its own tracker, drained when the subagent finishes, so tasks it + // creates are dispatched and the team create lock it takes is released. + detached, drainTeamDispatch := InjectTeamDispatch(detached, sm.postTurn) taskCtx, taskCancel := context.WithCancel(detached) subTask.cancelFunc = taskCancel @@ -90,6 +95,7 @@ func (sm *SubagentManager) SpawnWithReceipt( ParentFanout: cfg.MaxChildrenPerAgent, Depth: admissionDepth, }, func(runCtx context.Context, lease *orchestration.ChildRunLease) { + defer drainTeamDispatch() sm.markTaskRunning(subTask) iterations = sm.executeTask(withSubagentExecution(runCtx, scope, subTask.ID, subTask.Depth, lease), subTask) lease.Release() diff --git a/internal/tools/subagent_team_dispatch_test.go b/internal/tools/subagent_team_dispatch_test.go new file mode 100644 index 00000000..007b5a0f --- /dev/null +++ b/internal/tools/subagent_team_dispatch_test.go @@ -0,0 +1,128 @@ +package tools + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/providers" +) + +// teamDispatchRecorder is a PostTurnProcessor that records what was dispatched. +type teamDispatchRecorder struct { + mu sync.Mutex + processed map[uuid.UUID][]uuid.UUID +} + +func (r *teamDispatchRecorder) ProcessPendingTasks(_ context.Context, teamID uuid.UUID, taskIDs []uuid.UUID) error { + r.mu.Lock() + defer r.mu.Unlock() + if r.processed == nil { + r.processed = make(map[uuid.UUID][]uuid.UUID) + } + r.processed[teamID] = append(r.processed[teamID], taskIDs...) + return nil +} + +func (r *teamDispatchRecorder) DispatchUnblockedTasks(context.Context, uuid.UUID) {} + +func (r *teamDispatchRecorder) dispatched(teamID uuid.UUID) []uuid.UUID { + r.mu.Lock() + defer r.mu.Unlock() + return append([]uuid.UUID(nil), r.processed[teamID]...) +} + +// teamTaskCreatingProvider stands in for an agent whose turn calls +// team_tasks(action="create"): it records the tracker it saw and registers a +// task on it, exactly like executeCreate does. +type teamTaskCreatingProvider struct { + teamID uuid.UUID + taskID uuid.UUID + + mu sync.Mutex + seen *PendingTeamDispatch + called bool +} + +func (p *teamTaskCreatingProvider) Name() string { return "team-task-creating" } +func (p *teamTaskCreatingProvider) DefaultModel() string { return "provider-default" } + +func (p *teamTaskCreatingProvider) Chat(ctx context.Context, _ providers.ChatRequest) (*providers.ChatResponse, error) { + ptd := PendingTeamDispatchFromCtx(ctx) + p.mu.Lock() + p.seen = ptd + p.called = true + p.mu.Unlock() + if ptd != nil { + ptd.Add(p.teamID, p.taskID) + } + return &providers.ChatResponse{Content: "done", FinishReason: "stop"}, nil +} + +func (p *teamTaskCreatingProvider) ChatStream(ctx context.Context, req providers.ChatRequest, _ func(providers.StreamChunk)) (*providers.ChatResponse, error) { + return p.Chat(ctx, req) +} + +func (p *teamTaskCreatingProvider) tracker() (*PendingTeamDispatch, bool) { + p.mu.Lock() + defer p.mu.Unlock() + return p.seen, p.called +} + +// An async spawn is detached from the parent's turn: by the time the child runs, +// the parent's post-turn drain has already fired. The child must therefore get +// its own tracker, drained when the child's own run ends — otherwise every team +// task it creates stays pending forever and is never dispatched. +func TestAsyncSpawnDispatchesTeamTasksCreatedAfterParentTurnEnded(t *testing.T) { + recorder := &teamDispatchRecorder{} + provider := &teamTaskCreatingProvider{teamID: uuid.New(), taskID: uuid.New()} + manager := NewSubagentManager(provider, nil, "manager-default", nil, NewRegistry, SubagentConfig{ + MaxConcurrent: 20, + MaxSpawnDepth: 1, + MaxChildrenPerAgent: 5, + }) + manager.SetPostTurnProcessor(recorder) + + parentCtx, parentDrain := InjectTeamDispatch(subagentTestContext("parent"), recorder) + parentTracker := PendingTeamDispatchFromCtx(parentCtx) + if parentTracker == nil { + t.Fatal("parent context has no pending team dispatch tracker") + } + // The parent's turn ends before the detached child ever starts. + parentDrain() + + if _, err := manager.Spawn( + parentCtx, "parent", 0, "create a team task", "child", "", "test", "chat", "direct", nil, + ); err != nil { + t.Fatalf("Spawn() error = %v", err) + } + + deadline := time.Now().Add(10 * time.Second) + for { + if len(recorder.dispatched(provider.teamID)) > 0 { + break + } + if time.Now().After(deadline) { + _, called := provider.tracker() + t.Fatalf("team task was never dispatched (child ran = %v)", called) + } + time.Sleep(10 * time.Millisecond) + } + + if got := recorder.dispatched(provider.teamID); len(got) != 1 || got[0] != provider.taskID { + t.Fatalf("dispatched tasks = %v, want [%v]", got, provider.taskID) + } + childTracker, _ := provider.tracker() + if childTracker == nil { + t.Fatal("child run had no pending team dispatch tracker") + } + if childTracker == parentTracker { + t.Fatal("child run reused the parent's tracker; its tasks would never be drained") + } + if leftover := parentTracker.Drain(); len(leftover) != 0 { + t.Fatalf("parent tracker collected child tasks: %v", leftover) + } +}