diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index dc85574..7792405 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -545,6 +545,11 @@ failure before outputs/model work and a valid no-stage normalization run. ## Stage 13: Carry stage identity through ordinary and debug evidence +**Completed.** Machine records carry execution position and optional operator ID; +human labels include position, module, and the ID when supplied. Controlled +processing and focused report checks cover repeated and unnamed entries, escaped +labels, debug evidence, recoveries, failures, and unavailable candidates. + Own `internal/framework/outcomes`, `processreport`, `structuredoutput` call context, and `internal/core/reporting`. Use the [stage attribution contract](glossary-stages.md#stage-identity-and-attribution--resolved). diff --git a/internal/cli/process_inputs.go b/internal/cli/process_inputs.go index 8a13820..6e2d850 100644 --- a/internal/cli/process_inputs.go +++ b/internal/cli/process_inputs.go @@ -88,8 +88,8 @@ func prepareProcess(ctx context.Context, input processInput, loaded *processInpu } started := time.Now().UTC() snapshot := reporting.Snapshot{RunID: runID, Status: reporting.StatusInProgress, StartedAt: started, TranscriptPath: input.Transcript, OutputPath: input.Output.Value} - for _, item := range cfg.Pipeline.Stages { - snapshot.StageNames = append(snapshot.StageNames, item.Module) + for index, item := range cfg.Pipeline.Stages { + snapshot.StageNames = append(snapshot.StageNames, reporting.StageLabel(index, item.Module, item.ID)) } secrets := loaded.Runtime.ConfiguredSecrets(ctx) initial, err := renderReportBundle(snapshot, secrets...) diff --git a/internal/cli/process_ordered_integration_test.go b/internal/cli/process_ordered_integration_test.go index 81fdfc4..0b7cf39 100644 --- a/internal/cli/process_ordered_integration_test.go +++ b/internal/cli/process_ordered_integration_test.go @@ -15,8 +15,11 @@ import ( "testing" "gitea.maximumdirect.net/eric/audita/v3/internal/core/config" + "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/llm" + commandkit "go.maximumdirect.net/commandkit" ck "go.maximumdirect.net/configkit" ) @@ -28,6 +31,7 @@ func TestProcessingUsesPerEntryProfilesInOrder(t *testing.T) { var mu sync.Mutex var models []string var prompts []string + proposalCalls := 0 transport := controlledTransport(func(r *http.Request) (*http.Response, error) { var request struct { Model string `json:"model"` @@ -45,11 +49,18 @@ func TestProcessingUsesPerEntryProfilesInOrder(t *testing.T) { mu.Lock() models = append(models, request.Model) prompts = append(prompts, contents) + if request.Model == "model-one" { + proposalCalls++ + } + firstProposalCall := proposalCalls mu.Unlock() result := `{"validations":[{"correction_index":1,"approved":true,"confidence":1,"reason":"ok"}]}` switch request.Model { case "model-one": result = `{"corrections":[{"id":1,"original_text":"Alyce","corrected_text":"Alice","confidence":1}]}` + if firstProposalCall == 1 { + result = `{"invalid":true}` + } case "model-two": result = `{"corrections":[{"id":1,"original_text":"Sheeld","corrected_text":"Shield","confidence":1}]}` case "model-three": @@ -73,9 +84,11 @@ llm: {validation_profile: review} pipeline: stages: - module: glossary + id: "party/../|" glossaries: [party] llm: {proposal_profile: one} - module: glossary + id: spell glossaries: [spells] llm: {proposal_profile: two} - module: grammar @@ -103,7 +116,8 @@ pipeline: t.Fatal(err) } inputs := &processInputs{Configuration: loaded, Runtime: runtime, Source: source} - prepared, err := prepareProcess(context.Background(), processInput{Transcript: filepath.Join(dir, "transcript.json")}, inputs, &bytes.Buffer{}) + loaded.Config.Diagnostics.WorkDir = dir + prepared, err := prepareProcess(context.Background(), processInput{Transcript: filepath.Join(dir, "transcript.json"), Debug: commandkit.Supplied[bool]{Value: true}}, inputs, &bytes.Buffer{}) if err != nil { t.Fatal(err) } @@ -115,10 +129,29 @@ pipeline: if result.Transcript.Segments[0].Text != "Alice cast Shield." { t.Fatalf("result=%q models=%v", result.Transcript.Segments[0].Text, models) } - if len(prompts) > 3 && !strings.Contains(prompts[3], "Alice casts Sheeld.") { - t.Fatalf("second proposal missed earlier edit: %s", prompts[3]) + if len(prompts) > 4 && !strings.Contains(prompts[4], "Alice casts Sheeld.") { + t.Fatalf("second proposal missed earlier edit: %s", prompts[4]) } - if !reflect.DeepEqual(models, []string{"model-one", "review", "review", "model-two", "review", "review", "model-three", "review", "review"}) { + if len(result.Snapshot.WorkUnits) != 3 || result.Snapshot.WorkUnits[0].StageID == nil || *result.Snapshot.WorkUnits[0].StageID != "party/../|" || result.Snapshot.WorkUnits[1].StageIndex != 1 || result.Snapshot.WorkUnits[2].StageID != nil { + t.Fatalf("work unit identities: %+v", result.Snapshot.WorkUnits) + } + if len(result.Snapshot.Recoveries) != 1 || result.Snapshot.Recoveries[0].StageIndex != 0 || result.Snapshot.Recoveries[0].StageID == nil || *result.Snapshot.Recoveries[0].StageID != "party/../|" { + t.Fatalf("recovery attribution: %+v", result.Snapshot.Recoveries) + } + markdown := string(reporting.Markdown(result.Snapshot)) + if !strings.Contains(markdown, `1: glossary (party/../\|<tag>)`) || !strings.Contains(markdown, "2: glossary (spell)") || !strings.Contains(markdown, "3: grammar") { + t.Fatalf("report labels: %s", markdown) + } + for _, artifact := range []string{diagnostics.ArtifactCorrectionLedger, diagnostics.ArtifactLLMExecution} { + raw, readErr := os.ReadFile(filepath.Join(prepared.DebugPath(), artifact)) + if readErr != nil { + t.Fatal(readErr) + } + if !strings.Contains(string(raw), `"stage_index": 0`) || !strings.Contains(string(raw), `"stage_id": "party/../|\u003ctag\u003e"`) || !strings.Contains(string(raw), `"stage_index": 1`) || !strings.Contains(string(raw), `"stage_id": "spell"`) || !strings.Contains(string(raw), `"stage_index": 2`) { + t.Fatalf("%s missing attribution: %s", artifact, raw) + } + } + if !reflect.DeepEqual(models, []string{"model-one", "model-one", "review", "review", "model-two", "review", "review", "model-three", "review", "review"}) { t.Fatalf("models=%v", models) } } @@ -126,7 +159,7 @@ pipeline: func TestRecoverableModelFailureKeepsNormalizedTranscript(t *testing.T) { dir := t.TempDir() files := map[string]string{ - "config.yml": "version: 3\nreporting: {enabled: false}\nllm: {proposal_profile: local, transient_retries: 0}\npipeline:\n stages: [{module: grammar}]\n", + "config.yml": "version: 3\nreporting: {enabled: false}\nllm: {proposal_profile: local, transient_retries: 0}\npipeline:\n stages: [{module: grammar, id: failing}]\n", "prompts.yml": "profiles:\n - {id: local, endpoint: 'https://example.invalid/v1', model: fixture, api_key_required: false}\n", "transcript.json": `[{"speaker":"Alice","start":0,"end":1,"text":"hello"}]`, } @@ -160,7 +193,7 @@ func TestRecoverableModelFailureKeepsNormalizedTranscript(t *testing.T) { if err != nil { t.Fatal(err) } - if result.Transcript.Segments[0].Text != "hello" || len(result.Snapshot.Errors) != 1 || result.Snapshot.Errors[0].Scope != "work_unit" { + if result.Transcript.Segments[0].Text != "hello" || len(result.Snapshot.Errors) != 1 || result.Snapshot.Errors[0].Scope != "work_unit" || result.Snapshot.Errors[0].StageID == nil || *result.Snapshot.Errors[0].StageID != "failing" { t.Fatalf("result=%+v errors=%+v", result.Transcript, result.Snapshot.Errors) } } diff --git a/internal/cli/process_pipeline.go b/internal/cli/process_pipeline.go index df5bbc7..b9182be 100644 --- a/internal/cli/process_pipeline.go +++ b/internal/cli/process_pipeline.go @@ -46,7 +46,25 @@ func executePipeline(ctx context.Context, loaded *processInputs, prepared *prepa } var failures []processreport.RunFailure if runErr != nil { - failures = append(failures, processreport.RunFailure{Phase: "processing", Reason: outcomes.ReasonInternalFailure}) + 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 diff --git a/internal/core/reporting/markdown.go b/internal/core/reporting/markdown.go index 30eff22..7cb715a 100644 --- a/internal/core/reporting/markdown.go +++ b/internal/core/reporting/markdown.go @@ -2,6 +2,7 @@ package reporting import ( "fmt" + "html" "sort" "strings" @@ -171,6 +172,7 @@ func recoveryReasons(records []RecoveryRecord) map[string]int { } func escapeMarkdown(value string) string { + value = html.EscapeString(value) value = strings.ReplaceAll(value, "\\", "\\\\") value = strings.ReplaceAll(value, "|", "\\|") value = strings.ReplaceAll(value, "`", "\\`") diff --git a/internal/core/reporting/report.go b/internal/core/reporting/report.go index c3dd237..c6305ce 100644 --- a/internal/core/reporting/report.go +++ b/internal/core/reporting/report.go @@ -52,6 +52,7 @@ type Failure struct { type ErrorRecord struct { Scope string `yaml:"scope"` StageIndex *int `yaml:"stage_index,omitempty"` + StageID *string `yaml:"stage_id,omitempty"` ModuleKey string `yaml:"module_key,omitempty"` ModuleInstance string `yaml:"module_instance,omitempty"` ChunkIndex *int `yaml:"chunk_index,omitempty"` @@ -76,6 +77,7 @@ type FilteredProposal struct { type RecoveryRecord struct { StageIndex int `yaml:"stage_index"` + StageID *string `yaml:"stage_id,omitempty"` ModuleKey string `yaml:"module_key"` ModuleInstance string `yaml:"module_instance"` ChunkIndex int `yaml:"chunk_index"` @@ -115,6 +117,7 @@ type RecoveriesDocument struct { // WorkUnit is payload-free matrix input for report.md. type WorkUnit struct { StageIndex int + StageID *string ModuleKey string ModuleInstance string ChunkIndex int @@ -178,3 +181,11 @@ func YAML(document any, secrets ...string) ([]byte, error) { } return safe, nil } + +// StageLabel identifies a configured entry by position, module, and optional ID. +func StageLabel(index int, module string, id *string) string { + if id != nil { + return fmt.Sprintf("%d: %s (%s)", index+1, module, *id) + } + return fmt.Sprintf("%d: %s", index+1, module) +} diff --git a/internal/framework/contracts/contracts.go b/internal/framework/contracts/contracts.go index 0141ae2..e428d93 100644 --- a/internal/framework/contracts/contracts.go +++ b/internal/framework/contracts/contracts.go @@ -70,6 +70,8 @@ func SectionMetadataFromSection(section chunking.Section) SectionMetadata { // ExecutionContext carries shared execution state for proposal generation and validation. type ExecutionContext struct { + StageIndex *int `json:"stage_index,omitempty"` + StageID *string `json:"stage_id,omitempty"` // ReportDiagnosticError may be called concurrently; it must not affect processing. ReportDiagnosticError func(error) `json:"-"` Settings config.Stage `json:"-"` diff --git a/internal/framework/llm/observations.go b/internal/framework/llm/observations.go index 70c5fee..5b90e4e 100644 --- a/internal/framework/llm/observations.go +++ b/internal/framework/llm/observations.go @@ -92,6 +92,14 @@ func (c *ObservationCollector) Hooks() CallHooks { } func copyContext(value structuredoutput.DiagnosticContext) structuredoutput.DiagnosticContext { + if value.StageIndex != nil { + index := *value.StageIndex + value.StageIndex = &index + } + if value.StageID != nil { + id := *value.StageID + value.StageID = &id + } if value.SectionIndex != nil { index := *value.SectionIndex value.SectionIndex = &index diff --git a/internal/framework/outcomes/outcomes.go b/internal/framework/outcomes/outcomes.go index 74e6972..957b762 100644 --- a/internal/framework/outcomes/outcomes.go +++ b/internal/framework/outcomes/outcomes.go @@ -16,6 +16,8 @@ const ( // are scoped to a module instance, while batch indices are local to a validator // invocation within one section. type Reference struct { + StageIndex *int `json:"stage_index,omitempty"` + StageID *string `json:"stage_id,omitempty"` ModuleKey string `json:"module_key"` ModuleInstance string `json:"module_instance"` SectionIndex int `json:"section_index"` diff --git a/internal/framework/processreport/attribution_test.go b/internal/framework/processreport/attribution_test.go new file mode 100644 index 0000000..5f24abf --- /dev/null +++ b/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) + } +} diff --git a/internal/framework/processreport/correction_ledger.go b/internal/framework/processreport/correction_ledger.go index d1877aa..a32ce76 100644 --- a/internal/framework/processreport/correction_ledger.go +++ b/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 diff --git a/internal/framework/processreport/report.go b/internal/framework/processreport/report.go index 9e0fa6e..fd4b331 100644 --- a/internal/framework/processreport/report.go +++ b/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.", diff --git a/internal/framework/proposal_generation/generate.go b/internal/framework/proposal_generation/generate.go index 4865ffd..f1e01fd 100644 --- a/internal/framework/proposal_generation/generate.go +++ b/internal/framework/proposal_generation/generate.go @@ -49,6 +49,8 @@ type StructuredCorrectionSet struct { // Request captures reusable proposal-generation inputs for future modules. type Request struct { + StageIndex *int `json:"stage_index,omitempty"` + StageID *string `json:"stage_id,omitempty"` // ReportDiagnosticError may be called concurrently; it must not affect processing. ReportDiagnosticError func(error) `json:"-"` ModuleKey string `json:"module_key"` @@ -124,7 +126,7 @@ func GenerateCandidates(ctx context.Context, req Request) (Result, error) { completion, callErr = req.LLMClient.CompleteStructured(ctx, contracts.StructuredCompletionRequest{ PromptID: req.PromptID, Variables: req.Variables, Schema: responseschema.CorrectionSetKey, - Context: structuredoutput.DiagnosticContext{Role: "proposal", ModuleKey: req.ModuleKey, ModuleInstance: req.ModuleInstance, SectionIndex: sectionIndexPtr(req.Section)}, + Context: structuredoutput.DiagnosticContext{StageIndex: req.StageIndex, StageID: req.StageID, Role: "proposal", ModuleKey: req.ModuleKey, ModuleInstance: req.ModuleInstance, SectionIndex: sectionIndexPtr(req.Section)}, }, &response) sectionIndex := 0 if req.Section != nil { @@ -132,6 +134,7 @@ func GenerateCandidates(ctx context.Context, req Request) (Result, error) { } operation := outcomes.Operation{ Reference: outcomes.Reference{ + StageIndex: req.StageIndex, StageID: req.StageID, ModuleKey: req.ModuleKey, ModuleInstance: req.ModuleInstance, SectionIndex: sectionIndex, Kind: outcomes.OperationProposal, }, @@ -139,6 +142,7 @@ func GenerateCandidates(ctx context.Context, req Request) (Result, error) { } requestMetadata := map[string]any{ + "stage_index": req.StageIndex, "stage_id": req.StageID, "module_key": req.ModuleKey, "module_instance": req.ModuleInstance, "replacement_policy": req.ReplacementPolicy, diff --git a/internal/framework/proposal_generation/module_proposal.go b/internal/framework/proposal_generation/module_proposal.go index 29e8b92..0fea353 100644 --- a/internal/framework/proposal_generation/module_proposal.go +++ b/internal/framework/proposal_generation/module_proposal.go @@ -53,6 +53,7 @@ func ExecuteModuleProposal(ctx context.Context, req ModuleProposalRequest) (cont } generated, err := GenerateCandidates(ctx, Request{ + StageIndex: req.ProposalRequest.StageIndex, StageID: req.ProposalRequest.StageID, ModuleKey: req.ProposalRequest.RunSpec.ModuleKey, ModuleInstance: req.ProposalRequest.RunSpec.InstanceName, ReplacementPolicy: req.ProposalRequest.RunSpec.ReplacementPolicy, diff --git a/internal/framework/runner/runner.go b/internal/framework/runner/runner.go index 4af7b71..7a02ae6 100644 --- a/internal/framework/runner/runner.go +++ b/internal/framework/runner/runner.go @@ -38,6 +38,8 @@ type Runner struct { // ModuleResult captures deterministic per-module execution output. type ModuleResult struct { + StageIndex int `json:"stage_index"` + StageID *string `json:"stage_id,omitempty"` ModuleKey string `json:"module_key"` ModuleInstance string `json:"module_instance"` ReplacementPolicy proposals.ReplacementPolicy `json:"replacement_policy"` @@ -56,6 +58,8 @@ type ModuleResult struct { // WorkUnitOutcome keeps initial section identity while text evolves between // module instances. Detailed candidate and decision records are debug evidence. type WorkUnitOutcome struct { + StageIndex int `json:"stage_index"` + StageID *string `json:"stage_id,omitempty"` ModuleIndex int `json:"module_index"` ModuleKey string `json:"module_key"` ModuleInstance string `json:"module_instance"` @@ -149,7 +153,7 @@ func (r *Runner) run(ctx context.Context, input RunInput) (RunOutput, error) { for moduleIndex, spec := range input.ModuleSpecs { for _, section := range inventory { workUnits = append(workUnits, WorkUnitOutcome{ - ModuleIndex: moduleIndex, ModuleKey: spec.ModuleKey, + ModuleIndex: moduleIndex, StageIndex: spec.Index, StageID: spec.ID, ModuleKey: spec.ModuleKey, ModuleInstance: spec.InstanceName, Section: section, State: outcomes.WorkNotStarted, }) @@ -167,6 +171,7 @@ func (r *Runner) run(ctx context.Context, input RunInput) (RunOutput, error) { module, err := r.factory.ModuleForSpec(spec) if err != nil { failed := ModuleResult{ + StageIndex: spec.Index, StageID: spec.ID, ModuleKey: spec.ModuleKey, ModuleInstance: spec.InstanceName, Status: ModuleStatusFailed, @@ -196,6 +201,7 @@ func (r *Runner) run(ctx context.Context, input RunInput) (RunOutput, error) { } if pipelineErr != nil { failed := ModuleResult{ + StageIndex: spec.Index, StageID: spec.ID, ModuleKey: spec.ModuleKey, ModuleInstance: spec.InstanceName, ReplacementPolicy: policy, @@ -235,6 +241,7 @@ func (r *Runner) run(ctx context.Context, input RunInput) (RunOutput, error) { } results = append(results, ModuleResult{ + StageIndex: spec.Index, StageID: spec.ID, ModuleKey: spec.ModuleKey, ModuleInstance: spec.InstanceName, ReplacementPolicy: policy, @@ -358,6 +365,7 @@ func collectSectionProposals(ctx context.Context, input collectSectionProposalsI meta := contracts.SectionMetadataFromSection(section) proposalResult, err := input.Module.Propose(runCtx, contracts.ProposalRequest{ ExecutionContext: contracts.ExecutionContext{ + StageIndex: &input.Spec.Index, StageID: input.Spec.ID, Settings: input.Spec.Settings, WorkingTranscript: transcriptFromSection(section), Glossary: input.Spec.Glossary, @@ -399,7 +407,7 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput) } for i, section := range input.Sections { out.WorkUnits[i] = WorkUnitOutcome{ - ModuleKey: input.Spec.ModuleKey, ModuleInstance: input.Spec.InstanceName, + ModuleKey: input.Spec.ModuleKey, ModuleInstance: input.Spec.InstanceName, StageIndex: input.Spec.Index, StageID: input.Spec.ID, Section: contracts.SectionMetadataFromSection(section), State: outcomes.WorkNotStarted, } } @@ -631,6 +639,7 @@ func validateSectionCandidates(ctx context.Context, input validateSectionCandida validatorStarted := time.Now() vResult, err := validator.Validate(ctx, contracts.ValidationRequest{ + StageIndex: &input.Spec.Index, StageID: input.Spec.ID, WorkingTranscript: input.WorkingTranscript, CandidateProposal: eligible, ModuleKey: input.Spec.ModuleKey, diff --git a/internal/framework/structuredoutput/call.go b/internal/framework/structuredoutput/call.go index 4699c9f..7d50197 100644 --- a/internal/framework/structuredoutput/call.go +++ b/internal/framework/structuredoutput/call.go @@ -11,12 +11,14 @@ import ( // DiagnosticContext identifies the logical call without examining rendered text. type DiagnosticContext struct { - Role string `json:"role"` - ModuleKey string `json:"module_key,omitempty"` - ModuleInstance string `json:"module_instance,omitempty"` - SectionIndex *int `json:"section_index,omitempty"` - BatchIndex *int `json:"batch_index,omitempty"` - ValidatorName string `json:"validator_name,omitempty"` + StageIndex *int `json:"stage_index,omitempty"` + StageID *string `json:"stage_id,omitempty"` + Role string `json:"role"` + ModuleKey string `json:"module_key,omitempty"` + ModuleInstance string `json:"module_instance,omitempty"` + SectionIndex *int `json:"section_index,omitempty"` + BatchIndex *int `json:"batch_index,omitempty"` + ValidatorName string `json:"validator_name,omitempty"` } // Request names an embedded prompt and response schema. Profile selection and diff --git a/internal/framework/validators/llm_validators.go b/internal/framework/validators/llm_validators.go index 42a847e..853b6fc 100644 --- a/internal/framework/validators/llm_validators.go +++ b/internal/framework/validators/llm_validators.go @@ -94,6 +94,7 @@ func (v *LLMBackedValidator) Validate(ctx context.Context, req Request) (Result, } operationRef := func(kind outcomes.OperationKind, batchIndex *int) outcomes.Reference { return outcomes.Reference{ + StageIndex: req.StageIndex, StageID: req.StageID, ModuleKey: req.ModuleKey, ModuleInstance: req.ModuleInstance, SectionIndex: sectionIndex, Kind: kind, ValidatorName: v.name, ValidatorOrder: req.ValidatorOrder, BatchIndex: batchIndex, @@ -149,7 +150,7 @@ func (v *LLMBackedValidator) Validate(ctx context.Context, req Request) (Result, completion, err = req.LLMClient.CompleteStructured(ctx, StructuredCompletionRequest{ PromptID: promptMetadata.PromptID, Variables: variables, Schema: responseschema.ValidatorDecisionSetKey, - Context: structuredoutput.DiagnosticContext{Role: "validation", ModuleKey: req.ModuleKey, ModuleInstance: req.ModuleInstance, SectionIndex: req.SectionIndex, BatchIndex: &batch.BatchIndex, ValidatorName: v.name}, + Context: structuredoutput.DiagnosticContext{StageIndex: req.StageIndex, StageID: req.StageID, Role: "validation", ModuleKey: req.ModuleKey, ModuleInstance: req.ModuleInstance, SectionIndex: req.SectionIndex, BatchIndex: &batch.BatchIndex, ValidatorName: v.name}, }, &response) batchIndex := batch.BatchIndex operation := outcomes.Operation{ @@ -165,6 +166,7 @@ func (v *LLMBackedValidator) Validate(ctx context.Context, req Request) (Result, artifacts, diagnosticErr := req.DiagnosticsWriter.WriteInteraction( stage, map[string]any{ + "stage_index": req.StageIndex, "stage_id": req.StageID, "module_key": req.ModuleKey, "validator_name": v.name, "validator_type": v.validatorType, "batch_index": batch.BatchIndex, diff --git a/internal/framework/validators/models.go b/internal/framework/validators/models.go index 72c2a6e..fdc061b 100644 --- a/internal/framework/validators/models.go +++ b/internal/framework/validators/models.go @@ -32,6 +32,8 @@ const ( // Request is the runtime input shared by deterministic validators. type Request struct { + StageIndex *int `json:"stage_index,omitempty"` + StageID *string `json:"stage_id,omitempty"` // ReportDiagnosticError may be called concurrently; it must not affect processing. ReportDiagnosticError func(error) `json:"-"` SectionIndex *int `json:"section_index,omitempty"`