Map processing outcomes to versioned YAML reports
This commit is contained in:
@@ -1,164 +1,252 @@
|
||||
package processreport
|
||||
|
||||
import (
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
"gitea.maximumdirect.net/eric/audita/internal/core/chunking"
|
||||
"gitea.maximumdirect.net/eric/audita/internal/core/config"
|
||||
"gitea.maximumdirect.net/eric/audita/internal/core/diagnostics"
|
||||
"gitea.maximumdirect.net/eric/audita/internal/core/normalization"
|
||||
"gitea.maximumdirect.net/eric/audita/internal/core/reporting"
|
||||
"gitea.maximumdirect.net/eric/audita/internal/framework/contracts"
|
||||
"gitea.maximumdirect.net/eric/audita/internal/framework/outcomes"
|
||||
"gitea.maximumdirect.net/eric/audita/internal/framework/runner"
|
||||
stagewarnings "gitea.maximumdirect.net/eric/audita/internal/framework/warnings"
|
||||
)
|
||||
|
||||
// BuildInput contains already-computed process execution facts for report assembly.
|
||||
// RunFailure describes a fatal or output failure without carrying its raw cause.
|
||||
type RunFailure struct {
|
||||
Phase string
|
||||
Reason outcomes.ReasonCode
|
||||
OutputPath string
|
||||
Origin *outcomes.Reference
|
||||
StageIndex *int
|
||||
}
|
||||
|
||||
type BuildInput struct {
|
||||
Status string
|
||||
TranscriptPath string
|
||||
GlossaryPath string
|
||||
OutputPath string
|
||||
Modules []string
|
||||
OutputSchema string
|
||||
ConfigVersion *int
|
||||
StartedAt time.Time
|
||||
CompletedAt time.Time
|
||||
ErrorMessage string
|
||||
ErrorPhase string
|
||||
RunDirectoryPath string
|
||||
NormalizationSummary *normalization.NormalizationSummary
|
||||
ChunkingSummary *chunking.Summary
|
||||
RunOutput *runner.RunOutput
|
||||
Backends map[string]config.Backend
|
||||
RunID string
|
||||
Status string
|
||||
StartedAt time.Time
|
||||
CompletedAt *time.Time
|
||||
TranscriptPath string
|
||||
GlossaryPath string
|
||||
OutputPath string
|
||||
ModuleSpecs []contracts.ModuleRunSpec
|
||||
RunOutput *runner.RunOutput
|
||||
RunFailures []RunFailure
|
||||
}
|
||||
|
||||
// Build creates the public process report without owning command parsing or config loading.
|
||||
func Build(input BuildInput) reporting.ProcessReport {
|
||||
report := reporting.ProcessReport{
|
||||
ReportMetadata: reporting.ReportMetadata{
|
||||
ReportSchemaName: reporting.DefaultProcessReportSchemaName,
|
||||
ReportSchemaVersion: reporting.DefaultProcessReportSchemaVersion,
|
||||
OutputSchema: input.OutputSchema,
|
||||
ConfigVersion: input.ConfigVersion,
|
||||
},
|
||||
Phase: "default_pipeline",
|
||||
Status: input.Status,
|
||||
Operation: "process",
|
||||
TranscriptPath: input.TranscriptPath,
|
||||
GlossaryPath: input.GlossaryPath,
|
||||
OutputPath: input.OutputPath,
|
||||
Modules: append([]string(nil), input.Modules...),
|
||||
StartedAt: input.StartedAt,
|
||||
CompletedAt: &input.CompletedAt,
|
||||
ErrorPhase: input.ErrorPhase,
|
||||
// Build maps processing facts once into an immutable presentation snapshot.
|
||||
func Build(input BuildInput) reporting.Snapshot {
|
||||
snapshot := reporting.Snapshot{
|
||||
RunID: input.RunID, Status: input.Status, StartedAt: input.StartedAt.UTC(), CompletedAt: input.CompletedAt,
|
||||
TranscriptPath: input.TranscriptPath, GlossaryPath: input.GlossaryPath, OutputPath: input.OutputPath,
|
||||
StageNames: make([]string, len(input.ModuleSpecs)), Errors: make([]reporting.ErrorRecord, 0),
|
||||
Recoveries: make([]reporting.RecoveryRecord, 0), WorkUnits: make([]reporting.WorkUnit, 0),
|
||||
}
|
||||
if input.RunDirectoryPath != "" {
|
||||
runSucceeded := input.Status == "success"
|
||||
metadata := diagnostics.BuildDiagnosticsMetadata(input.RunDirectoryPath, runSucceeded)
|
||||
report.Diagnostics = &metadata
|
||||
for i, spec := range input.ModuleSpecs {
|
||||
snapshot.StageNames[i] = spec.InstanceName
|
||||
}
|
||||
if input.ErrorMessage != "" {
|
||||
report.ErrorMessage = input.ErrorMessage
|
||||
}
|
||||
if input.NormalizationSummary != nil {
|
||||
report.InputSegmentCount = &input.NormalizationSummary.InputSegmentCount
|
||||
report.NormalizedSegmentCount = &input.NormalizationSummary.OutputSegmentCount
|
||||
report.NormalizationMerges = &input.NormalizationSummary.MergesPerformed
|
||||
report.NormalizationIDReassignments = &input.NormalizationSummary.IDsReassigned
|
||||
report.NormalizationSkipped.DifferentSpeakers = &input.NormalizationSummary.SkippedMerges.DifferentSpeakers
|
||||
report.NormalizationSkipped.GapTooLarge = &input.NormalizationSummary.SkippedMerges.GapTooLarge
|
||||
report.NormalizationSkipped.DurationExceeded = &input.NormalizationSummary.SkippedMerges.DurationExceeded
|
||||
report.NormalizationSkipped.TokenLimitExceeded = &input.NormalizationSummary.SkippedMerges.TokenLimitExceeded
|
||||
}
|
||||
if input.ChunkingSummary != nil {
|
||||
report.Chunking = &reporting.ChunkingSummary{
|
||||
ChunkCount: input.ChunkingSummary.ChunkCount,
|
||||
MinEstimatedTokens: input.ChunkingSummary.MinEstimatedTokens,
|
||||
MaxEstimatedTokens: input.ChunkingSummary.MaxEstimatedTokens,
|
||||
TotalEstimatedTokens: input.ChunkingSummary.TotalEstimatedTokens,
|
||||
TargetSections: input.ChunkingSummary.TargetSections,
|
||||
MaxSectionTokens: input.ChunkingSummary.MaxSectionTokens,
|
||||
MinSectionTokens: input.ChunkingSummary.MinSectionTokens,
|
||||
}
|
||||
}
|
||||
report.ModulesSummary, report.ModuleResults = buildModuleReporting(input.RunOutput)
|
||||
if input.RunOutput != nil && input.RunOutput.ObservationsAvailable {
|
||||
summary := mapLLMSummary(input.RunOutput.LLMSummary)
|
||||
report.LLMSummary = &summary
|
||||
}
|
||||
return report
|
||||
}
|
||||
|
||||
func buildModuleReporting(runOutput *runner.RunOutput) (*reporting.ModulesSummary, []reporting.ModuleReport) {
|
||||
if runOutput == nil || len(runOutput.ModuleResults) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
moduleReports := make([]reporting.ModuleReport, 0, len(runOutput.ModuleResults))
|
||||
summary := &reporting.ModulesSummary{ModuleCount: len(runOutput.ModuleResults)}
|
||||
for _, r := range runOutput.ModuleResults {
|
||||
startedAt := r.StartedAt
|
||||
completedAt := r.CompletedAt
|
||||
moduleReports = append(moduleReports, reporting.ModuleReport{
|
||||
ModuleKey: r.ModuleKey,
|
||||
ModuleInstance: r.ModuleInstance,
|
||||
ReplacementPolicy: string(r.ReplacementPolicy),
|
||||
Status: r.Status,
|
||||
ProposalCount: r.ProposalCount,
|
||||
Warnings: append([]stagewarnings.StageWarning(nil), r.Warnings...),
|
||||
ValidatorDecisions: mapValidatorDecisions(r.ValidatorDecisions),
|
||||
ValidatorRejected: mapValidatorRejected(r.ValidatorRejected),
|
||||
AppliedChanges: r.AppliedChanges,
|
||||
SkippedChanges: r.SkippedChanges,
|
||||
ErrorMessage: r.ErrorMessage,
|
||||
StartedAt: &startedAt,
|
||||
CompletedAt: &completedAt,
|
||||
if input.RunOutput != nil {
|
||||
units := append([]runner.WorkUnitOutcome(nil), input.RunOutput.WorkUnits...)
|
||||
sort.SliceStable(units, func(i, j int) bool {
|
||||
if units[i].ModuleIndex != units[j].ModuleIndex {
|
||||
return units[i].ModuleIndex < units[j].ModuleIndex
|
||||
}
|
||||
return units[i].Section.Index < units[j].Section.Index
|
||||
})
|
||||
summary.TotalAppliedChanges += len(r.AppliedChanges)
|
||||
summary.TotalSkippedChanges += len(r.SkippedChanges) + len(r.ValidatorRejected)
|
||||
if r.Status == runner.ModuleStatusFailed && summary.FailedModuleInstance == "" {
|
||||
summary.FailedModuleInstance = r.ModuleInstance
|
||||
total, started, completed, interrupted, notStarted, withErrors := len(units), 0, 0, 0, 0, 0
|
||||
corrections, llmOperations, proposalResponses, recoveryCount, filteringResponses := 0, 0, 0, 0, 0
|
||||
for _, unit := range units {
|
||||
if unit.State == outcomes.WorkNotStarted {
|
||||
notStarted++
|
||||
} else {
|
||||
started++
|
||||
}
|
||||
if unit.State == outcomes.WorkCompleted {
|
||||
completed++
|
||||
}
|
||||
if unit.State == outcomes.WorkInterrupted {
|
||||
interrupted++
|
||||
}
|
||||
corrections += unit.CorrectionsApplied
|
||||
row := reporting.WorkUnit{StageIndex: unit.ModuleIndex, ModuleKey: unit.ModuleKey, ModuleInstance: unit.ModuleInstance,
|
||||
ChunkIndex: unit.Section.Index, StartSegmentID: unit.Section.StartSegmentID, EndSegmentID: unit.Section.EndSegmentID,
|
||||
State: string(unit.State), CorrectionsApplied: unit.CorrectionsApplied}
|
||||
operations := append([]outcomes.Operation(nil), unit.Operations...)
|
||||
sort.SliceStable(operations, func(i, j int) bool { return operationLess(operations[i], operations[j]) })
|
||||
failures := make([]reporting.Failure, 0)
|
||||
for _, operation := range operations {
|
||||
if operation.Reference.Kind != outcomes.OperationValidationInput {
|
||||
llmOperations++
|
||||
}
|
||||
if operation.Reference.Kind == outcomes.OperationProposal && operation.Succeeded {
|
||||
proposalResponses++
|
||||
}
|
||||
if operation.Succeeded && len(operation.RecoveryReasons) > 0 {
|
||||
recoveryCount++
|
||||
row.RecoveryCount++
|
||||
if len(operation.Filtered) > 0 && operation.Reference.Kind == outcomes.OperationProposal {
|
||||
filteringResponses++
|
||||
}
|
||||
snapshot.Recoveries = append(snapshot.Recoveries, mapRecovery(unit, operation))
|
||||
}
|
||||
if operation.FailureReason != "" && isRecoverableReason(operation.FailureReason) {
|
||||
failures = append(failures, mapFailure(operation))
|
||||
}
|
||||
}
|
||||
if len(failures) > 0 {
|
||||
withErrors++
|
||||
row.ErrorCount = 1
|
||||
stage, chunk, start, end, applied := unit.ModuleIndex, unit.Section.Index, unit.Section.StartSegmentID, unit.Section.EndSegmentID, unit.CorrectionsApplied
|
||||
snapshot.Errors = append(snapshot.Errors, reporting.ErrorRecord{Scope: "work_unit", StageIndex: &stage,
|
||||
ModuleKey: unit.ModuleKey, ModuleInstance: unit.ModuleInstance, ChunkIndex: &chunk,
|
||||
StartSegmentID: &start, EndSegmentID: &end, CorrectionsApplied: &applied, Failures: failures})
|
||||
}
|
||||
snapshot.WorkUnits = append(snapshot.WorkUnits, row)
|
||||
}
|
||||
runErrors := len(input.RunFailures)
|
||||
snapshot.Summary = reporting.Summary{
|
||||
WorkUnitsTotal: &total, WorkUnitsStarted: &started, WorkUnitsCompleted: &completed,
|
||||
WorkUnitsInterrupted: &interrupted, WorkUnitsNotStarted: ¬Started, WorkUnitsWithErrors: &withErrors,
|
||||
RunErrors: &runErrors, CorrectionsApplied: &corrections, LLMOperations: &llmOperations,
|
||||
ProposalResponses: &proposalResponses, Recoveries: &recoveryCount,
|
||||
ProposalResponsesWithFiltering: &filteringResponses,
|
||||
}
|
||||
} else if len(input.RunFailures) > 0 {
|
||||
runErrors := len(input.RunFailures)
|
||||
snapshot.Summary.RunErrors = &runErrors
|
||||
}
|
||||
|
||||
return summary, moduleReports
|
||||
for _, failure := range sortedRunFailures(input.RunFailures) {
|
||||
record := reporting.ErrorRecord{Scope: "run", Phase: failure.Phase, Reason: string(failure.Reason),
|
||||
Message: safeMessage(failure.Reason), OutputPath: failure.OutputPath, StageIndex: failure.StageIndex}
|
||||
if failure.Origin != nil {
|
||||
record.ModuleKey, record.ModuleInstance = failure.Origin.ModuleKey, failure.Origin.ModuleInstance
|
||||
chunk := failure.Origin.SectionIndex
|
||||
record.ChunkIndex = &chunk
|
||||
record.Operation = string(failure.Origin.Kind)
|
||||
record.Validator = failure.Origin.ValidatorName
|
||||
record.BatchIndex = failure.Origin.BatchIndex
|
||||
}
|
||||
snapshot.Errors = append(snapshot.Errors, record)
|
||||
}
|
||||
return snapshot
|
||||
}
|
||||
|
||||
func mapValidatorDecisions(in []runner.ValidatorDecisionRecord) []reporting.ValidatorDecisionReport {
|
||||
if len(in) == 0 {
|
||||
return nil
|
||||
func mapFailure(operation outcomes.Operation) reporting.Failure {
|
||||
indices := append([]int(nil), operation.ProposalIndices...)
|
||||
sort.Ints(indices)
|
||||
return reporting.Failure{Operation: string(operation.Reference.Kind), Validator: operation.Reference.ValidatorName,
|
||||
BatchIndex: operation.Reference.BatchIndex, Reason: string(operation.FailureReason), Message: safeMessage(operation.FailureReason),
|
||||
ProposalIndices: indices, ExecutionCounts: mapCounts(operation.Counts)}
|
||||
}
|
||||
|
||||
func mapRecovery(unit runner.WorkUnitOutcome, operation outcomes.Operation) reporting.RecoveryRecord {
|
||||
reasons := make([]string, len(operation.RecoveryReasons))
|
||||
for i, reason := range operation.RecoveryReasons {
|
||||
reasons[i] = string(reason)
|
||||
}
|
||||
out := make([]reporting.ValidatorDecisionReport, len(in))
|
||||
for i, d := range in {
|
||||
out[i] = reporting.ValidatorDecisionReport{
|
||||
ValidatorName: d.ValidatorName,
|
||||
ProposalIndex: d.ProposalIndex,
|
||||
Approved: d.Approved,
|
||||
ReasonCode: d.ReasonCode,
|
||||
Message: d.Message,
|
||||
DiagnosticArtifactPath: d.DiagnosticArtifactPath,
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func mapValidatorRejected(in []runner.ValidatorRejectedChange) []reporting.ValidatorRejectedReport {
|
||||
if len(in) == 0 {
|
||||
return nil
|
||||
}
|
||||
out := make([]reporting.ValidatorRejectedReport, len(in))
|
||||
for i, d := range in {
|
||||
out[i] = reporting.ValidatorRejectedReport{
|
||||
ValidatorName: d.ValidatorName,
|
||||
ProposalIndex: d.ProposalIndex,
|
||||
ModuleKey: d.ModuleKey,
|
||||
ModuleInstance: d.ModuleInstance,
|
||||
TargetSegmentID: d.TargetSegmentID,
|
||||
OriginalText: d.OriginalText,
|
||||
CorrectedText: d.CorrectedText,
|
||||
ReasonCode: d.ReasonCode,
|
||||
Message: d.Message,
|
||||
sort.Strings(reasons)
|
||||
filtered := make([]reporting.FilteredProposal, len(operation.Filtered))
|
||||
for i, item := range operation.Filtered {
|
||||
filtered[i] = reporting.FilteredProposal{ProposalIndex: item.ProposalIndex, Reason: string(item.Reason), Count: item.Count}
|
||||
}
|
||||
sort.Slice(filtered, func(i, j int) bool { return filtered[i].ProposalIndex < filtered[j].ProposalIndex })
|
||||
return reporting.RecoveryRecord{StageIndex: unit.ModuleIndex, ModuleKey: unit.ModuleKey, ModuleInstance: unit.ModuleInstance,
|
||||
ChunkIndex: unit.Section.Index, StartSegmentID: unit.Section.StartSegmentID, EndSegmentID: unit.Section.EndSegmentID,
|
||||
Operation: string(operation.Reference.Kind), Validator: operation.Reference.ValidatorName, BatchIndex: operation.Reference.BatchIndex,
|
||||
Reasons: reasons, Message: "Model operation succeeded after recovery or proposal filtering.",
|
||||
ExecutionCounts: mapCounts(operation.Counts), FilteredProposals: filtered}
|
||||
}
|
||||
|
||||
func mapCounts(counts outcomes.ExecutionCounts) reporting.ExecutionCounts {
|
||||
if !counts.Available {
|
||||
return reporting.ExecutionCounts{}
|
||||
}
|
||||
attempts, retries, repairs, exhausted := counts.ProviderAttempts, counts.TransientRetries, counts.StructuralRepairs, counts.RetriesExhausted
|
||||
return reporting.ExecutionCounts{ProviderAttempts: &attempts, TransientRetries: &retries,
|
||||
StructuralRepairs: &repairs, RetriesExhausted: &exhausted, HTTPStatus: counts.HTTPStatus}
|
||||
}
|
||||
|
||||
func isRecoverableReason(reason outcomes.ReasonCode) bool {
|
||||
switch reason {
|
||||
case outcomes.ReasonOperationTimeout, outcomes.ReasonCapacityExceeded, outcomes.ReasonTransportFailed,
|
||||
outcomes.ReasonProviderResponseInvalid, outcomes.ReasonProviderFailed,
|
||||
outcomes.ReasonInvalidOutputExhausted, outcomes.ReasonInvalidOutput,
|
||||
outcomes.ReasonValidationInputTooLarge, outcomes.ReasonValidationDecisionsInvalid:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func safeMessage(reason outcomes.ReasonCode) string {
|
||||
switch reason {
|
||||
case outcomes.ReasonOperationTimeout:
|
||||
return "Model operation timed out."
|
||||
case outcomes.ReasonCapacityExceeded:
|
||||
return "Model capacity was unavailable."
|
||||
case outcomes.ReasonTransportFailed:
|
||||
return "Model transport failed."
|
||||
case outcomes.ReasonProviderResponseInvalid:
|
||||
return "Provider response was invalid."
|
||||
case outcomes.ReasonProviderFailed:
|
||||
return "Provider request failed."
|
||||
case outcomes.ReasonInvalidOutputExhausted:
|
||||
return "Model output remained invalid after repair attempts."
|
||||
case outcomes.ReasonInvalidOutput:
|
||||
return "Model output was invalid."
|
||||
case outcomes.ReasonValidationInputTooLarge:
|
||||
return "Validation input exceeded its size limit."
|
||||
case outcomes.ReasonValidationDecisionsInvalid:
|
||||
return "Validation decisions were unusable."
|
||||
case outcomes.ReasonCanceled:
|
||||
return "Processing was canceled."
|
||||
case outcomes.ReasonSetupFailure:
|
||||
return "Processing setup failed."
|
||||
default:
|
||||
return "Processing failed."
|
||||
}
|
||||
}
|
||||
|
||||
func operationLess(a, b outcomes.Operation) bool {
|
||||
rank := func(kind outcomes.OperationKind) int {
|
||||
switch kind {
|
||||
case outcomes.OperationProposal:
|
||||
return 0
|
||||
case outcomes.OperationValidationInput:
|
||||
return 1
|
||||
default:
|
||||
return 2
|
||||
}
|
||||
}
|
||||
if rank(a.Reference.Kind) != rank(b.Reference.Kind) && (a.Reference.Kind == outcomes.OperationProposal || b.Reference.Kind == outcomes.OperationProposal) {
|
||||
return rank(a.Reference.Kind) < rank(b.Reference.Kind)
|
||||
}
|
||||
order := func(p *int) int {
|
||||
if p == nil {
|
||||
return -1
|
||||
}
|
||||
return *p
|
||||
}
|
||||
if order(a.Reference.ValidatorOrder) != order(b.Reference.ValidatorOrder) {
|
||||
return order(a.Reference.ValidatorOrder) < order(b.Reference.ValidatorOrder)
|
||||
}
|
||||
return rank(a.Reference.Kind) < rank(b.Reference.Kind) || (rank(a.Reference.Kind) == rank(b.Reference.Kind) && order(a.Reference.BatchIndex) < order(b.Reference.BatchIndex))
|
||||
}
|
||||
|
||||
func sortedRunFailures(in []RunFailure) []RunFailure {
|
||||
out := append([]RunFailure(nil), in...)
|
||||
phase := map[string]int{"configuration": 0, "destination": 1, "initialization": 2, "input": 3, "normalization": 4, "chunking": 5, "processing": 6, "encoding": 7, "transcript_write": 8, "report_write": 9, "debug": 10}
|
||||
sort.SliceStable(out, func(i, j int) bool {
|
||||
a, aok := phase[out[i].Phase]
|
||||
b, bok := phase[out[j].Phase]
|
||||
if !aok {
|
||||
a = 100
|
||||
}
|
||||
if !bok {
|
||||
b = 100
|
||||
}
|
||||
if a != b {
|
||||
return a < b
|
||||
}
|
||||
return out[i].OutputPath < out[j].OutputPath
|
||||
})
|
||||
return out
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user