Identify pipeline entries in reports and diagnostics

internal/framework/processreport/attribution_test.go

@@ -0,0 +1,36 @@
+package processreport
+
+import (
+	"strings"
+	"testing"
+	"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"
+	"gitea.maximumdirect.net/eric/audita/v3/internal/framework/validators"
+)
+
+func TestUnavailableCandidateAndFailureKeepIdentity(t *testing.T) {
+	id := "review/name"
+	index := 1
+	output := &runner.RunOutput{InventoryAvailable: true, ModuleResults: []runner.ModuleResult{
+		{StageIndex: 1, StageID: &id, ModuleKey: "grammar", ModuleInstance: "grammar_2", ValidatorRejected: []runner.ValidatorRejectedChange{{ProposalIndex: 3, Disposition: validators.DispositionUnavailable, ReasonCode: "provider_failed"}}},
+	}, WorkUnits: []runner.WorkUnitOutcome{{ModuleIndex: 1, StageIndex: 1, StageID: &id, ModuleKey: "grammar", ModuleInstance: "grammar_2", State: outcomes.WorkCompleted, Operations: []outcomes.Operation{{Reference: outcomes.Reference{StageIndex: &index, StageID: &id, Kind: outcomes.OperationValidation}, FailureReason: outcomes.ReasonProviderFailed}}}}}
+	ledger := BuildCorrectionLedger(CorrectionLedgerInput{RunOutput: output})
+	if len(ledger) != 1 || ledger[0].Disposition != "unavailable" || ledger[0].StageIndex != 1 || ledger[0].StageID == nil || *ledger[0].StageID != id {
+		t.Fatalf("ledger=%+v", ledger)
+	}
+	snapshot := Build(BuildInput{StartedAt: time.Now(), Status: reporting.StatusCompleted, ModuleSpecs: []contracts.ModuleRunSpec{{Index: 0, ModuleKey: "glossary"}, {Index: 1, ID: &id, ModuleKey: "grammar"}}, RunOutput: output})
+	if len(snapshot.Errors) != 1 || snapshot.Errors[0].StageID == nil || *snapshot.Errors[0].StageID != id || snapshot.Errors[0].StageIndex == nil || *snapshot.Errors[0].StageIndex != 1 {
+		t.Fatalf("errors=%+v", snapshot.Errors)
+	}
+	raw, err := reporting.YAML(snapshot.ErrorsDocument())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(string(raw), "stage_id: review/name") || !strings.Contains(string(raw), "stage_index: 1") {
+		t.Fatalf("error report: %s", raw)
+	}
+}

internal/framework/processreport/correction_ledger.go

