diff --git a/workflow/inproc/observability_test.go b/workflow/inproc/observability_test.go index 481fc00e..17edb7a9 100644 --- a/workflow/inproc/observability_test.go +++ b/workflow/inproc/observability_test.go @@ -4,15 +4,26 @@ package inproc_test import ( "context" + "errors" + "fmt" "strings" "testing" "github.com/microsoft/agent-framework-go/workflow" "github.com/microsoft/agent-framework-go/workflow/inproc" + internalobservability "github.com/microsoft/agent-framework-go/workflow/internal/observability" "github.com/microsoft/agent-framework-go/workflow/internal/workflowtest" "github.com/microsoft/agent-framework-go/workflow/observability" ) +type unserializablePayload struct { + Value string +} + +func (unserializablePayload) MarshalJSON() ([]byte, error) { + return nil, errors.New("marshal failed") +} + func TestObservability_CreatesWorkflowEndToEndSpans(t *testing.T) { tracer := workflowtest.NewRecordingTracer() wf := newTelemetryWorkflow(t, tracer, workflow.TelemetryOptions{}) @@ -194,6 +205,49 @@ func TestObservability_SensitiveDataControlsMessageContent(t *testing.T) { } } +func TestObservability_UnserializableSensitiveDataDoesNotFailWorkflow(t *testing.T) { + tracer := workflowtest.NewRecordingTracer() + var received []unserializablePayload + start := workflow.NewExecutor("start", func(input string) unserializablePayload { + return unserializablePayload{Value: strings.ToUpper(input)} + }).Bind() + sink := workflow.NewExecutor("sink", func(msg unserializablePayload) string { + received = append(received, msg) + return "done" + }).Bind() + + wf, err := workflow.NewBuilder(start). + AddEdge(start, sink). + WithOutputFrom(sink). + WithTelemetry(tracer, workflow.TelemetryOptions{EnableSensitiveData: true}). + Build() + if err != nil { + t.Fatalf("Build: %v", err) + } + + run, err := inproc.Default.Run(context.Background(), wf, "hello") + if err != nil { + t.Fatalf("Run: %v", err) + } + outputs := collectOutputValues(run.OutgoingEvents()) + if err := run.Close(context.Background()); err != nil { + t.Fatalf("Close: %v", err) + } + + if len(outputs) != 1 || outputs[0] != "done" { + t.Fatalf("outputs = %#v, want []string{\"done\"}", outputs) + } + if len(received) != 1 || received[0].Value != "HELLO" { + t.Fatalf("received = %#v, want one delivered payload with value HELLO", received) + } + + wantFallback := fmt.Sprintf("[Unserializable: %T]", unserializablePayload{}) + messageSpan := workflowtest.FindSpanWithPrefix(t, tracer.Spans(), "message.send") + messageSpan.RequireAttributeValue(t, internalobservability.TagMessageContent, wantFallback) + sinkSpan := workflowtest.FindSpanWithPrefix(t, tracer.Spans(), "executor.process sink") + sinkSpan.RequireAttributeValue(t, internalobservability.TagExecutorInput, wantFallback) +} + func TestObservability_RunSpansAreEnded(t *testing.T) { tests := []struct { name string diff --git a/workflow/internal/observability/observability_test.go b/workflow/internal/observability/observability_test.go index 68898f0f..b0534295 100644 --- a/workflow/internal/observability/observability_test.go +++ b/workflow/internal/observability/observability_test.go @@ -13,6 +13,12 @@ import ( workflowobservability "github.com/microsoft/agent-framework-go/workflow/observability" ) +type unserializableValue struct{} + +func (unserializableValue) MarshalJSON() ([]byte, error) { + return nil, errors.New("marshal failed") +} + func attributeValue(t *testing.T, attrs []workflowobservability.Attribute, key string) string { t.Helper() for _, attr := range attrs { @@ -112,3 +118,38 @@ func TestStartExecutorProcessEmitsExecutorType(t *testing.T) { } } } + +func TestSerializedAttributeUsesFallbackForMarshalErrors(t *testing.T) { + attr := observability.SerializedAttribute("message.content", unserializableValue{}) + value, ok := attr.Value.(string) + if !ok { + t.Fatalf("attribute value type = %T, want string", attr.Value) + } + want := "[Unserializable: observability_test.unserializableValue]" + if value != want { + t.Fatalf("attribute value = %q, want %q", value, want) + } +} + +func TestSensitiveDataUsesFallbackForExecutorInputAndOutput(t *testing.T) { + span := &fakeSpan{} + telemetry := observability.New(observability.Options{ + Tracer: &fakeTracer{span: span}, + EnableSensitiveData: true, + }) + + message := unserializableValue{} + _, activity := telemetry.StartExecutorProcess(context.Background(), "exec1", "pkg.Type", "message", message, nil) + if activity == nil { + t.Fatal("expected an activity span") + } + telemetry.SetExecutorOutput(activity, message) + + want := "[Unserializable: observability_test.unserializableValue]" + if got := attributeValue(t, span.attrs, observability.TagExecutorInput); got != want { + t.Fatalf("executor.input = %q, want %q", got, want) + } + if got := attributeValue(t, span.attrs, observability.TagExecutorOutput); got != want { + t.Fatalf("executor.output = %q, want %q", got, want) + } +}