diff --git a/internal/tools/delegate.go b/internal/tools/delegate.go index ecc5c460..2bfca7f4 100644 --- a/internal/tools/delegate.go +++ b/internal/tools/delegate.go @@ -51,6 +51,13 @@ type DelegationTask struct { progressEnabled bool `json:"-"` // resolved from team settings or global default } +// originKey returns a composite key scoping this delegation to its origin conversation. +// Used for sibling counting and artifact accumulation so that delegations from +// different channels/chats are NOT treated as siblings of each other. +func (t *DelegationTask) originKey() string { + return t.SourceAgentID.String() + ":" + t.OriginChannel + ":" + t.OriginChatID +} + // DelegateOpts configures a single delegation call. type DelegateOpts struct { TargetAgentKey string diff --git a/internal/tools/delegate_async.go b/internal/tools/delegate_async.go index 48ed8106..2f4c4ca7 100644 --- a/internal/tools/delegate_async.go +++ b/internal/tools/delegate_async.go @@ -68,8 +68,11 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts) close(progressDone) duration := time.Since(startTime) - // Count sibling delegations still running (exclude self) - siblings := dm.ListActive(task.SourceAgentID) + // Count sibling delegations still running (exclude self). + // Scoped by origin (channel + chatID) so delegations from different + // conversations are NOT treated as siblings of each other. + oKey := task.originKey() + siblings := dm.ListActiveForOrigin(oKey) siblingCount := 0 for _, s := range siblings { if s.ID != task.ID { @@ -104,7 +107,7 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts) if task.TeamTaskID != uuid.Nil { arts.CompletedTaskIDs = []string{task.TeamTaskID.String()} } - dm.accumulateArtifacts(task.SourceAgentID, arts) + dm.accumulateArtifacts(oKey, arts) // Emit accumulated event so WS clients know this delegation finished // but results are being held until siblings complete. @@ -134,7 +137,7 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts) dm.progressSent.Delete(task.SourceAgentID.String() + ":" + task.OriginChatID) // Last completion: collect all accumulated artifacts + own result - artifacts := dm.collectArtifacts(task.SourceAgentID) + artifacts := dm.collectArtifacts(oKey) if result != nil { artifacts.Media = append(artifacts.Media, result.MediaPaths...) artifacts.Results = append(artifacts.Results, DelegateResultSummary{ diff --git a/internal/tools/delegate_state.go b/internal/tools/delegate_state.go index 9ce83b85..7942405e 100644 --- a/internal/tools/delegate_state.go +++ b/internal/tools/delegate_state.go @@ -93,6 +93,21 @@ func (dm *DelegateManager) ListActive(sourceAgentID uuid.UUID) []*DelegationTask return tasks } +// ListActiveForOrigin returns active delegations scoped to the same origin conversation. +// Only delegations matching the same source agent + channel + chatID are returned. +// This prevents cross-session sibling counting when multiple chats delegate concurrently. +func (dm *DelegateManager) ListActiveForOrigin(originKey string) []*DelegationTask { + var tasks []*DelegationTask + dm.active.Range(func(_, val any) bool { + t := val.(*DelegationTask) + if t.originKey() == originKey && t.Status == "running" { + tasks = append(tasks, t) + } + return true + }) + return tasks +} + // ActiveCountForLink counts running delegations for a specific source→target pair. func (dm *DelegateManager) ActiveCountForLink(sourceID, targetID uuid.UUID) int { count := 0 @@ -119,11 +134,11 @@ func (dm *DelegateManager) ActiveCountForTarget(targetID uuid.UUID) int { return count } -// accumulateArtifacts merges new artifacts into the pending set for a source agent. +// accumulateArtifacts merges new artifacts into the pending set for an origin conversation. // Called for intermediate delegation completions (when siblings are still running). -func (dm *DelegateManager) accumulateArtifacts(sourceAgentID uuid.UUID, arts *DelegateArtifacts) { - key := sourceAgentID.String() - existing, _ := dm.pendingArtifacts.Load(key) +// The originKey scopes accumulation to the same source agent + channel + chatID. +func (dm *DelegateManager) accumulateArtifacts(originKey string, arts *DelegateArtifacts) { + existing, _ := dm.pendingArtifacts.Load(originKey) var merged DelegateArtifacts if existing != nil { merged = *existing.(*DelegateArtifacts) @@ -131,14 +146,13 @@ func (dm *DelegateManager) accumulateArtifacts(sourceAgentID uuid.UUID, arts *De merged.Media = append(merged.Media, arts.Media...) merged.Results = append(merged.Results, arts.Results...) merged.CompletedTaskIDs = append(merged.CompletedTaskIDs, arts.CompletedTaskIDs...) - dm.pendingArtifacts.Store(key, &merged) + dm.pendingArtifacts.Store(originKey, &merged) } -// collectArtifacts retrieves and removes all accumulated artifacts for a source agent. +// collectArtifacts retrieves and removes all accumulated artifacts for an origin conversation. // Called when the last delegation completes (siblingCount == 0). -func (dm *DelegateManager) collectArtifacts(sourceAgentID uuid.UUID) *DelegateArtifacts { - key := sourceAgentID.String() - if pending, ok := dm.pendingArtifacts.LoadAndDelete(key); ok { +func (dm *DelegateManager) collectArtifacts(originKey string) *DelegateArtifacts { + if pending, ok := dm.pendingArtifacts.LoadAndDelete(originKey); ok { return pending.(*DelegateArtifacts) } return &DelegateArtifacts{}