plugin/guardrail/approval: synchronize tool-call context before review

agent/graphagent/graph_agent.go

@@ -166,7 +166,7 @@ func (ga *GraphAgent) runWithoutBarrier(ctx context.Context, invocation *agent.I
 		innerChan,
 		suppressHiddenCompletion,
 		func(evt *event.Event) error {
-			if !terminalMessageFilter.Allows(evt) {
+			if !forwardTerminalMessage(ctx, invocation, terminalMessageFilter, evt) {
 				return nil
 			}
 			return event.EmitEvent(ctx, out, evt)
@@ -246,7 +246,7 @@ func (ga *GraphAgent) runWithBarrier(ctx context.Context, invocation *agent.Invo
 	terminalMessageFilter := newTerminalMessageFilter(invocation, ga.graph)
 	for evt := range innerChan {
 		fullRespEvent = recordTraceEvent(tracker, tokenUsage, fullRespEvent, evt)
-		if !terminalMessageFilter.Allows(evt) {
+		if !forwardTerminalMessage(ctx, invocation, terminalMessageFilter, evt) {
 			continue
 		}
 		if err := event.EmitEvent(ctx, out, evt); err != nil {

agent/graphagent/terminal_message_filter.go

@@ -10,12 +10,15 @@
 package graphagent
 
 import (
+	"context"
 	"encoding/json"
 	"strings"
 
 	"trpc.group/trpc-go/trpc-agent-go/agent"
 	"trpc.group/trpc-go/trpc-agent-go/event"
 	"trpc.group/trpc-go/trpc-agent-go/graph"
+	"trpc.group/trpc-go/trpc-agent-go/internal/eventvisibility"
+	"trpc.group/trpc-go/trpc-agent-go/internal/state/barrier"
 )
 
 type terminalMessageFilter struct {
@@ -280,3 +283,19 @@ func matchesFilterPrefix(filterKey string, prefix string) bool {
 	prefixWithDelimiter := prefix + agent.EventFilterKeyDelimiter
 	return strings.HasPrefix(filterKey, prefixWithDelimiter)
 }
+
+// forwardTerminalMessage preserves the runner's processing path for completed
+// hidden messages needed by approval reviewers. Partial messages and standalone
+// runs retain their existing filtering behavior.
+func forwardTerminalMessage(
+	ctx context.Context,
+	inv *agent.Invocation,
+	filter *terminalMessageFilter,
+	evt *event.Event,
+) bool {
+	if filter.Allows(evt) {
+		return true
+	}
+	return barrier.ToolCallPersistenceRequired(inv) && evt != nil && evt.Response != nil &&
+		!evt.IsPartial && eventvisibility.Hide(ctx, evt.ID)
+}

agent/invocation.go

@@ -57,6 +57,9 @@ const (
 	flusherStateKey = "__flush_session__"
 	// barrierStateKey is the invocation state key used by internal barrier flag.
 	barrierStateKey = "__graph_barrier__"
+	// toolCallPersistenceStateKey carries the barrier requirement for tool
+	// callbacks that read the current assistant response from the session.
+	toolCallPersistenceStateKey = "__tool_call_persistence__"
 	// appenderStateKey is the invocation state key used by internal appender
 	// attachment (see internal/state/appender).
 	appenderStateKey = "__append_event__"
@@ -1752,6 +1755,7 @@ func isCloneStateKey(key string) bool {
 	switch key {
 	case flusherStateKey,
 		barrierStateKey,
+		toolCallPersistenceStateKey,
 		appenderStateKey,
 		liveSessionStateKey,
 		streamHubStateKey,

examples/guardrail/approval/README.md

@@ -13,6 +13,31 @@ This example shows how to wire the top-level `guardrail` plugin with the built-i
   - `hostexec_kill_session` is denied by policy
 - Observe approval logs for approved and denied review decisions
 
+With a reviewer configured, the approval plugin makes runner-managed LLMAgents
+wait until the runner has processed the completed assistant/tool-call event
+before tool callbacks run. This includes serial and parallel tools and candidate
+attempts. When the session append succeeds, the reviewer can read the current
+assistant message and all tool calls in that response from the session. Append
+failures are logged and do not stop tool processing; a failed completion wait
+does stop tool processing. Existing transcript filters and size limits still
+apply.
+
+With `agent.WithGraphTerminalMessagesOnly(true)`, completed hidden graph messages
+remain available for session processing and approval review while staying out
+of the caller-visible stream. Event plugins may replace an event, including its
+ID, without discarding the original producer's completion notification.
+
+Partial responses remain streaming. Standalone agents without a completion
+consumer and runs without a configured reviewer retain their existing behavior.
+Approval still runs in `BeforeTool`; this does not change callback ordering or
+the `Reviewer` interface.
+
+For native Graph LLM nodes, enable
+`agent.WithGraphEmitFinalModelResponses(true)` to make completed model responses
+available in the session transcript. Graph's existing node barrier orders those
+events before a subsequent tools node; its default response emission policy is
+unchanged.
+
 ## Prerequisites
 
 - Go 1.21 or later

internal/eventvisibility/eventvisibility.go

@@ -0,0 +1,60 @@
+//
+// Tencent is pleased to support the open source community by making trpc-agent-go available.
+//
+// Copyright (C) 2026 Tencent.  All rights reserved.
+//
+// trpc-agent-go is licensed under the Apache License Version 2.0.
+//
+//
+
+// Package eventvisibility keeps runner-only event forwarding decisions out of
+// event payloads and persisted session data.
+package eventvisibility
+
+import (
+	"context"
+	"sync"
+)
+
+type contextKey struct{}
+type hiddenEvents struct {
+	mu  sync.Mutex
+	ids map[string]struct{}
+}
+
+// WithCapture creates a run-scoped registry for events that must be processed
+// internally without being forwarded to the caller. Derived contexts share it;
+// a new call creates an independent registry. Unconsumed entries live only as
+// long as the run's context.
+func WithCapture(ctx context.Context) context.Context {
+	return context.WithValue(ctx, contextKey{}, &hiddenEvents{ids: make(map[string]struct{})})
+}
+
+// Hide marks id for internal processing. It reports false if there is no
+// runner capture registry or id is empty, so standalone producers can retain
+// their normal filtering behavior. Concurrent calls are safe.
+func Hide(ctx context.Context, id string) bool {
+	state, _ := ctx.Value(contextKey{}).(*hiddenEvents)
+	if state == nil || id == "" {
+		return false
+	}
+	state.mu.Lock()
+	defer state.mu.Unlock()
+	state.ids[id] = struct{}{}
+	return true
+}
+
+// TakeHidden consumes the forwarding decision for the original event ID before
+// event plugins can replace it. Missing registries and IDs return false.
+// Concurrent calls are safe.
+func TakeHidden(ctx context.Context, id string) bool {
+	state, _ := ctx.Value(contextKey{}).(*hiddenEvents)
+	if state == nil {
+		return false
+	}
+	state.mu.Lock()
+	defer state.mu.Unlock()
+	_, hidden := state.ids[id]
+	delete(state.ids, id)
+	return hidden
+}

internal/eventvisibility/eventvisibility_test.go

@@ -0,0 +1,50 @@
+//
+// Tencent is pleased to support the open source community by making trpc-agent-go available.
+//
+// Copyright (C) 2026 Tencent.  All rights reserved.
+//
+// trpc-agent-go is licensed under the Apache License Version 2.0.
+//
+//
+
+package eventvisibility
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestHiddenEventsAreRunScoped(t *testing.T) {
+	parent := context.Background()
+	require.False(t, Hide(parent, "event"))
+	require.False(t, TakeHidden(parent, "event"))
+	ctx := WithCapture(parent)
+	require.False(t, Hide(ctx, ""))
+	child, cancel := context.WithCancel(ctx)
+	defer cancel()
+	require.True(t, Hide(child, "event"))
+	other := WithCapture(ctx)
+	require.False(t, TakeHidden(other, "event"), "nested runner runs must not consume their parent's decisions")
+	require.True(t, TakeHidden(ctx, "event"))
+	require.False(t, TakeHidden(ctx, "event"), "consumed decisions must be released")
+}
+
+func TestHiddenEventsAcrossConcurrentNodes(t *testing.T) {
+	ctx := WithCapture(context.Background())
+	var wg sync.WaitGroup
+	for i := 0; i < 20; i++ {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			Hide(ctx, fmt.Sprint(i))
+		}(i)
+	}
+	wg.Wait()
+	for i := 0; i < 20; i++ {
+		require.True(t, TakeHidden(ctx, fmt.Sprint(i)))
+	}
+}

internal/flow/llmflow/llmflow.go

@@ -37,6 +37,7 @@ import (
 	"trpc.group/trpc-go/trpc-agent-go/internal/modelcontext"
 	imodelrequest "trpc.group/trpc-go/trpc-agent-go/internal/modelrequest"
 	"trpc.group/trpc-go/trpc-agent-go/internal/responseusage"
+	"trpc.group/trpc-go/trpc-agent-go/internal/state/barrier"
 	"trpc.group/trpc-go/trpc-agent-go/internal/state/steer"
 	"trpc.group/trpc-go/trpc-agent-go/internal/state/summaryfork"
 	"trpc.group/trpc-go/trpc-agent-go/internal/state/summaryinject"
@@ -1002,11 +1003,16 @@ func (p *streamingResponseProcessor) process(
 		}
 		return true
 	}
-	llmResponseEvent := p.emitLLMResponse(
+	llmResponseEvent, err := p.emitLLMResponse(
 		eventInvocation,
 		response,
 		traceDetails,
 	)
+	if err != nil {
+		*p.err = err
+		responseErr = err
+		return false
+	}
 	p.lastEvent = llmResponseEvent
 	if p.tracker != nil {
 		p.tracker.SetLastEvent(p.lastEvent)
@@ -1191,13 +1197,24 @@ func (p *streamingResponseProcessor) emitLLMResponse(
 	eventInvocation *agent.Invocation,
 	response *model.Response,
 	traceDetails bool,
-) *event.Event {
+) (*event.Event, error) {
 	llmResponseEvent := p.flow.createLLMResponseEvent(
 		eventInvocation,
 		p.currentInvocation,
 		response,
 		p.llmRequest,
 	)
+	// Agent wrappers forward events asynchronously. BeforeTool reviewers need
+	// this response in the session before callbacks run or parallel tools clone
+	// the session. Wait for this event's acknowledgement, not a channel drain.
+	waitForPersistence := barrier.Enabled(eventInvocation) &&
+		barrier.ToolCallPersistenceRequired(eventInvocation) &&
+		!response.IsPartial && response.IsToolCallResponse()
+	if waitForPersistence {
+		llmResponseEvent.RequiresCompletion = true
+	}
+	// Snapshot before emission transfers ownership to downstream event plugins.
+	completionID := agent.GetAppendEventNoticeKey(llmResponseEvent.ID)
 	emitCtx := p.ctx
 	var emitSpan oteltrace.Span
 	emitStarted := false
@@ -1209,9 +1226,21 @@ func (p *streamingResponseProcessor) emitLLMResponse(
 			latencyResponseAttrs(response)...,
 		)
 	}
-	agent.EmitEvent(emitCtx, eventInvocation, p.eventChan, llmResponseEvent)
-	finishLatencySpan(emitSpan, emitStarted, nil)
-	return llmResponseEvent
+	err := agent.EmitEvent(emitCtx, eventInvocation, p.eventChan, llmResponseEvent)
+	if !waitForPersistence {
+		// Preserve emission behavior for runs without this synchronization need.
+		finishLatencySpan(emitSpan, emitStarted, nil)
+		return llmResponseEvent, nil
+	}
+	if err == nil {
+		err = eventInvocation.AddNoticeChannelAndWait(p.ctx,
+			completionID, flowEventWaitTimeout(p.ctx))
+	}
+	finishLatencySpan(emitSpan, emitStarted, err)
+	if err != nil {
+		return nil, fmt.Errorf("wait for tool-call event persistence: %w", err)
+	}
+	return llmResponseEvent, nil
 }
 
 func (p *streamingResponseProcessor) traceChat(