fix(delegation): scope sibling counting by origin conversation

Delegations from different channels/chats sharing the same source agent
were incorrectly treated as siblings. This caused cross-session artifact
accumulation: media from a Telegram delegation could be routed to WS
and vice versa, with the first-completing delegation's announce being
suppressed entirely.

Add originKey() (sourceAgentID:channel:chatID) to DelegationTask and
use it for ListActiveForOrigin, accumulateArtifacts, collectArtifacts.
This commit is contained in:
viettranx committed 2026-03-07 23:32:35 +07:00
1 parent 27f2c0d80a
commit c636dde54e
3 files changed
+37 -13

No files matched your search

+7
View File
@@ -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
+7 -4
View File
@@ -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{
+23 -9
View File
@@ -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{}