audita/internal/cli/process_pipeline.go

package cli

import (
	"context"
	"errors"
	"fmt"

	"gitea.maximumdirect.net/eric/audita/v3/internal/core/diagnostics"
	"gitea.maximumdirect.net/eric/audita/v3/internal/core/reporting"
	"gitea.maximumdirect.net/eric/audita/v3/internal/core/schema"
	"gitea.maximumdirect.net/eric/audita/v3/internal/framework/contracts"
	"gitea.maximumdirect.net/eric/audita/v3/internal/framework/llm"
	"gitea.maximumdirect.net/eric/audita/v3/internal/framework/modules"
	"gitea.maximumdirect.net/eric/audita/v3/internal/framework/outcomes"
	"gitea.maximumdirect.net/eric/audita/v3/internal/framework/processreport"
	"gitea.maximumdirect.net/eric/audita/v3/internal/framework/runner"
	"gitea.maximumdirect.net/eric/audita/v3/internal/framework/structuredoutput"
)

func executePipeline(ctx context.Context, loaded *processInputs, prepared *preparedProcess, normalized *schema.Transcript, warn func(error)) (processResult, error) {
	cfg := &loaded.Configuration.Config
	specs := contracts.ResolveModuleRunSpecs(cfg)
	debug := prepared.Prepared.Debug
	var observer *llm.ObservationCollector
	var hooks llm.CallHooks
	if debug != nil {
		observer = llm.NewObservationCollector()
		hooks = observer.Hooks()
	}
	for i := range specs {
		item := &specs[i]
		proposal, err := loaded.Runtime.NewRoleClient("proposal", *item.Settings.LLM.ProposalProfile, item.Settings.LLM, hooks)
		if err != nil {
			return processResult{}, fmt.Errorf("%s proposal client: %w", item.InstanceName, err)
		}
		validation, err := loaded.Runtime.NewRoleClient("validation", *item.Settings.LLM.ValidationProfile, item.Settings.LLM, hooks)
		if err != nil {
			return processResult{}, fmt.Errorf("%s validation client: %w", item.InstanceName, err)
		}
		item.ProposalClient, item.ValidationClient = proposal, validation
	}
	debugPath := prepared.DebugPath()
	output, runErr := runner.New(modules.NewFactory()).Run(ctx, runner.RunInput{
		Config: cfg, Transcript: normalized, ModuleSpecs: specs,
		ProposalDiagnosticsDir: debugPath, ValidationDiagnosticsDir: debugPath,
		CaptureCandidates: debugPath != "", ReportDiagnosticError: warn,
	})
	if debug != nil {
		warn(debug.WriteJSONArtifact(diagnostics.ArtifactLLMExecution, processreport.BuildLLMExecution(observer.Snapshot())))
		warn(debug.WriteJSONArtifact(diagnostics.ArtifactCorrectionLedger, processreport.BuildCorrectionLedger(processreport.CorrectionLedgerInput{RunDirectoryPath: debug.Path(), RunOutput: &output})))
	}
	var failures []processreport.RunFailure
	if runErr != nil {
		failure := processreport.RunFailure{Phase: "processing", Reason: outcomes.ReasonInternalFailure}
		var cause *structuredoutput.Failure
		if errors.As(runErr, &cause) {
			failure.Reason = cause.Reason
		} else if errors.Is(runErr, context.Canceled) || (ctx.Err() != nil && errors.Is(runErr, ctx.Err())) {
			failure.Reason = outcomes.ReasonCanceled
		}
		if len(output.ModuleResults) > 0 {
			last := output.ModuleResults[len(output.ModuleResults)-1]
			if last.Status == runner.ModuleStatusFailed {
				index := last.StageIndex
				failure.Origin = &outcomes.Reference{StageIndex: &index, StageID: last.StageID, ModuleKey: last.ModuleKey, ModuleInstance: last.ModuleInstance}
			}
		}
		failures = append(failures, failure)
	}
	snapshot := processreport.Build(processreport.BuildInput{RunID: prepared.RunID, Status: reporting.StatusInProgress, StartedAt: prepared.Started,
		ModuleSpecs: specs, RunOutput: &output, RunFailures: failures, SummaryOnly: !cfg.Reporting.Enabled})
	return processResult{Transcript: output.FinalTranscript, Snapshot: snapshot}, runErr
}