fix(teams): dispatch team tasks created by detached child runs (#1528)

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 <noreply@anthropic.com>

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.

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{}

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)

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
 }

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()

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)
+	}
+}