Pipeline pattern workflow

.mockery.yaml

@@ -11,6 +11,7 @@ packages:
       Workflow:
       Orchestrator:
       Mutagen:
+      MutationStreamer:
   gooze.dev/pkg/gooze/internal/adapter:
     config:
       dir: internal/adapter/mocks

cmd/list.go

@@ -22,12 +22,15 @@ func newListCmd() *cobra.Command {
 			paths := parsePaths(args)
 			useCache := !viper.GetBool(noCacheFlagName)
 			reportsPath := m.Path(viper.GetString(outputFlagName))
+			shardIndex, totalShards := parseShardFlag(runShardFlag)
 
 			return workflow.Estimate(context.Background(), domain.EstimateArgs{
-				Paths:    paths,
-				Exclude:  viper.GetStringSlice(excludeConfigKey),
-				UseCache: useCache,
-				Reports:  reportsPath,
+				Paths:           paths,
+				Exclude:         viper.GetStringSlice(excludeConfigKey),
+				UseCache:        useCache,
+				Reports:         reportsPath,
+				ShardIndex:      shardIndex,
+				TotalShardCount: totalShards,
 			})
 		},
 	}

cmd/root.go

@@ -24,6 +24,7 @@ var fsAdapter adapter.SourceFSAdapter
 var testAdapter adapter.TestRunnerAdapter
 var orchestrator domain.Orchestrator
 var mutagen domain.Mutagen
+var mutationStreamer domain.MutationStreamer
 var workflow domain.Workflow
 var ui controller.UI
 
@@ -40,6 +41,9 @@ var noCacheFlag bool
 // excludePatterns is a root-level flag that filters files for applicable commands.
 var excludePatterns []string
 
