Merge pull request #1489 from itsddvn/fix/delegation-artifact-media-dedup

fix(collaboration): protect artifact and media delivery lifecycle

internal/agent/loop_context.go

@@ -403,8 +403,8 @@ func (l *Loop) injectContext(ctx context.Context, req *RunRequest) (contextSetup
 	// Inject agent key into context for tool-level resolution (multiple agents share tool registry)
 	ctx = tools.WithToolAgentKey(ctx, l.id)
 
-	// Inject delivered media tracker so write_file and message tool can coordinate:
-	// write_file(deliver=true) marks paths, message self-send guard checks before allowing.
+	// Inject delivered media tracker so automatic output collection and direct
+	// media sends can coordinate within this run.
 	ctx = tools.WithDeliveredMedia(ctx, tools.NewDeliveredMedia())
 
 	// Security: truncate oversized user messages gracefully (feed truncation notice into LLM)

internal/agent/loop_media_delivery_dedup_test.go

@@ -0,0 +1,122 @@
+package agent
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/nextlevelbuilder/goclaw/internal/bus"
+	"github.com/nextlevelbuilder/goclaw/internal/providers"
+	"github.com/nextlevelbuilder/goclaw/internal/tools"
+)
+
+func TestProcessToolResultBlocksMessageSelfSendForQueuedMedia(t *testing.T) {
+	workspace := t.TempDir()
+	workspace, err := filepath.EvalSymlinks(workspace)
+	if err != nil {
+		t.Fatal(err)
+	}
+	mediaPath := filepath.Join(workspace, "generated-image.png")
+	if err := os.WriteFile(mediaPath, []byte("image"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+
+	delivered := tools.NewDeliveredMedia()
+	ctx := context.Background()
+	ctx = tools.WithToolWorkspace(ctx, workspace)
+	ctx = tools.WithToolChannel(ctx, "lark-test")
+	ctx = tools.WithToolChatID(ctx, "chat-1")
+	ctx = tools.WithDeliveredMedia(ctx, delivered)
+
+	loop := &Loop{id: "test-agent"}
+	state := &runState{}
+	loop.processToolResult(
+		ctx,
+		state,
+		&RunRequest{RunID: "run-1"},
+		func(AgentEvent) {},
+		providers.ToolCall{ID: "create-1", Name: "create_image"},
+		"create_image",
+		&tools.Result{
+			ForLLM: "MEDIA:" + mediaPath,
+			Media: []bus.MediaFile{{
+				Path:     mediaPath,
+				MimeType: "image/png",
+			}},
+		},
+		false,
+	)
+
+	if len(state.mediaResults) != 1 {
+		t.Fatalf("queued media = %d, want 1", len(state.mediaResults))
+	}
+
+	messageTool := tools.NewMessageTool(workspace, true)
+	messageTool.SetMessageBus(bus.New())
+	result := messageTool.Execute(ctx, map[string]any{
+		"action":  "send",
+		"channel": "lark-test",
+		"target":  "chat-1",
+		"message": "MEDIA:" + mediaPath,
+	})
+	if !result.IsError {
+		t.Fatal("message self-send succeeded for media already queued by the agent loop")
+	}
+	if !strings.Contains(result.ForLLM, "already queued for automatic delivery") {
+		t.Fatalf("message self-send blocked for the wrong reason: %q", result.ForLLM)
+	}
+
+	otherPath := filepath.Join(workspace, "manual-attachment.txt")
+	if err := os.WriteFile(otherPath, []byte("attachment"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	result = messageTool.Execute(ctx, map[string]any{
+		"action":  "send",
+		"channel": "lark-test",
+		"target":  "chat-1",
+		"message": "Attachments:\nMEDIA:" + mediaPath + "\nMEDIA:" + otherPath,
+	})
+	if !result.IsError {
+		t.Fatal("mixed message self-send included media already queued by the agent loop")
+	}
+	if !strings.Contains(result.ForLLM, "already queued for automatic delivery") {
+		t.Fatalf("mixed message self-send blocked for the wrong reason: %q", result.ForLLM)
+	}
+}
+
+func TestProcessToolResultMarksLegacyMediaAsQueued(t *testing.T) {
+	workspace := t.TempDir()
+	workspace, err := filepath.EvalSymlinks(workspace)
+	if err != nil {
+		t.Fatal(err)
+	}
+	mediaPath := filepath.Join(workspace, "legacy-output.pdf")
+	if err := os.WriteFile(mediaPath, []byte("pdf"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+
+	delivered := tools.NewDeliveredMedia()
+	ctx := tools.WithToolWorkspace(context.Background(), workspace)
+	ctx = tools.WithDeliveredMedia(ctx, delivered)
+
+	state := &runState{}
+	(&Loop{id: "test-agent"}).processToolResult(
+		ctx,
+		state,
+		&RunRequest{RunID: "run-legacy"},
+		func(AgentEvent) {},
+		providers.ToolCall{ID: "exec-1", Name: "exec"},
+		"exec",
+		&tools.Result{ForLLM: "MEDIA:" + mediaPath},
+		false,
+	)
+
+	if len(state.mediaResults) != 1 {
+		t.Fatalf("queued legacy media = %d, want 1", len(state.mediaResults))
+	}
+	if !delivered.IsDelivered(mediaPath) {
+		t.Fatalf("legacy media %q was not marked for automatic delivery", mediaPath)
+	}
+}

internal/agent/loop_tools.go

@@ -118,6 +118,12 @@ func (l *Loop) processToolResult(
 				mr.Prompt = result.MediaPrompts[i]
 			}
 			rs.mediaResults = append(rs.mediaResults, mr)
+			// The file is now queued for the run's automatic outbound delivery.
+			// Mark it so a later message(MEDIA:same_path) self-send cannot publish
+			// the same attachment before the final response is dispatched.
+			if dm := tools.DeliveredMediaFromCtx(ctx); dm != nil {
+				dm.Mark(cleaned)
+			}
 		}
 	} else if mr := parseMediaResult(result.ForLLM); mr != nil {
 		// Security (egress boundary): a tool's MEDIA:<path> output is taken
@@ -129,6 +135,9 @@ func (l *Loop) processToolResult(
 		if cleaned, ok := confineToAnyRoot(mr.Path, mediaRoots); ok {
 			mr.Path = cleaned
 			rs.mediaResults = append(rs.mediaResults, *mr)
+			if dm := tools.DeliveredMediaFromCtx(ctx); dm != nil {
+				dm.Mark(cleaned)
+			}
 		} else {
 			slog.Warn("security.media_path_rejected",
 				"agent", l.id, "tool", tc.Name, "path", mr.Path,

internal/tools/context_keys.go

@@ -774,12 +774,12 @@ func WorkstationIDFromCtx(ctx context.Context) string {
 	return v
 }
 
-// --- Delivered media tracker (write_file → message self-send dedup) ---
+// --- Delivered media tracker (automatic delivery → direct-send dedup) ---
 
 const ctxDeliveredMedia toolContextKey = "tool_delivered_media"
 
-// DeliveredMedia tracks file paths already queued for auto-delivery by write_file.
-// Injected once per run via WithDeliveredMedia; write_file marks paths, message reads them.
+// DeliveredMedia tracks file paths already queued or sent during an agent run.
+// Automatic media producers mark paths, and direct-send tools consult the tracker.
 // Thread-safe: tools may execute in parallel goroutines.
 type DeliveredMedia struct {
 	mu    sync.Mutex

internal/tools/delegate_tool.go

@@ -86,6 +86,9 @@ type DelegateTool struct {
 	sweeperStop    chan struct{}
 	sweeperDone    chan struct{}
 
+	activeArtifactMu sync.RWMutex
+	activeArtifacts  map[string]uint32
+
 	completionMu     sync.Mutex
 	completionClosed bool
 	completionWG     sync.WaitGroup
@@ -190,15 +193,16 @@ func NewDelegateToolWithAdmission(
 		admission = orchestration.NewChildRunAdmission(32, 128)
 	}
 	return &DelegateTool{
-		links:       links,
-		agents:      agents,
-		eventBus:    eb,
-		runFn:       runFn,
-		admission:   admission,
-		retained:    make(map[string]retainedDelegationArtifact),
-		sweeperStop: make(chan struct{}),
-		sweeperDone: make(chan struct{}),
-		closeDone:   make(chan struct{}),
+		links:           links,
+		agents:          agents,
+		eventBus:        eb,
+		runFn:           runFn,
+		admission:       admission,
+		retained:        make(map[string]retainedDelegationArtifact),
+		sweeperStop:     make(chan struct{}),
+		sweeperDone:     make(chan struct{}),
+		activeArtifacts: make(map[string]uint32),
+		closeDone:       make(chan struct{}),
 	}
 }
 
@@ -717,6 +721,9 @@ func collectDelegateInputs(callerWorkspace string, explicitInputs, currentMedia
 }
 
 func (t *DelegateTool) runArtifactExchange(ctx context.Context, job *delegateArtifactJob) (dr DelegateResult, returnErr error) {
+	t.beginDelegationArtifactExchange(job.tenantWorkspace, job.delegationID)
+	defer t.endDelegationArtifactExchange(job.tenantWorkspace, job.delegationID)
+
 	exchange, err := NewDelegationArtifactExchange(
 		job.tenantWorkspace,
 		job.tenantID,

internal/tools/delegate_tool_hooks_test.go

@@ -1800,6 +1800,178 @@ func TestDelegateTool_RecoversStaleArtifactLifecycleStates(t *testing.T) {
 	}
 }
 
+func TestDelegateTool_PeriodicArtifactMaintenancePreservesActiveExchange(t *testing.T) {
+	tenantWorkspace := t.TempDir()
+	callerWorkspace := filepath.Join(tenantWorkspace, "agents", "caller")
+	if err := os.MkdirAll(callerWorkspace, 0o750); err != nil {
+		t.Fatal(err)
+	}
+	callerRoot, err := OpenDelegationArtifactRoot(callerWorkspace)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer callerRoot.Close()
+
+	delegationID := uuid.New()
+	runStarted := make(chan struct{})
+	resumeRun := make(chan struct{})
+	var tool *DelegateTool
+	tool = NewDelegateTool(
+		noopAgentLink{},
+		noopAgentCRUD{},
+		nil,
+		func(_ context.Context, req DelegateRequest) (DelegateResult, error) {
+			close(runStarted)
+			<-resumeRun
+			if !tool.delegationArtifactExchangeActive(tenantWorkspace, delegationID) {
+				return DelegateResult{}, errors.New("delegation artifact exchange became inactive during run")
+			}
+			if err := os.WriteFile(
+				filepath.Join(req.DelegateOutputsPath, "result.txt"),
+				[]byte("active result"),
+				0o600,
+			); err != nil {
+				return DelegateResult{}, err
+			}
+			return DelegateResult{Content: "done"}, nil
+		},
+	)
+	tool.workspace = tenantWorkspace
+	defer tool.Close()
+
+	job := &delegateArtifactJob{
+		req:             DelegateRequest{DelegationID: delegationID.String()},
+		callerRoot:      callerRoot,
+		callerWorkspace: callerWorkspace,
+		tenantWorkspace: tenantWorkspace,
+		tenantID:        store.MasterTenantID,
+		delegationID:    delegationID,
+	}
+	job.callerLocation = tool.resolveDelegationCallerLocation(job)
+
+	runDone := make(chan error, 1)
+	go func() {
+		_, err := tool.runArtifactExchange(context.Background(), job)
+		runDone <- err
+	}()
+	select {
+	case <-runStarted:
+	case <-time.After(2 * time.Second):
+		close(resumeRun)
+		t.Fatal("delegated run did not start")
+	}
+	tool.addRetainedDelegationArtifact(retainedDelegationArtifact{
+		tenantWorkspace: tenantWorkspace,
+		tenantID:        store.MasterTenantID,
+		delegationID:    uuid.New(),
+		retainUntil:     time.Now().Add(time.Hour),
+	})
+	tool.maintainDelegationArtifacts(time.Now())
+
+	lifecyclePath := filepath.Join(
+		tenantWorkspace,
+		"collaboration",
+		"delegations",
+		delegationID.String(),
+		delegationArtifactLifecycleFile,
+	)
+	stateBytes, stateErr := os.ReadFile(lifecyclePath)
+	var state delegationArtifactLifecycleState
+	if stateErr == nil {
+		stateErr = json.Unmarshal(stateBytes, &state)
+	}
+	close(resumeRun)
+	runErr := <-runDone
+	if stateErr != nil {
+		t.Fatal(stateErr)
+	}
+	if state.Status != artifactLifecycleRunning {
+		t.Fatalf("active lifecycle status = %q, want %q", state.Status, artifactLifecycleRunning)
+	}
+	if runErr != nil {
+		t.Fatalf("active exchange run after maintenance: %v", runErr)
+	}
+	if tool.delegationArtifactExchangeActive(tenantWorkspace, delegationID) {
+		t.Fatal("completed delegation artifact exchange remained active")
+	}
+	publishedOutput := filepath.Join(
+		callerWorkspace,
+		".delegations",
+		delegationID.String(),
+		"outputs",
+		"result.txt",
+	)
+	if got, err := os.ReadFile(publishedOutput); err != nil || string(got) != "active result" {
+		t.Fatalf("published output = %q, %v", got, err)
+	}
+}
+
+func TestDelegateTool_PeriodicArtifactMaintenanceRecoversUnregisteredExchange(t *testing.T) {
+	tenantWorkspace := t.TempDir()
+	callerWorkspace := filepath.Join(tenantWorkspace, "agents", "caller")
+	if err := os.MkdirAll(callerWorkspace, 0o750); err != nil {
+		t.Fatal(err)
+	}
+	callerRoot, err := OpenDelegationArtifactRoot(callerWorkspace)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer callerRoot.Close()
+
+	delegationID := uuid.New()
+	exchange, err := NewDelegationArtifactExchange(
+		tenantWorkspace,
+		store.MasterTenantID,
+		delegationID,
+		DelegationArtifactLimits{},
+		time.Minute,
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer exchange.Close()
+
+	tool := NewDelegateTool(
+		noopAgentLink{},
+		noopAgentCRUD{},
+		nil,
+		func(_ context.Context, _ DelegateRequest) (DelegateResult, error) {
+			return DelegateResult{}, nil
+		},
+	)
+	tool.workspace = tenantWorkspace
+	defer tool.Close()
+
+	job := &delegateArtifactJob{
+		req:             DelegateRequest{DelegationID: delegationID.String()},
+		callerRoot:      callerRoot,
+		callerWorkspace: callerWorkspace,
+		tenantWorkspace: tenantWorkspace,
+		tenantID:        store.MasterTenantID,
+		delegationID:    delegationID,
+	}
+	job.callerLocation = tool.resolveDelegationCallerLocation(job)
+	if err := tool.updateActiveDelegationLifecycle(exchange, job); err != nil {
+		t.Fatalf("update active lifecycle: %v", err)
+	}
+	if err := tool.markDelegationRunning(exchange, job, time.Now()); err != nil {
+		t.Fatalf("mark running: %v", err)
+	}
+
+	tool.maintainDelegationArtifacts(time.Now())
+
+	state, err := readDelegationArtifactLifecycleState(exchange.root, delegationID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if state.Status != artifactLifecycleFailed || state.ReasonCode != "artifact_recovered_stale" {
+		t.Fatalf("unregistered lifecycle = %#v, want recovered stale failure", state)
+	}
+	if _, ok := tool.retained[retainedDelegationArtifactKey(tenantWorkspace, delegationID)]; !ok {
+		t.Fatalf("unregistered exchange was not added to retained registry: %#v", tool.retained)
+	}
+}
+
 func TestDelegateTool_RecoveryPromotesDurablePublishingState(t *testing.T) {
 	tenantWorkspace := t.TempDir()
 	callerWorkspace := filepath.Join(tenantWorkspace, "agents", "caller")

internal/tools/delegation_artifact_janitor.go

@@ -263,6 +263,43 @@ func retainedDelegationArtifactKey(tenantWorkspace string, delegationID uuid.UUI
 	return tenantWorkspace + "\x00" + delegationID.String()
 }
 
+func (t *DelegateTool) beginDelegationArtifactExchange(
+	tenantWorkspace string,
+	delegationID uuid.UUID,
+) {
+	key := retainedDelegationArtifactKey(tenantWorkspace, delegationID)
+	t.activeArtifactMu.Lock()
+	defer t.activeArtifactMu.Unlock()
+	if t.activeArtifacts == nil {
+		t.activeArtifacts = make(map[string]uint32)
+	}
+	t.activeArtifacts[key]++
+}
+
+func (t *DelegateTool) endDelegationArtifactExchange(
+	tenantWorkspace string,
+	delegationID uuid.UUID,
+) {
+	key := retainedDelegationArtifactKey(tenantWorkspace, delegationID)
+	t.activeArtifactMu.Lock()
+	defer t.activeArtifactMu.Unlock()
+	if t.activeArtifacts[key] <= 1 {
+		delete(t.activeArtifacts, key)
+		return
+	}
+	t.activeArtifacts[key]--
+}
+
+func (t *DelegateTool) delegationArtifactExchangeActive(
+	tenantWorkspace string,
+	delegationID uuid.UUID,
+) bool {
+	key := retainedDelegationArtifactKey(tenantWorkspace, delegationID)
+	t.activeArtifactMu.RLock()
+	defer t.activeArtifactMu.RUnlock()
+	return t.activeArtifacts[key] > 0
+}
+
 func (t *DelegateTool) runDelegationArtifactSweeper() {
 	ticker := time.NewTicker(delegationArtifactSweepInterval)
 	defer func() {
@@ -272,14 +309,18 @@ func (t *DelegateTool) runDelegationArtifactSweeper() {
 	for {
 		select {
 		case now := <-ticker.C:
-			t.recoverRetainedDelegationExchanges()
-			t.sweepRetainedDelegationExchanges(now)
+			t.maintainDelegationArtifacts(now)
 		case <-t.sweeperStop:
 			return
 		}
 	}
 }
 
+func (t *DelegateTool) maintainDelegationArtifacts(now time.Time) {
+	t.recoverRetainedDelegationExchanges()
+	t.sweepRetainedDelegationExchanges(now)
+}
+
 func (t *DelegateTool) recoverRetainedDelegationExchanges() {
 	for _, tenantWorkspace := range t.delegationTenantWorkspaces() {
 		if err := t.recoverTenantDelegationExchanges(tenantWorkspace); err != nil {
@@ -350,6 +391,9 @@ func (t *DelegateTool) recoverTenantDelegationExchanges(tenantWorkspace string)
 			}
 			continue
 		}
+		if t.delegationArtifactExchangeActive(tenantWorkspace, delegationID) {
+			continue
+		}
 		exchangeRoot, err := delegationsRoot.openSubroot(name)
 		if err != nil {
 			slog.Warn("delegate.artifact_exchange_recovery_open_failed",

internal/tools/message.go

@@ -141,9 +141,8 @@ func (t *MessageTool) Execute(ctx context.Context, args map[string]any) *Result
 	// Self-send guard: prevent agent from sending to its own chat via message tool.
 	// Text self-sends are always blocked (response goes through normal outbound).
 	// MEDIA self-sends are allowed ONLY when the file was NOT already queued for
-	// delivery (i.e. write_file was called with deliver=false). This prevents both
-	// duplicate delivery (deliver=true then message MEDIA:) and runaway retry loops
-	// (deliver=false then message MEDIA: blocked unconditionally).
+	// automatic delivery. This prevents both duplicate delivery and runaway retry
+	// loops while preserving explicit delivery of files that are not auto-queued.
 	ctxChannel := ToolChannelFromCtx(ctx)
 	ctxChatID := ToolChatIDFromCtx(ctx)
 	isSelfSend := ctxChannel != "" && ctxChatID != "" && channel == ctxChannel && target == ctxChatID
@@ -152,21 +151,22 @@ func (t *MessageTool) Execute(ctx context.Context, args map[string]any) *Result
 		if !isMediaSend {
 			return ErrorResult("You are already responding to this chat. Your response text will be delivered automatically. Do not use the message tool to send text to your own chat — just include the content in your response text. To deliver files, use write_file with deliver=true instead.")
 		}
-		// MEDIA self-send: block if ALL referenced files are already queued for delivery.
-		// Extracts paths from both standalone "MEDIA:path" and embedded multi-line messages.
+		// MEDIA self-send: block if any referenced file is already queued for
+		// delivery. Sending a mixed set would still duplicate the queued files;
+		// the model can retry with only the undelivered references.
 		if dm := DeliveredMediaFromCtx(ctx); dm != nil {
 			mediaRefs := embeddedMediaPattern.FindAllString(message, -1)
-			allDelivered := len(mediaRefs) > 0
+			hasDelivered := false
 			for _, raw := range mediaRefs {
 				if filePath, ok := t.resolveMediaPath(ctx, raw); ok {
-					if !dm.IsDelivered(filePath) {
-						allDelivered = false
+					if dm.IsDelivered(filePath) {
+						hasDelivered = true
 						break
 					}
 				}
 			}
-			if allDelivered {
-				return ErrorResult("This file is already queued for automatic delivery via write_file(deliver=true). Do not send it again. To deliver files that were written with deliver=false, use write_file again with deliver=true, or use message(MEDIA:path) which is allowed for undelivered files.")
+			if hasDelivered {
+				return ErrorResult("One or more files are already queued for automatic delivery. Do not send them again with message(MEDIA:path). Retry with only files that are not queued for automatic delivery.")
 			}
 		}
 	}