fix(control-plane): prevent structured logs from leaking execution payloads (#701)

* fix(control-plane): prevent structured logs from leaking execution payloads (#560)

Add logging configuration (level + redact_payloads) to control what
execution data appears in structured log events and the internal event bus.

Changes:
- Add LoggingConfig with 'level' and 'redact_payloads' options
- Support AGENTFIELD_LOG_LEVEL and AGENTFIELD_LOG_REDACT_PAYLOADS env vars
- Guard execution input/output/context in event publishing behind redaction flag
- Default to redact_payloads=true (safe) — opt-in via config to see full payloads
- Replace 32 log.Printf calls in storage layer with leveled logger.Logger calls
- Add InitLoggerWithLevel() for string-based log level configuration
- Re-initialize logger from config at server startup

Closes #560

* fix: address Copilot review comments

- Guard req.Result behind redactPayloads in handleStatusUpdate event data
- Remove raw data preview from corrupted JSON warning log (log only metadata: context + length)

* test(control-plane): cover payload redaction branches for patch-coverage gate (#701)

The structured-log payload redaction added in #701 gated execution
input/result/context data behind `redactPayloads`. Those branches were only
exercised on their redact-enabled default, leaving the opt-out paths uncovered
and dropping control-plane patch coverage below the 80% floor.

Adds behavior tests that subscribe to the execution event bus and assert the
observable contract: input/result/context payloads are omitted from published
events when redaction is enabled (the safe default) and present only when an
operator explicitly disables it. Covers completeExecution, failExecution,
completeReplayHit, handleStatusUpdate, and the event-context path.

Patch coverage on touched lines: 68% -> 93%. Additive only; does not weaken the
redaction logic.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Abir Abbas <abirabbas1998@gmail.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: 3 people

control-plane/cmd/agentfield-server/main.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/Agent-Field/agentfield/control-plane/internal/cli"
 	"github.com/Agent-Field/agentfield/control-plane/internal/config"
+	"github.com/Agent-Field/agentfield/control-plane/internal/logger"
 	"github.com/Agent-Field/agentfield/control-plane/internal/server"
 	"github.com/Agent-Field/agentfield/control-plane/internal/utils"
 	"github.com/Agent-Field/agentfield/control-plane/web/client"
@@ -70,6 +71,13 @@ func runServer(cmd *cobra.Command, args []string) {
 		log.Fatalf("Failed to load configuration: %v", err)
 	}
 
+	// Re-initialize logger with configured level now that config is loaded.
+	// The CLI root command sets a default (info/debug based on --verbose),
+	// but the YAML/env-based level takes precedence once available.
+	if cfg.Logging.Level != "" {
+		logger.InitLoggerWithLevel(cfg.Logging.Level)
+	}
+
 	// Override port from flag if provided
 	if cmd.Flags().Lookup("port").Changed {
 		port, _ := cmd.Flags().GetInt("port")

control-plane/config/agentfield.yaml

@@ -21,6 +21,23 @@ type Config struct {
 	UI         UIConfig         `yaml:"ui" mapstructure:"ui"`
 	API        APIConfig        `yaml:"api" mapstructure:"api"`
 	Telemetry  TelemetryConfig  `yaml:"telemetry" mapstructure:"telemetry"`
+	Logging    LoggingConfig    `yaml:"logging" mapstructure:"logging"`
+}
+
+// LoggingConfig controls structured logging behavior.
+type LoggingConfig struct {
+	// Level sets the minimum log level: "debug", "info", "warn", "error".
+	// Defaults to "info".
+	Level string `yaml:"level" mapstructure:"level"`
+	// RedactPayloads controls whether execution input/output payloads are
+	// omitted from structured log events and internal event bus data.
+	// Defaults to true (payloads are redacted).
+	RedactPayloads *bool `yaml:"redact_payloads" mapstructure:"redact_payloads"`
+}
+
+// ShouldRedactPayloads returns true (the safe default) unless explicitly set to false.
+func (l LoggingConfig) ShouldRedactPayloads() bool {
+	return l.RedactPayloads == nil || *l.RedactPayloads
 }
 
 // TelemetryConfig controls anonymous OSS usage telemetry. It is separate from
@@ -468,6 +485,9 @@ func ApplyDefaults(cfg *Config) {
 	if cfg.Telemetry.Timeout <= 0 {
 		cfg.Telemetry.Timeout = 800 * time.Millisecond
 	}
+	if cfg.Logging.Level == "" {
+		cfg.Logging.Level = "info"
+	}
 }
 
 // ApplyEnvOverrides applies environment variable overrides to the config.
@@ -781,6 +801,15 @@ func ApplyEnvOverrides(cfg *Config) {
 			}
 		}
 	}
+
+	// Logging overrides
+	if val := os.Getenv("AGENTFIELD_LOG_LEVEL"); val != "" {
+		cfg.Logging.Level = strings.ToLower(strings.TrimSpace(val))
+	}
+	if val := os.Getenv("AGENTFIELD_LOG_REDACT_PAYLOADS"); val != "" {
+		b := parseEnvBool(val)
+		cfg.Logging.RedactPayloads = &b
+	}
 }
 
 func parseEnvBool(value string) bool {

control-plane/internal/config/config.go

@@ -491,3 +491,79 @@ func TestApplyEnvOverridesIgnoresInvalidValues(t *testing.T) {
 		t.Fatalf("expected connector enabled to become false for non-true value")
 	}
 }
+
+func TestLoggingConfigShouldRedactPayloads(t *testing.T) {
+	t.Parallel()
+
+	// Default (nil pointer) should redact
+	cfg := LoggingConfig{}
+	if !cfg.ShouldRedactPayloads() {
+		t.Fatal("expected ShouldRedactPayloads() to return true when RedactPayloads is nil")
+	}
+
+	// Explicitly true
+	true_ := true
+	cfg.RedactPayloads = &true_
+	if !cfg.ShouldRedactPayloads() {
+		t.Fatal("expected ShouldRedactPayloads() to return true when RedactPayloads is true")
+	}
+
+	// Explicitly false
+	false_ := false
+	cfg.RedactPayloads = &false_
+	if cfg.ShouldRedactPayloads() {
+		t.Fatal("expected ShouldRedactPayloads() to return false when RedactPayloads is false")
+	}
+}
+
+func TestLoggingApplyDefaults(t *testing.T) {
+	t.Parallel()
+
+	cfg := Config{}
+	ApplyDefaults(&cfg)
+
+	if cfg.Logging.Level != "info" {
+		t.Fatalf("expected default logging level 'info', got %q", cfg.Logging.Level)
+	}
+}
+
+func TestLoggingEnvOverrides(t *testing.T) {
+	// Test log level override
+	t.Run("log_level", func(t *testing.T) {
+		os.Setenv("AGENTFIELD_LOG_LEVEL", "debug")
+		defer os.Unsetenv("AGENTFIELD_LOG_LEVEL")
+
+		cfg := Config{}
+		ApplyEnvOverrides(&cfg)
+
+		if cfg.Logging.Level != "debug" {
+			t.Fatalf("expected log level 'debug', got %q", cfg.Logging.Level)
+		}
+	})
+
+	// Test redact payloads override
+	t.Run("redact_payloads_false", func(t *testing.T) {
+		os.Setenv("AGENTFIELD_LOG_REDACT_PAYLOADS", "false")
+		defer os.Unsetenv("AGENTFIELD_LOG_REDACT_PAYLOADS")
+
+		cfg := Config{}
+		ApplyEnvOverrides(&cfg)
+
+		if cfg.Logging.RedactPayloads == nil || *cfg.Logging.RedactPayloads != false {
+			t.Fatal("expected RedactPayloads to be false")
+		}
+	})
+
+	// Test redact payloads override (true)
+	t.Run("redact_payloads_true", func(t *testing.T) {
+		os.Setenv("AGENTFIELD_LOG_REDACT_PAYLOADS", "true")
+		defer os.Unsetenv("AGENTFIELD_LOG_REDACT_PAYLOADS")
+
+		cfg := Config{}
+		ApplyEnvOverrides(&cfg)
+
+		if cfg.Logging.RedactPayloads == nil || *cfg.Logging.RedactPayloads != true {
+			t.Fatal("expected RedactPayloads to be true")
+		}
+	})
+}

control-plane/internal/config/config_additional_test.go

@@ -132,14 +132,15 @@ type replayHit struct {
 }
 
 type executionController struct {
-	store         ExecutionStore
-	httpClient    *http.Client
-	payloads      services.PayloadStore
-	webhooks      services.WebhookDispatcher
-	eventBus      *events.ExecutionEventBus
-	timeout       time.Duration
-	internalToken string // sent as Authorization header when forwarding to agents
-	readARDConfig func() config.ARDConfig
+	store          ExecutionStore
+	httpClient     *http.Client
+	payloads       services.PayloadStore
+	webhooks       services.WebhookDispatcher
+	eventBus       *events.ExecutionEventBus
+	timeout        time.Duration
+	internalToken  string // sent as Authorization header when forwarding to agents
+	readARDConfig  func() config.ARDConfig
+	redactPayloads bool
 }
 
 type asyncExecutionJob struct {
@@ -166,8 +167,19 @@ var (
 
 	completionOnce  sync.Once
 	completionQueue chan completionJob
+
+	// defaultRedactPayloads controls whether execution input/output data is
+	// excluded from internal event bus payloads. Set at server startup from
+	// config.Logging.ShouldRedactPayloads(). Default true (safe).
+	defaultRedactPayloads = true
 )
 
+// SetRedactPayloads configures the package-level default for payload redaction.
+// Call this once at server startup after loading config.
+func SetRedactPayloads(redact bool) {
+	defaultRedactPayloads = redact
+}
+
 const (
 	maxWebhookHeaders      = 20
 	maxWebhookHeaderLength = 512
@@ -225,12 +237,13 @@ func newExecutionController(store ExecutionStore, payloads services.PayloadStore
 		httpClient: &http.Client{
 			Timeout: timeout,
 		},
-		payloads:      payloads,
-		webhooks:      webhooks,
-		eventBus:      store.GetExecutionEventBus(),
-		timeout:       timeout,
-		internalToken: internalToken,
-		readARDConfig: ardConfigReader,
+		payloads:       payloads,
+		webhooks:       webhooks,
+		eventBus:       store.GetExecutionEventBus(),
+		timeout:        timeout,
+		internalToken:  internalToken,
+		readARDConfig:  ardConfigReader,
+		redactPayloads: defaultRedactPayloads,
 	}
 }
 
@@ -953,15 +966,17 @@ func (c *executionController) handleStatusUpdate(ctx *gin.Context) {
 	}
 
 	eventData := map[string]interface{}{
-		"result":   req.Result,
 		"error":    req.Error,
 		"progress": req.Progress,
 	}
 	if req.StatusReason != nil && strings.TrimSpace(*req.StatusReason) != "" {
 		eventData["status_reason"] = strings.TrimSpace(*req.StatusReason)
 	}
-	if inputPayload := decodeJSON(updated.InputPayload); inputPayload != nil {
-		eventData["input"] = inputPayload
+	if !c.redactPayloads {
+		eventData["result"] = req.Result
+		if inputPayload := decodeJSON(updated.InputPayload); inputPayload != nil {
+			eventData["input"] = inputPayload
+		}
 	}
 	c.publishExecutionEvent(updated, normalizedStatus, eventData)
 
@@ -1076,7 +1091,7 @@ func (c *executionController) publishExecutionEventWithReasonerInfo(exec *types.
 		data["actor_id"] = *exec.ActorID
 	}
 	storedPayload := types.DecodeStoredExecutionPayload(exec.InputPayload)
-	if storedPayload.Context != nil {
+	if !c.redactPayloads && storedPayload.Context != nil {
 		data["context"] = storedPayload.Context
 	}
 	if workflowExec, err := c.store.GetWorkflowExecution(context.Background(), exec.ExecutionID); err == nil && workflowExec != nil {
@@ -1694,10 +1709,12 @@ func (c *executionController) completeReplayHit(ctx context.Context, plan *prepa
 			"source_execution_id": plan.replayHit.SourceExecutionID,
 			"source_run_id":       plan.replayHit.SourceRunID,
 		},
-		"result": decodeJSON(result),
 	}
-	if inputPayload := decodeJSON(plan.exec.InputPayload); inputPayload != nil {
-		eventData["input"] = inputPayload
+	if !c.redactPayloads {
+		eventData["result"] = decodeJSON(result)
+		if inputPayload := decodeJSON(plan.exec.InputPayload); inputPayload != nil {
+			eventData["input"] = inputPayload
+		}
 	}
 	c.publishExecutionEventWithReasonerInfo(updated, string(types.ExecutionStatusSucceeded), eventData, plan.agent, &plan.target.TargetName)
 	return nil
@@ -1871,11 +1888,13 @@ func (c *executionController) completeExecution(ctx context.Context, plan *prepa
 				c.triggerWebhook(plan.exec.ExecutionID)
 			}
 			eventData := map[string]interface{}{}
-			if payload := decodeJSON(result); payload != nil {
-				eventData["result"] = payload
-			}
-			if inputPayload := decodeJSON(plan.exec.InputPayload); inputPayload != nil {
-				eventData["input"] = inputPayload
+			if !c.redactPayloads {
+				if payload := decodeJSON(result); payload != nil {
+					eventData["result"] = payload
+				}
+				if inputPayload := decodeJSON(plan.exec.InputPayload); inputPayload != nil {
+					eventData["input"] = inputPayload
+				}
 			}
 			c.publishExecutionEventWithReasonerInfo(updated, string(types.ExecutionStatusSucceeded), eventData, plan.agent, &plan.target.TargetName)
 			return nil
@@ -1955,11 +1974,13 @@ func (c *executionController) failExecution(ctx context.Context, plan *preparedE
 			eventData := map[string]interface{}{
 				"error": errMsg,
 			}
-			if payload := decodeJSON(result); payload != nil {
-				eventData["result"] = payload
-			}
-			if inputPayload := decodeJSON(plan.exec.InputPayload); inputPayload != nil {
-				eventData["input"] = inputPayload
+			if !c.redactPayloads {
+				if payload := decodeJSON(result); payload != nil {
+					eventData["result"] = payload
+				}
+				if inputPayload := decodeJSON(plan.exec.InputPayload); inputPayload != nil {
+					eventData["input"] = inputPayload
+				}
 			}
 			c.publishExecutionEventWithReasonerInfo(updated, string(types.ExecutionStatusFailed), eventData, plan.agent, &plan.target.TargetName)
 			return nil