diff --git a/agent/agent.go b/agent/agent.go index ac2efd6d..8a3eb731 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -8,6 +8,7 @@ import ( "iter" "log/slog" "slices" + "sync/atomic" "github.com/google/uuid" "github.com/microsoft/agent-framework-go/message" @@ -166,7 +167,12 @@ type Agent struct { runOptions []Option logger *slog.Logger - historyProvider HistoryProvider + historyProvider HistoryProvider + // historyCleared records that a run promoted its session to service-managed + // history and cleared the configured provider globally (matching the .NET + // clear-on-conflict semantics). It is set instead of mutating historyProvider + // so a shared *Agent can be run concurrently without a data race. + historyCleared atomic.Bool hasConfiguredHistory bool // hasDefaultHistoryProvider is true when New synthesized the in-memory // history provider because Config.HistoryProvider was nil. The synthesized @@ -416,7 +422,7 @@ func (a *Agent) historyProviderForContinuationStore(session *Session, noSession } func (a *Agent) historyProviderForSession(session *Session, noSession bool) HistoryProvider { - if a.historyProvider == nil || session == nil { + if a.historyProvider == nil || session == nil || a.historyCleared.Load() { return nil } if !a.hasDefaultHistoryProvider { @@ -466,7 +472,7 @@ func (a *Agent) handleHistoryProviderConflict(ctx context.Context, provider Hist return false, errors.New("only Session.ServiceID or HistoryProvider may be used, but not both; the service returned an ID indicating service-managed history while the agent has a HistoryProvider configured") } if !a.keepHistoryOnConflict { - a.historyProvider = nil + a.historyCleared.Store(true) return false, nil } return true, nil diff --git a/agent/agent_test.go b/agent/agent_test.go index 98025060..30c7e5bc 100644 --- a/agent/agent_test.go +++ b/agent/agent_test.go @@ -7,6 +7,7 @@ import ( "errors" "iter" "slices" + "sync" "testing" "github.com/microsoft/agent-framework-go/agent" @@ -2135,6 +2136,49 @@ func TestAgent_Run_PipelineOrder_AgentHistoryContextProviderMiddlewareRun(t *tes } } +func TestAgent_Run_HistoryProvider_ConcurrentConflictClearIsRaceFree(t *testing.T) { + historyProvider := agent.NewHistoryProvider(agent.HistoryProviderConfig{ + SourceID: "history", + Provide: func(_ context.Context, _ agent.InvokingContext) ([]*message.Message, error) { + return nil, nil + }, + Store: func(context.Context, agent.InvokedContext) error { + return nil + }, + }) + // Every run promotes its own session to service-managed mid-run, which drives + // the clear-on-conflict path that used to mutate the shared Agent field. + runFn := func(_ context.Context, _ []*message.Message, options ...agent.Option) iter.Seq2[*agent.ResponseUpdate, error] { + session, _ := agent.GetOption(options, agent.WithSession) + session.SetServiceID("server-managed") + return func(yield func(*agent.ResponseUpdate, error) bool) { + yield(&agent.ResponseUpdate{Role: message.RoleAssistant, Contents: []message.Content{&message.TextContent{Text: "ok"}}}, nil) + } + } + a := agent.New(agent.ProviderConfig{Run: runFn}, agent.Config{ + ID: "test-agent", + Name: "test-agent", + HistoryProvider: historyProvider, + AllowHistoryProviderConflict: true, + SuppressHistoryProviderConflictWarning: true, + }) + + const goroutines = 64 + var wg sync.WaitGroup + wg.Add(goroutines) + for i := 0; i < goroutines; i++ { + go func() { + defer wg.Done() + // Each goroutine drives a shared *Agent with its own session, so the + // only shared state exercised is the agent's history-provider handling. + if _, err := a.RunText(t.Context(), "input", agent.WithSession(agenttest.CreateSession())).Collect(); err != nil { + t.Errorf("unexpected run error: %v", err) + } + }() + } + wg.Wait() +} + func toolNames(tools []tool.Tool) []string { names := make([]string, 0, len(tools)) for _, tool := range tools {