255 lines
10 KiB
Go
255 lines
10 KiB
Go
package processreport
|
|
|
|
import (
|
|
"sort"
|
|
"time"
|
|
|
|
"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"
|
|
)
|
|
|
|
// 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 {
|
|
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 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),
|
|
}
|
|
for i, spec := range input.ModuleSpecs {
|
|
snapshot.StageNames[i] = spec.InstanceName
|
|
}
|
|
if input.RunOutput != nil && input.RunOutput.InventoryAvailable {
|
|
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
|
|
})
|
|
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
|
|
}
|
|
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 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)
|
|
}
|
|
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."
|
|
case outcomes.ReasonInputInvalid:
|
|
return "Input was invalid."
|
|
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
|
|
}
|