feat(op-acceptor): add CI parallelism splitting and report-from-events mode

Add --split-total / --split-index flags for deterministic round-robin test
distribution across CircleCI parallel nodes. Each node independently discovers
all tests, sorts by package|function name, and runs only its assigned subset.

Add --report-from-events flag that reads a merged raw_go_events.log file and
generates a consolidated HTML report through the standard FileLogger pipeline,
enabling a post-parallelism report consolidation job.

Also consolidate duplicate CSV parsing helpers into a single parseCSV function
and remove unreachable gate validation code.

Author: scharissis

op-acceptor/config.go

@@ -37,48 +37,50 @@ type Config struct {
 	FlakeShake           bool                   // Enable flake-shake mode for test stability validation
 	FlakeShakeIterations int                    // Number of times to run each test in flake-shake mode
 	DryRun               bool                   // If true, show what tests would be run without executing them
+	ReportFromEvents     string                 // Path to raw events file for report-only mode (empty = normal mode)
+	SplitTotal           int                    // Total split nodes for CI parallelism (0 = no splitting)
+	SplitIndex           int                    // This node's index (0-based) for CI parallelism
 	Log                  log.Logger
 	ExcludeGates         []string // List of gate IDs whose tests should be excluded
 }
 
-// NewConfig creates a new Config from cli context
-// parseGates parses a comma-separated string of gate IDs into a slice
-func parseGates(gateStr string) []string {
-	if gateStr == "" {
+// parseCSV splits a comma-separated string into a trimmed, non-empty slice.
+func parseCSV(s string) []string {
+	if s == "" {
 		return nil
 	}
-	var gates []string
-	for _, g := range strings.Split(gateStr, ",") {
-		g = strings.TrimSpace(g)
-		if g != "" {
-			gates = append(gates, g)
+	var out []string
+	for _, v := range strings.Split(s, ",") {
+		v = strings.TrimSpace(v)
+		if v != "" {
+			out = append(out, v)
 		}
 	}
-	return gates
+	return out
 }
 
 // NewConfig creates a new Config from cli context
 func NewConfig(ctx *cli.Context, log log.Logger, testDir string, validatorConfig string, gate string) (*Config, error) {
-	gates := parseGates(gate)
-	// Parse flags
-	if err := flags.CheckRequired(ctx); err != nil {
-		return nil, fmt.Errorf("missing required flags: %w", err)
-	}
-	if testDir == "" {
-		return nil, errors.New("test directory is required")
+	gates := parseCSV(gate)
+
+	reportFromEvents := ctx.String(flags.ReportFromEvents.Name)
+
+	// In report-from-events mode, testdir is not required
+	if reportFromEvents == "" {
+		if err := flags.CheckRequired(ctx); err != nil {
+			return nil, fmt.Errorf("missing required flags: %w", err)
+		}
+		if testDir == "" {
+			return nil, errors.New("test directory is required")
+		}
 	}
 
 	// Determine if we're in gateless mode (gate not specified). Validator config is optional in gateless mode
 	gatelessMode := len(gates) == 0
 
-	// In gateless mode, we don't require validator config or gate
-	if !gatelessMode {
-		if validatorConfig == "" {
-			return nil, errors.New("validator configuration file is required when not in gateless mode")
-		}
-		if len(gates) == 0 {
-			return nil, errors.New("gate is required when not in gateless mode")
-		}
+	// In gate mode, validator config is required
+	if !gatelessMode && validatorConfig == "" {
+		return nil, errors.New("validator configuration file is required when not in gateless mode")
 	}
 
 	var absValidatorConfig string
@@ -120,7 +122,15 @@ func NewConfig(ctx *cli.Context, log log.Logger, testDir string, validatorConfig
 
 	devnetEnvURL := ctx.String(flags.DevnetEnvURL.Name)
 
-	excludeGates := parseExcludeGates(ctx.String(flags.ExcludeGates.Name))
+	splitTotal := ctx.Int(flags.SplitTotal.Name)
+	splitIndex := ctx.Int(flags.SplitIndex.Name)
+	if splitTotal > 0 {
+		if splitIndex < 0 || splitIndex >= splitTotal {
+			return nil, fmt.Errorf("split-index must be >= 0 and < split-total (%d), got %d", splitTotal, splitIndex)
+		}
+	}
+
+	excludeGates := parseCSV(ctx.String(flags.ExcludeGates.Name))
 
 	// Conflict: selected gates are also excluded
 	for _, g := range gates {
@@ -153,26 +163,11 @@ func NewConfig(ctx *cli.Context, log log.Logger, testDir string, validatorConfig
 		FlakeShake:           ctx.Bool(flags.FlakeShake.Name),
 		FlakeShakeIterations: ctx.Int(flags.FlakeShakeIterations.Name),
 		DryRun:               ctx.Bool(flags.DryRun.Name),
+		ReportFromEvents:     ctx.String(flags.ReportFromEvents.Name),
+		SplitTotal:           splitTotal,
+		SplitIndex:           splitIndex,
 		LogDir:               logDir,
 		Log:                  log,
 		ExcludeGates:         excludeGates,
 	}, nil
 }
-
-// parseExcludeGates determines which gates to exclude based on env/flag.
-
-func parseExcludeGates(value string) []string {
-	val := strings.TrimSpace(value)
-	if val == "" {
-		// No default exclusions: empty means no skip gates
-		return nil
-	}
-	parts := strings.Split(val, ",")
-	var gates []string
-	for _, p := range parts {
-		if s := strings.TrimSpace(p); s != "" {
-			gates = append(gates, s)
-		}
-	}
-	return gates
-}

op-acceptor/flags/flags.go

@@ -185,6 +185,24 @@ var (
 		EnvVars: opservice.PrefixEnvVar(EnvVarPrefix, "EXCLUDE_GATES"),
 		Usage:   "Comma-separated list of gate IDs to blacklist globally across all modes.",
 	}
+	ReportFromEvents = &cli.StringFlag{
+		Name:    "report-from-events",
+		Value:   "",
+		EnvVars: opservice.PrefixEnvVar(EnvVarPrefix, "REPORT_FROM_EVENTS"),
+		Usage:   "Path to a raw_go_events.log file (or merged file). Generates an HTML report from the events without running tests.",
+	}
+	SplitTotal = &cli.IntFlag{
+		Name:    "split-total",
+		Value:   0,
+		EnvVars: opservice.PrefixEnvVar(EnvVarPrefix, "SPLIT_TOTAL"),
+		Usage:   "Total number of split nodes for CI parallelism. 0 = no splitting.",
+	}
+	SplitIndex = &cli.IntFlag{
+		Name:    "split-index",
+		Value:   0,
+		EnvVars: opservice.PrefixEnvVar(EnvVarPrefix, "SPLIT_INDEX"),
+		Usage:   "Index of this node (0-based) for CI parallelism.",
+	}
 )
 
 var requiredFlags = []cli.Flag{
@@ -212,6 +230,9 @@ var optionalFlags = []cli.Flag{
 	FlakeShakeIterations,
 	ExcludeGates,
 	DryRun,
+	ReportFromEvents,
+	SplitTotal,
+	SplitIndex,
 }
 var Flags []cli.Flag
 

op-acceptor/nat.go

@@ -105,6 +105,21 @@ func New(ctx context.Context, config *Config, version string, shutdownCallback f
 		return nil, errors.New("config is required")
 	}
 
+	// In report-from-events mode, we only need config + log dir.
+	// Skip registry, runner, orchestrator, and addons setup.
+	if config.ReportFromEvents != "" {
+		config.Log.Info("Report-from-events mode: skipping test runner setup")
+		return &nat{
+			ctx:              ctx,
+			config:           config,
+			version:          version,
+			done:             make(chan struct{}),
+			shutdownCallback: shutdownCallback,
+			networkName:      "report",
+			tracer:           otel.Tracer("op-acceptor"),
+		}, nil
+	}
+
 	reg, err := registry.NewRegistry(registry.Config{
 		Log:                 config.Log,
 		ValidatorConfigFile: config.ValidatorConfig,
@@ -182,6 +197,8 @@ func New(ctx context.Context, config *Config, version string, shutdownCallback f
 		Concurrency:        config.Concurrency,
 		ShowProgress:       config.ShowProgress,
 		ProgressInterval:   config.ProgressInterval,
+		SplitTotal:         config.SplitTotal,
+		SplitIndex:         config.SplitIndex,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to create test runner: %w", err)
@@ -307,6 +324,8 @@ func (n *nat) Start(ctx context.Context) error {
 			"concurrency", snap.Runner.Concurrency,
 			"show_progress", snap.Runner.ShowProgress,
 			"progress_interval", snap.Runner.ProgressInterval,
+			"split_total", n.config.SplitTotal,
+			"split_index", n.config.SplitIndex,
 			"test_log_level", snap.Logging.TestLogLevel,
 			"output_realtime_logs", snap.Logging.OutputRealtimeLogs,
 			"network", snap.NetworkName,
@@ -318,6 +337,18 @@ func (n *nat) Start(ctx context.Context) error {
 		"config.ValidatorConfig", n.config.ValidatorConfig,
 		"config.LogDir", n.config.LogDir)
 
+	// Report-from-events mode: generate HTML report from a raw events file, then exit.
+	if n.config.ReportFromEvents != "" {
+		err := n.reportFromEvents()
+		if err != nil {
+			return err
+		}
+		go func() {
+			n.shutdownCallback(nil)
+		}()
+		return nil
+	}
+
 	// Run tests immediately on startup (or dry-run)
 	if n.config.DryRun {
 		err := n.dryRun(ctx)
@@ -1011,9 +1042,39 @@ func (n *nat) dryRun(ctx context.Context) error {
 		gateValidators[gateID] = gateTests
 	}
 
+	// Apply CI split filtering if configured (mirrors runner.collectTestWork behavior)
+	if n.config.SplitTotal > 0 {
+		// Collect all work items in the same way as the runner
+		var allWork []runner.TestWork
+		for gateName, gateTests := range gateValidators {
+			for _, v := range gateTests {
+				allWork = append(allWork, runner.TestWork{
+					Validator: v,
+					GateID:    gateName,
+					SuiteID:   v.Suite,
+				})
+			}
+		}
+		filtered := runner.ApplySplitFilter(allWork, n.config.SplitTotal, n.config.SplitIndex)
+
+		// Rebuild gateValidators from the filtered work items
+		gateValidators = make(map[string][]types.ValidatorMetadata)
+		for _, w := range filtered {
+			gateValidators[w.GateID] = append(gateValidators[w.GateID], w.Validator)
+		}
+		n.config.Log.Info("DRY RUN: Applied CI split filter",
+			"splitTotal", n.config.SplitTotal,
+			"splitIndex", n.config.SplitIndex,
+			"workItems", len(filtered))
+	}
+
 	t := table.NewWriter()
 	t.SetOutputMirror(os.Stdout)
-	t.SetTitle("DRY RUN: Planned Test Execution (network: " + n.networkName + ")")
+	titleSuffix := ""
+	if n.config.SplitTotal > 0 {
+		titleSuffix = fmt.Sprintf(" [split %d/%d]", n.config.SplitIndex, n.config.SplitTotal)
+	}
+	t.SetTitle("DRY RUN: Planned Test Execution (network: " + n.networkName + ")" + titleSuffix)
 
 	headers := []interface{}{"TYPE", "ID", "TESTS", "STATUS"}
 	t.AppendHeader(table.Row(headers))
@@ -1100,6 +1161,52 @@ func (n *nat) dryRun(ctx context.Context) error {
 	return nil
 }
 
+// reportFromEvents reads a raw_go_events.log file (possibly merged from multiple
+// parallel nodes) and generates a consolidated report without running any tests.
+// It reuses the standard FileLogger pipeline so that all sinks (HTML, text summary,
+// per-test logs, etc.) are exercised through the same code path as normal execution.
+func (n *nat) reportFromEvents() error {
+	eventsPath := n.config.ReportFromEvents
+	n.config.Log.Info("Generating report from events file", "path", eventsPath)
+
+	f, err := os.Open(eventsPath)
+	if err != nil {
+		return fmt.Errorf("failed to open events file %s: %w", eventsPath, err)
+	}
+	defer f.Close()
+
+	results, err := runner.ParseMultiPackageEvents(f)
+	if err != nil {
+		return fmt.Errorf("failed to parse events: %w", err)
+	}
+
+	n.config.Log.Info("Parsed test results from events", "packages", len(results))
+
+	// Use the same FileLogger pipeline as normal test execution
+	runID := uuid.New().String()
+	fileLogger, err := logging.NewFileLogger(n.config.LogDir, runID, n.networkName, n.config.TargetGate)
+	if err != nil {
+		return fmt.Errorf("failed to create file logger: %w", err)
+	}
+
+	// Feed all parsed results through the standard sink pipeline
+	for _, result := range results {
+		if err := fileLogger.LogTestResult(result, runID); err != nil {
+			return fmt.Errorf("failed to log result for %s: %w", result.Metadata.Package, err)
+		}
+	}
+
+	// Finalize all sinks (generates HTML, text summary, etc.)
+	if err := fileLogger.Complete(runID); err != nil {
+		return fmt.Errorf("failed to complete report: %w", err)
+	}
+
+	logDir, _ := fileLogger.GetDirectoryForRunID(runID)
+	n.config.Log.Info("Consolidated report generated", "path", logDir)
+
+	return nil
+}
+
 // WaitForShutdown waits for all goroutines to finish
 func (n *nat) WaitForShutdown(ctx context.Context) error {
 	timeout := time.NewTimer(time.Second * 5)