@@ -25,6 +25,8 @@ type CorrectionLedgerInput struct {
 }
 
 type CorrectionLedgerEntry struct {
+	StageIndex                    int                             `json:"stage_index"`
+	StageID                       *string                         `json:"stage_id,omitempty"`
 	RunID                         string                          `json:"run_id,omitempty"`
 	ModuleKey                     string                          `json:"module_key"`
 	ModuleInstance                string                          `json:"module_instance"`
@@ -71,6 +73,7 @@ func BuildCorrectionLedger(input CorrectionLedgerInput) []CorrectionLedgerEntry
 		for _, change := range module.AppliedChanges {
 			seen[identity(module.ModuleInstance, change.ProposalIndex)] = true
 			entries = append(entries, CorrectionLedgerEntry{
+				StageIndex: module.StageIndex, StageID: module.StageID,
 				RunID:                         runID,
 				ModuleKey:                     module.ModuleKey,
 				ModuleInstance:                module.ModuleInstance,
@@ -88,6 +91,7 @@ func BuildCorrectionLedger(input CorrectionLedgerInput) []CorrectionLedgerEntry
 		for _, change := range module.SkippedChanges {
 			seen[identity(module.ModuleInstance, change.ProposalIndex)] = true
 			entries = append(entries, CorrectionLedgerEntry{
+				StageIndex: module.StageIndex, StageID: module.StageID,
 				RunID:                         runID,
 				ModuleKey:                     module.ModuleKey,
 				ModuleInstance:                module.ModuleInstance,
@@ -110,6 +114,7 @@ func BuildCorrectionLedger(input CorrectionLedgerInput) []CorrectionLedgerEntry
 			}
 			seen[identity(module.ModuleInstance, rejection.ProposalIndex)] = true
 			entries = append(entries, CorrectionLedgerEntry{
+				StageIndex: module.StageIndex, StageID: module.StageID,
 				RunID:                         runID,
 				ModuleKey:                     module.ModuleKey,
 				ModuleInstance:                module.ModuleInstance,
@@ -127,6 +132,7 @@ func BuildCorrectionLedger(input CorrectionLedgerInput) []CorrectionLedgerEntry
 		}
 		if module.Status == runner.ModuleStatusFailed {
 			entries = append(entries, CorrectionLedgerEntry{
+				StageIndex: module.StageIndex, StageID: module.StageID,
 				RunID:                 runID,
 				ModuleKey:             module.ModuleKey,
 				ModuleInstance:        module.ModuleInstance,
@@ -141,11 +147,14 @@ func BuildCorrectionLedger(input CorrectionLedgerInput) []CorrectionLedgerEntry
 			continue
 		}
 		var decisions []runner.ValidatorDecisionRecord
+		var stageIndex int
+		var stageID *string
 		policy := ""
 		for _, module := range runOutput.ModuleResults {
 			if module.ModuleInstance != candidate.ModuleInstance {
 				continue
 			}
+			stageIndex, stageID = module.StageIndex, module.StageID
 			policy = string(module.ReplacementPolicy)
 			for _, decision := range module.ValidatorDecisions {
 				if decision.ProposalIndex == candidate.ProposalIndex {
@@ -155,7 +164,7 @@ func BuildCorrectionLedger(input CorrectionLedgerInput) []CorrectionLedgerEntry
 			break
 		}
 		entries = append(entries, CorrectionLedgerEntry{
-			RunID: runID, ModuleKey: candidate.ModuleKey, ModuleInstance: candidate.ModuleInstance,
+			StageIndex: stageIndex, StageID: stageID, RunID: runID, ModuleKey: candidate.ModuleKey, ModuleInstance: candidate.ModuleInstance,
 			ProposalIndex: candidate.ProposalIndex, SegmentID: candidate.TargetSegmentID,
 			OriginalText: candidate.OriginalText, ProposedCorrectedText: &candidate.CorrectedText,
 			ReplacementPolicy: policy, Disposition: correctionDispositionInterrupted,
@@ -166,8 +175,8 @@ func BuildCorrectionLedger(input CorrectionLedgerInput) []CorrectionLedgerEntry
 	}
 
 	sort.SliceStable(entries, func(i, j int) bool {
-		if entries[i].ModuleInstance != entries[j].ModuleInstance {
-			return entries[i].ModuleInstance < entries[j].ModuleInstance
+		if entries[i].StageIndex != entries[j].StageIndex {
+			return entries[i].StageIndex < entries[j].StageIndex
 		}
 		if entries[i].ProposalIndex != entries[j].ProposalIndex {
 			return entries[i].ProposalIndex < entries[j].ProposalIndex

internal/framework/processreport/report.go

@@ -41,7 +41,7 @@ func Build(input BuildInput) reporting.Snapshot {
 		Recoveries: make([]reporting.RecoveryRecord, 0), WorkUnits: make([]reporting.WorkUnit, 0),
 	}
 	for i, spec := range input.ModuleSpecs {
-		snapshot.StageNames[i] = spec.InstanceName
+		snapshot.StageNames[i] = reporting.StageLabel(spec.Index, spec.ModuleKey, spec.ID)
 	}
 	if input.RunOutput != nil && input.RunOutput.InventoryAvailable {
 		units := append([]runner.WorkUnitOutcome(nil), input.RunOutput.WorkUnits...)
@@ -66,7 +66,7 @@ func Build(input BuildInput) reporting.Snapshot {
 				interrupted++
 			}
 			corrections += unit.CorrectionsApplied
-			row := reporting.WorkUnit{StageIndex: unit.ModuleIndex, ModuleKey: unit.ModuleKey, ModuleInstance: unit.ModuleInstance,
+			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}
 			operations := append([]outcomes.Operation(nil), unit.Operations...)
@@ -95,7 +95,7 @@ func Build(input BuildInput) reporting.Snapshot {
 				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,
+				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})
 			}
@@ -118,6 +118,10 @@ func Build(input BuildInput) reporting.Snapshot {
 			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
+			}
 			chunk := failure.Origin.SectionIndex
 			record.ChunkIndex = &chunk
 			record.Operation = string(failure.Origin.Kind)
@@ -148,7 +152,7 @@ func mapRecovery(unit runner.WorkUnitOutcome, operation outcomes.Operation) repo
 		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,
+	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.",