+var runParallelFlag int
+var runShardFlag string
+
 func init() {
 	configureRootFlags(rootCmd)
 
@@ -52,12 +56,14 @@ func init() {
 	testAdapter = adapter.NewLocalTestRunnerAdapter()
 	orchestrator = domain.NewOrchestrator(fsAdapter, testAdapter)
 	mutagen = domain.NewMutagen(goFileAdapter, soirceFSAdapter)
+	mutationStreamer = domain.NewMutationStreamer(fsAdapter, reportStore, mutagen)
 	workflow = domain.NewWorkflow(
 		soirceFSAdapter,
 		reportStore,
 		ui,
 		orchestrator,
 		mutagen,
+		mutationStreamer,
 	)
 }
 
@@ -131,6 +137,8 @@ func configureRootFlags(cmd *cobra.Command) {
 	cmd.PersistentFlags().StringArrayVarP(&excludePatterns, excludeFlagName, "x", viper.GetStringSlice(excludeConfigKey), "exclude files matching regex (can be repeated)")
 	bindFlagToConfig(cmd.PersistentFlags().Lookup(excludeFlagName), excludeConfigKey)
 
+	cmd.PersistentFlags().StringVarP(&runShardFlag, "shard", "s", "", "shard index and total shard count in the format INDEX/TOTAL (e.g., 0/3)")
+
 	// Add verbose and log output flags (handled directly; not bound to Viper).
 	cmd.PersistentFlags().BoolVarP(&verboseFlag, "verbose", "v", false, "enable verbose logging")
 	cmd.PersistentFlags().StringVar(&logOutputFlag, "log-output", "", "path to the log output file")

cmd/run.go

@@ -13,8 +13,6 @@ import (
 )
 
 var mutationTimeoutFlag int64
-var runParallelFlag int
-var runShardFlag string
 
 // runCmd represents the run command.
 var runCmd = newRunCmd()
@@ -34,15 +32,14 @@ func newRunCmd() *cobra.Command {
 
 			return workflow.Test(context.Background(), domain.TestArgs{
 				EstimateArgs: domain.EstimateArgs{
-					Paths:    paths,
-					Exclude:  viper.GetStringSlice(excludeConfigKey),
-					UseCache: useCache,
-					Reports:  reportsPath,
+					Paths:           paths,
+					Exclude:         viper.GetStringSlice(excludeConfigKey),
+					UseCache:        useCache,
+					Reports:         reportsPath,
+					ShardIndex:      shardIndex,
+					TotalShardCount: totalShards,
 				},
-				Reports:         reportsPath,
 				Threads:         threads,
-				ShardIndex:      shardIndex,
-				TotalShardCount: totalShards,
 				MutationTimeout: time.Duration(timeoutSeconds) * time.Second,
 			})
 		},
@@ -63,7 +60,6 @@ func configureRunFlags(cmd *cobra.Command) {
 
 	cmd.Flags().IntVarP(&runParallelFlag, runParallelFlagName, "p", viper.GetInt(runParallelConfigKey), "number of parallel workers for mutation testing")
 	bindFlagToConfig(cmd.Flags().Lookup(runParallelFlagName), runParallelConfigKey)
-	cmd.Flags().StringVarP(&runShardFlag, "shard", "s", "", "shard index and total shard count in the format INDEX/TOTAL (e.g., 0/3)")
 }
 
 func parseShardFlag(shard string) (int, int) {

internal/adapter/report_store.go

@@ -22,12 +22,16 @@ const indexFileName = "_index.yaml"
 
 // ReportStore persists and retrieves mutation reports.
 type ReportStore interface {
+	// Deprecated: Use SaveReportsStream for better performance and lower memory usage on large mutation sets.
 	SaveReports(ctx context.Context, path m.Path, reports []m.Report) error
+	// Deprecated: Use SaveReportsStream or better performance and lower memory usage on large mutation sets.
 	SaveSpillReports(ctx context.Context, path m.Path, reports pkg.FileSpill[m.Report]) error
+	SaveReportsStream(ctx context.Context, path m.Path, reports <-chan m.Report) error
 	RegenerateIndex(ctx context.Context, path m.Path) error
 	LoadReports(ctx context.Context, path m.Path) ([]m.Report, error)
 	LoadSpillReports(ctx context.Context, path m.Path) (pkg.FileSpill[m.Report], error)
 	CheckUpdates(ctx context.Context, path m.Path, sources []m.Source) ([]m.Source, error)
+	CheckUpdate(ctx context.Context, path m.Path, source m.Source) (bool, error)
 	CleanReports(ctx context.Context, path m.Path, sources []m.Source) error
 }
 
@@ -160,6 +164,39 @@ func (rs *LocalReportStore) SaveSpillReports(ctx context.Context, path m.Path, r
 	})
 }
 
+// SaveReportsStream writes reports from the provided channel into the specified directory until the channel is closed.
+func (rs *LocalReportStore) SaveReportsStream(ctx context.Context, path m.Path, reports <-chan m.Report) error {
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	dirPath := string(path)
+	if dirPath == "" {
+		return fmt.Errorf("reports directory path is required")
+	}
+	if err := os.MkdirAll(string(path), 0o750); err != nil {
+		return fmt.Errorf("create reports directory: %w", err)
+	}
+	for report := range reports {
+		reportHash := rs.computeReportHash(report.Result)
+		if reportHash == "" {
+			return nil
+		}
+
+		data, err := rs.marshalReport(report)
+		if err != nil {
+			return fmt.Errorf("marshal report to YAML: %w", err)
+		}
+
+		name := reportHash + ".yaml"
+
+		fullPath := filepath.Join(dirPath, name)
+		if err := os.WriteFile(fullPath, data, 0o600); err != nil {
+			return fmt.Errorf("write report file %s: %w", fullPath, err)
+		}
+	}
+	return nil
+}
+
 // RegenerateIndex rebuilds and writes `_index.yaml` from the report files in `path`.
 func (rs *LocalReportStore) RegenerateIndex(ctx context.Context, path m.Path) error {
 	if err := ctx.Err(); err != nil {
@@ -467,6 +504,54 @@ func (rs *LocalReportStore) CheckUpdates(ctx context.Context, path m.Path, sourc
 	return changed, nil
 }
 
+func (rs *LocalReportStore) CheckUpdate(ctx context.Context, path m.Path, source m.Source) (bool, error) {
+	if err := ctx.Err(); err != nil {
+		return false, err
+	}
+	dirPath := string(path)
+	if dirPath == "" {
+		return false, fmt.Errorf("reports directory path is required")
+	}
+
+	exists, err := rs.reportsDirExists(dirPath)
+	if err != nil {
+		return false, err
+	}
+
+	if !exists {
+		// No prior reports: source needs to be tested
+		return true, nil
+	}
+
+	if source.Origin == nil || source.Origin.FullPath == "" {
+		return false, fmt.Errorf("source missing origin information")
+	}
+
+	reports, err := rs.loadReportsFromDir(dirPath)
+	if err != nil {
+		return false, err
+	}
+
+	stored := rs.buildStoredSourceState(reports)
+	pathStr := string(source.Origin.FullPath)
+
+	st, ok := stored[pathStr]
+	if !ok {
+		// No stored report for this source: needs testing
+		return true, nil
+	}
+
+	if rs.sourceHashChanged(st.source, source) {
+		return true, nil
+	}
+
+	if rs.mutatorsChanged(st.mutator) {
+		return true, nil
+	}
+
+	return false, nil
+}
+
 func (rs *LocalReportStore) buildCurrentSourceMap(sources []m.Source) map[string]m.Source {
 	currentByPath := map[string]m.Source{}