72 lines
3.4 KiB
Go
72 lines
3.4 KiB
Go
package cli
|
|
|
|
import (
|
|
"context"
|
|
"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"
|
|
)
|
|
|
|
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)
|
|
observer := llm.NewObservationCollector()
|
|
for i := range specs {
|
|
item := &specs[i]
|
|
proposal, err := loaded.Runtime.NewRoleClient("proposal", *item.Settings.LLM.ProposalProfile, item.Settings.LLM, observer.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, observer.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,
|
|
})
|
|
output.Observations = observer.Snapshot()
|
|
output.ObservationsAvailable = true
|
|
output.LLMSummary = llm.Summarize(output.Observations)
|
|
if debug := prepared.Prepared.Debug; debug != nil {
|
|
warn(debug.WriteJSONArtifact(diagnostics.ArtifactLLMExecution, processreport.BuildLLMExecution(output.Observations)))
|
|
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}
|
|
if ctx.Err() != nil {
|
|
failure.Reason = outcomes.ReasonCanceled
|
|
}
|
|
if len(output.ModuleResults) > 0 {
|
|
last := output.ModuleResults[len(output.ModuleResults)-1]
|
|
index := last.StageIndex
|
|
failure.StageIndex = &index
|
|
origin := outcomes.Reference{StageIndex: &index, StageID: last.StageID, ModuleKey: last.ModuleKey, ModuleInstance: last.ModuleInstance}
|
|
for _, operation := range last.Operations {
|
|
if operation.FailureReason != "" {
|
|
origin = operation.Reference
|
|
failure.Reason = operation.FailureReason
|
|
break
|
|
}
|
|
}
|
|
failure.Origin = &origin
|
|
}
|
|
failures = append(failures, failure)
|
|
}
|
|
snapshot := processreport.Build(processreport.BuildInput{RunID: prepared.RunID, Status: reporting.StatusInProgress, StartedAt: prepared.Started, ModuleSpecs: specs, RunOutput: &output, RunFailures: failures})
|
|
return processResult{Transcript: output.FinalTranscript, Snapshot: snapshot}, runErr
|
|
}
|