audita/internal/framework/processreport/report.go

package processreport

import (
	"sort"
	"time"

	"gitea.maximumdirect.net/eric/audita/v3/internal/core/reporting"
	"gitea.maximumdirect.net/eric/audita/v3/internal/framework/contracts"
	"gitea.maximumdirect.net/eric/audita/v3/internal/framework/outcomes"
	"gitea.maximumdirect.net/eric/audita/v3/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
	OutputPath     string
	ModuleSpecs    []contracts.ModuleRunSpec
	RunOutput      *runner.RunOutput
	RunFailures    []RunFailure
	SummaryOnly    bool
}

// 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, OutputPath: input.OutputPath,
	}
	details := !input.SummaryOnly
	if details {
		snapshot.StageNames = make([]string, len(input.ModuleSpecs))
		snapshot.Errors = make([]reporting.ErrorRecord, 0)
		snapshot.Recoveries = make([]reporting.RecoveryRecord, 0)
		snapshot.WorkUnits = make([]reporting.WorkUnit, 0)
		for i, spec := range input.ModuleSpecs {
			snapshot.StageNames[i] = reporting.StageLabel(spec.Index, spec.ModuleKey, spec.ID)
		}
	}
	if input.RunOutput != nil && input.RunOutput.InventoryAvailable {
		units := input.RunOutput.WorkUnits
		if details {
			units = append([]runner.WorkUnitOutcome(nil), units...)
			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
			operations := unit.Operations
			if details {
				operations = append([]outcomes.Operation(nil), operations...)
				sort.SliceStable(operations, func(i, j int) bool { return operationLess(operations[i], operations[j]) })
			}
			var failures []reporting.Failure
			unitHasError := false
			unitRecoveries := 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++
					unitRecoveries++
					if len(operation.Filtered) > 0 && operation.Reference.Kind == outcomes.OperationProposal {
						filteringResponses++
					}
					if details {
						snapshot.Recoveries = append(snapshot.Recoveries, mapRecovery(unit, operation))
					}
				}
				if operation.FailureReason != "" && isRecoverableReason(operation.FailureReason) {
					unitHasError = true
					if details {
						failures = append(failures, mapFailure(operation))
					}
				}
			}
			if unitHasError {
				withErrors++
			}
			if !details {
				continue
			}
			row := reporting.WorkUnit{StageIndex: unit.ModuleIndex, StageID: unit.StageID, 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, RecoveryCount: unitRecoveries}
			if unitHasError {
				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, StageID: unit.StageID,
					ModuleKey: unit.ModuleKey, ModuleInstance: unit.ModuleInstance, ChunkIndex: &chunk,
					StartSegmentID: &start, EndSegmentID: &end, CorrectionsApplied: &applied, Failures: failures})
			}
			snapshot.WorkUnits = append(snapshot.WorkUnits, row)
			for _, repair := range unit.Repairs {
				if repair.AttemptCount == 0 {
					continue
				}
				snapshot.RepairOutcomes = append(snapshot.RepairOutcomes, reporting.RepairOutcome{StageIndex: unit.ModuleIndex, ChunkIndex: unit.Section.Index, ProposalIndex: repair.ProposalIndex, Attempts: repair.AttemptCount, Outcome: repair.Outcome})
			}
		}
		sort.SliceStable(snapshot.RepairOutcomes, func(i, j int) bool {
			a, b := snapshot.RepairOutcomes[i], snapshot.RepairOutcomes[j]
			if a.StageIndex != b.StageIndex {
				return a.StageIndex < b.StageIndex
			}
			if a.ChunkIndex != b.ChunkIndex {
				return a.ChunkIndex < b.ChunkIndex
			}
			return a.ProposalIndex < b.ProposalIndex
		})
		runErrors := len(input.RunFailures)
		snapshot.Summary = reporting.Summary{
			WorkUnitsTotal: &total, WorkUnitsStarted: &started, WorkUnitsCompleted: &completed,
			WorkUnitsInterrupted: &interrupted, WorkUnitsNotStarted: &notStarted, 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
			record.StageID = failure.Origin.StageID
			if record.StageIndex == nil {
				record.StageIndex = failure.Origin.StageIndex
			}
			if failure.Origin.Kind != "" {
				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, StageID: unit.StageID, 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 response 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 "Some required validation decisions were unavailable."
	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
		}
	}
	order := func(p *int) int {
		if p == nil {
			return -1
		}
		return *p
	}
	aRepair, bRepair := a.Reference.RepairAttempt != nil, b.Reference.RepairAttempt != nil
	if aRepair != bRepair {
		return !aRepair
	}
	if aRepair {
		if order(a.Reference.ProposalIndex) != order(b.Reference.ProposalIndex) {
			return order(a.Reference.ProposalIndex) < order(b.Reference.ProposalIndex)
		}
		if order(a.Reference.RepairAttempt) != order(b.Reference.RepairAttempt) {
			return order(a.Reference.RepairAttempt) < order(b.Reference.RepairAttempt)
		}
		if (a.Reference.Kind == outcomes.OperationCorrectionRepair) != (b.Reference.Kind == outcomes.OperationCorrectionRepair) {
			return a.Reference.Kind == outcomes.OperationCorrectionRepair
		}
	}
	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)
	}

	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
}