fix: address release review findings — distributed CEL context, email limits, metrics

- MakeGlobalStepExecutor now loads failed-continued steps into CEL context,
  matching the sequential resumeExecution path so downstream expressions
  can reference error/output from continue_on_error steps
- email/receive: cap limit param at 200 to prevent unbounded fetches
- email_receive fetchMessages: remove redundant defer cmd.Close() that
  caused a double-close after the explicit close at end of function
- metrics: add EmailTriggersSkippedTotal counter; increment in
  EmailTriggerPoller.startPoller when maxConnections limit is reached

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: michaelmcnees

internal/connector/email_receive.go

@@ -40,6 +40,8 @@ func (c *EmailReceiveConnector) Execute(ctx context.Context, params map[string]a
 		filter = v
 	}
 
+	const maxEmailFetchLimit = 200
+
 	limit := 10
 	switch v := params["limit"].(type) {
 	case int:
@@ -51,6 +53,9 @@ func (c *EmailReceiveConnector) Execute(ctx context.Context, params map[string]a
 			limit = int(v)
 		}
 	}
+	if limit > maxEmailFetchLimit {
+		limit = maxEmailFetchLimit
+	}
 
 	markSeen := false
 	if v, ok := params["mark_seen"].(bool); ok {
@@ -160,7 +165,6 @@ func buildSearchCriteria(filter string) *imap.SearchCriteria {
 // output map format expected by the connector.
 func fetchMessages(client *imapclient.Client, uidSet imap.UIDSet, opts *imap.FetchOptions) ([]map[string]any, error) {
 	cmd := client.Fetch(uidSet, opts)
-	defer func() { _ = cmd.Close() }()
 
 	var out []map[string]any
 

internal/engine/engine.go

@@ -725,6 +725,18 @@ func (e *Engine) MakeGlobalStepExecutor() StepExecutor {
 			celCtx.Steps[name] = map[string]any{"output": output}
 		}
 
+		// Load failed-but-continued steps for CEL context.
+		failedContinuedSteps, err := e.loadFailedContinuedSteps(ctx, execID)
+		if err != nil {
+			return nil, fmt.Errorf("loading failed continued steps: %w", err)
+		}
+		for name, fcs := range failedContinuedSteps {
+			celCtx.Steps[name] = map[string]any{
+				"output": fcs.output,
+				"error":  fcs.errMsg,
+			}
+		}
+
 		// Load artifacts for CEL context.
 		e.loadArtifactsIntoCELContext(ctx, execID, celCtx)
 

internal/metrics/metrics.go

@@ -151,6 +151,10 @@ var (
 		Name: "mantle_email_connection_errors_total",
 		Help: "Total IMAP connection errors",
 	}, []string{"workflow"})
+	EmailTriggersSkippedTotal = promauto.NewCounter(prometheus.CounterOpts{
+		Name: "mantle_email_triggers_skipped_total",
+		Help: "Total email triggers skipped due to connection limit",
+	})
 )
 
 // Budget metrics.

internal/server/email_trigger.go

@@ -121,6 +121,7 @@ func (e *EmailTriggerPoller) startPoller(ctx context.Context, t TriggerRecord) {
 			"trigger_id", t.ID,
 			"workflow", t.WorkflowName,
 			"max", e.maxConnections)
+		metrics.EmailTriggersSkippedTotal.Inc()
 		return
 	}