From b6fb91b1639a4e66aa214919221cedb4d3ada111 Mon Sep 17 00:00:00 2001 From: ntduc Date: Fri, 31 Jul 2026 08:39:50 +0700 Subject: [PATCH 1/2] fix(collaboration): preserve active artifact exchanges --- internal/tools/delegate_tool.go | 25 ++- internal/tools/delegate_tool_hooks_test.go | 172 ++++++++++++++++++ internal/tools/delegation_artifact_janitor.go | 48 ++++- 3 files changed, 234 insertions(+), 11 deletions(-) diff --git a/internal/tools/delegate_tool.go b/internal/tools/delegate_tool.go index 96e86cea..c55fa5ec 100644 --- a/internal/tools/delegate_tool.go +++ b/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, diff --git a/internal/tools/delegate_tool_hooks_test.go b/internal/tools/delegate_tool_hooks_test.go index 72f5128a..f228c2f1 100644 --- a/internal/tools/delegate_tool_hooks_test.go +++ b/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") diff --git a/internal/tools/delegation_artifact_janitor.go b/internal/tools/delegation_artifact_janitor.go index 7184ec7f..9943dba3 100644 --- a/internal/tools/delegation_artifact_janitor.go +++ b/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", From 4426f7614020b1696ad433e378580478bfde088a Mon Sep 17 00:00:00 2001 From: ntduc Date: Fri, 31 Jul 2026 08:40:02 +0700 Subject: [PATCH 2/2] fix(agent): deduplicate automatic media delivery --- internal/agent/loop_context.go | 4 +- .../agent/loop_media_delivery_dedup_test.go | 122 ++++++++++++++++++ internal/agent/loop_tools.go | 9 ++ internal/tools/context_keys.go | 6 +- internal/tools/message.go | 20 +-- 5 files changed, 146 insertions(+), 15 deletions(-) create mode 100644 internal/agent/loop_media_delivery_dedup_test.go diff --git a/internal/agent/loop_context.go b/internal/agent/loop_context.go index 53df6375..d0dbab65 100644 --- a/internal/agent/loop_context.go +++ b/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) diff --git a/internal/agent/loop_media_delivery_dedup_test.go b/internal/agent/loop_media_delivery_dedup_test.go new file mode 100644 index 00000000..73ce7412 --- /dev/null +++ b/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) + } +} diff --git a/internal/agent/loop_tools.go b/internal/agent/loop_tools.go index 3f04fa71..7639051d 100644 --- a/internal/agent/loop_tools.go +++ b/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: 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, diff --git a/internal/tools/context_keys.go b/internal/tools/context_keys.go index 65f21e33..82364d10 100644 --- a/internal/tools/context_keys.go +++ b/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 diff --git a/internal/tools/message.go b/internal/tools/message.go index 6687fcd9..fc294417 100644 --- a/internal/tools/message.go +++ b/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.") } } }