Skip report and debug work for disabled outputs

This commit is contained in:
2026-11-10 20:48:32 +00:00
parent a9b7c37a7e
commit 71a9b1f560
11 changed files with 161 additions and 100 deletions

View File

@@ -87,24 +87,31 @@ func prepareProcess(ctx context.Context, input processInput, loaded *processInpu
return nil, err
}
started := time.Now().UTC()
snapshot := reporting.Snapshot{RunID: runID, Status: reporting.StatusInProgress, StartedAt: started, TranscriptPath: input.Transcript, OutputPath: input.Output.Value}
for index, item := range cfg.Pipeline.Stages {
snapshot.StageNames = append(snapshot.StageNames, reporting.StageLabel(index, item.Module, item.ID))
var secrets []string
if cfg.Reporting.Enabled || input.Debug.Value {
secrets = loaded.Runtime.ConfiguredSecrets(ctx)
}
secrets := loaded.Runtime.ConfiguredSecrets(ctx)
initial, err := renderReportBundle(snapshot, secrets...)
if err != nil {
return nil, err
}
failed := func(primary error, failures []diagnostics.FileFailure) diagnostics.ReportBundle {
failedSnapshot := snapshot
completed := time.Now().UTC()
failedSnapshot.CompletedAt = &completed
if primary != nil {
failedSnapshot.Errors = []reporting.ErrorRecord{{Scope: "run", Phase: "initialization", Reason: "canceled", Message: "Processing was canceled."}}
var initial diagnostics.ReportBundle
var failed func(error, []diagnostics.FileFailure) diagnostics.ReportBundle
if cfg.Reporting.Enabled {
snapshot := reporting.Snapshot{RunID: runID, Status: reporting.StatusInProgress, StartedAt: started, TranscriptPath: input.Transcript, OutputPath: input.Output.Value}
for index, item := range cfg.Pipeline.Stages {
snapshot.StageNames = append(snapshot.StageNames, reporting.StageLabel(index, item.Module, item.ID))
}
initial, err = renderReportBundle(snapshot, secrets...)
if err != nil {
return nil, err
}
failed = func(primary error, failures []diagnostics.FileFailure) diagnostics.ReportBundle {
failedSnapshot := snapshot
completed := time.Now().UTC()
failedSnapshot.CompletedAt = &completed
if primary != nil {
failedSnapshot.Errors = []reporting.ErrorRecord{{Scope: "run", Phase: "initialization", Reason: "canceled", Message: "Processing was canceled."}}
}
bundle, _ := renderReportBundle(failedOutputSnapshot(failedSnapshot, failures), secrets...)
return bundle
}
bundle, _ := renderReportBundle(failedOutputSnapshot(failedSnapshot, failures), secrets...)
return bundle
}
var debug *preflight.DebugInput
var warnings []error

View File

@@ -73,7 +73,6 @@ func TestProcessingUsesPerEntryProfilesInOrder(t *testing.T) {
dir := t.TempDir()
files := map[string]string{
"config.yml": `version: 3
reporting: {enabled: false}
glossary_collections:
party:
entries: [{id: alice, name: Alice, category: person}]
@@ -117,6 +116,7 @@ pipeline:
}
inputs := &processInputs{Configuration: loaded, Runtime: runtime, Source: source}
loaded.Config.Diagnostics.WorkDir = dir
loaded.Config.Reporting.StdoutDir = 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)
@@ -159,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, id: failing}]\n",
"config.yml": "version: 3\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"}]`,
}
@@ -184,6 +184,7 @@ func TestRecoverableModelFailureKeepsNormalizedTranscript(t *testing.T) {
t.Fatal(err)
}
inputs := &processInputs{Configuration: loaded, Runtime: runtime, Source: source}
loaded.Config.Reporting.StdoutDir = dir
prepared, err := prepareProcess(context.Background(), processInput{Transcript: filepath.Join(dir, "transcript.json")}, inputs, &bytes.Buffer{})
if err != nil {
t.Fatal(err)

View File

@@ -31,31 +31,39 @@ func finishProcess(inv commandkit.Invocation, input processInput, loaded *proces
if primary != nil {
snapshot = failedProcessingSnapshot(snapshot, primary, inv.Context.Err())
}
reports, err := renderReportBundle(snapshot, secrets...)
if err != nil {
if primary == nil {
primary = err
reportPaths := prepared.Prepared.Output.ReportPaths()
var reports diagnostics.ReportBundle
var failed func([]diagnostics.FileFailure) diagnostics.ReportBundle
if len(reportPaths) > 0 {
var err error
reports, err = renderReportBundle(snapshot, secrets...)
if err != nil {
if primary == nil {
primary = err
}
snapshot.Status = reporting.StatusFailed
}
snapshot.Status = reporting.StatusFailed
}
failed := func(failures []diagnostics.FileFailure) diagnostics.ReportBundle {
result, bundleErr := renderReportBundle(failedOutputSnapshot(snapshot, failures), secrets...)
if bundleErr != nil {
return reports
failed = func(failures []diagnostics.FileFailure) diagnostics.ReportBundle {
result, bundleErr := renderReportBundle(failedOutputSnapshot(snapshot, failures), secrets...)
if bundleErr != nil {
return reports
}
return result
}
return result
}
if canceled := inv.Context.Err(); primary == nil && canceled != nil {
primary = canceled
snapshot = failedProcessingSnapshot(snapshot, canceled, canceled)
reports = failed(nil)
if failed != nil {
reports = failed(nil)
}
}
result := prepared.Prepared.Output.Finish(primary, transcript, reports, failed)
if result != nil {
return commandkit.Failure(safe.String(result.Error())).WithCause(result)
}
_ = reporter.AtEnd().Info(reporting.CompletionLine(snapshot))
for _, path := range prepared.Prepared.Output.ReportPaths() {
for _, path := range reportPaths {
_ = reporter.AtEnd().Info(fmt.Sprintf("Report: %s", safe.String(path)))
}
if prepared.Prepared.Debug != nil {

View File

@@ -20,14 +20,20 @@ import (
func executePipeline(ctx context.Context, loaded *processInputs, prepared *preparedProcess, normalized *schema.Transcript, warn func(error)) (processResult, error) {
cfg := &loaded.Configuration.Config
specs := contracts.ResolveModuleRunSpecs(cfg)
observer := llm.NewObservationCollector()
debug := prepared.Prepared.Debug
var observer *llm.ObservationCollector
var hooks llm.CallHooks
if debug != nil {
observer = llm.NewObservationCollector()
hooks = observer.Hooks()
}
for i := range specs {
item := &specs[i]
proposal, err := loaded.Runtime.NewRoleClient("proposal", *item.Settings.LLM.ProposalProfile, item.Settings.LLM, observer.Hooks())
proposal, err := loaded.Runtime.NewRoleClient("proposal", *item.Settings.LLM.ProposalProfile, item.Settings.LLM, hooks)
if err != nil {
return processResult{}, fmt.Errorf("%s proposal client: %w", item.InstanceName, err)
}
validation, err := loaded.Runtime.NewRoleClient("validation", *item.Settings.LLM.ValidationProfile, item.Settings.LLM, observer.Hooks())
validation, err := loaded.Runtime.NewRoleClient("validation", *item.Settings.LLM.ValidationProfile, item.Settings.LLM, hooks)
if err != nil {
return processResult{}, fmt.Errorf("%s validation client: %w", item.InstanceName, err)
}
@@ -39,7 +45,7 @@ func executePipeline(ctx context.Context, loaded *processInputs, prepared *prepa
ProposalDiagnosticsDir: debugPath, ValidationDiagnosticsDir: debugPath,
CaptureCandidates: debugPath != "", ReportDiagnosticError: warn,
})
if debug := prepared.Prepared.Debug; debug != nil {
if debug != nil {
warn(debug.WriteJSONArtifact(diagnostics.ArtifactLLMExecution, processreport.BuildLLMExecution(observer.Snapshot())))
warn(debug.WriteJSONArtifact(diagnostics.ArtifactCorrectionLedger, processreport.BuildCorrectionLedger(processreport.CorrectionLedgerInput{RunDirectoryPath: debug.Path(), RunOutput: &output})))
}
@@ -61,6 +67,7 @@ func executePipeline(ctx context.Context, loaded *processInputs, prepared *prepa
}
failures = append(failures, failure)
}
snapshot := processreport.Build(processreport.BuildInput{RunID: prepared.RunID, Status: reporting.StatusInProgress, StartedAt: prepared.Started, ModuleSpecs: specs, RunOutput: &output, RunFailures: failures})
snapshot := processreport.Build(processreport.BuildInput{RunID: prepared.RunID, Status: reporting.StatusInProgress, StartedAt: prepared.Started,
ModuleSpecs: specs, RunOutput: &output, RunFailures: failures, SummaryOnly: !cfg.Reporting.Enabled})
return processResult{Transcript: output.FinalTranscript, Snapshot: snapshot}, runErr
}

View File

@@ -69,26 +69,29 @@ func (c *RoleClient) CompleteStructured(parent context.Context, req structuredou
}
terminalMu.Unlock()
}()
record := structuredoutput.CallRecord{CallID: c.runtime.nextCallID.Add(1), Context: copyContext(req.Context), Started: time.Now().UTC()}
record.Context.Role = c.role
record.Prompt = structuredoutput.PromptMetadata{PromptID: req.PromptID, ProfileID: c.profileID}
if c.hooks.OnCall != nil {
c.hooks.OnCall(record)
var record structuredoutput.CallRecord
if c.hooks.OnCall != nil || c.hooks.OnEventWithCall != nil {
record.CallID = c.runtime.nextCallID.Add(1)
record.Context = copyContext(req.Context)
record.Context.Role = c.role
}
defer func() {
record.Finished = time.Now().UTC()
if err != nil {
var failure *structuredoutput.Failure
if errors.As(err, &failure) && failure.Reason == outcomes.ReasonInvalidOutputExhausted {
record.Failure = structuredoutput.FailureMalformedOutput
} else {
record.Failure = structuredoutput.FailureRuntime
if onCall := c.hooks.OnCall; onCall != nil {
record.Started = time.Now().UTC()
record.Prompt = structuredoutput.PromptMetadata{PromptID: req.PromptID, ProfileID: c.profileID}
onCall(record)
defer func() {
record.Finished = time.Now().UTC()
if err != nil {
var failure *structuredoutput.Failure
if errors.As(err, &failure) && failure.Reason == outcomes.ReasonInvalidOutputExhausted {
record.Failure = structuredoutput.FailureMalformedOutput
} else {
record.Failure = structuredoutput.FailureRuntime
}
}
}
if c.hooks.OnCall != nil {
c.hooks.OnCall(record)
}
}()
onCall(record)
}()
}
if parent.Err() != nil {
return response, classifiedFailure(parent, ctx, parent.Err(), nil)
}

View File

@@ -29,6 +29,7 @@ type BuildInput struct {
ModuleSpecs []contracts.ModuleRunSpec
RunOutput *runner.RunOutput
RunFailures []RunFailure
SummaryOnly bool
}
// Build maps processing facts once into an immutable presentation snapshot.
@@ -36,20 +37,28 @@ 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,
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] = reporting.StageLabel(spec.Index, spec.ModuleKey, spec.ID)
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 := 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
})
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 {
@@ -65,12 +74,14 @@ func Build(input BuildInput) reporting.Snapshot {
interrupted++
}
corrections += unit.CorrectionsApplied
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...)
sort.SliceStable(operations, func(i, j int) bool { return operationLess(operations[i], operations[j]) })
failures := make([]reporting.Failure, 0)
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++
@@ -80,18 +91,31 @@ func Build(input BuildInput) reporting.Snapshot {
}
if operation.Succeeded && len(operation.RecoveryReasons) > 0 {
recoveryCount++
row.RecoveryCount++
unitRecoveries++
if len(operation.Filtered) > 0 && operation.Reference.Kind == outcomes.OperationProposal {
filteringResponses++
}
snapshot.Recoveries = append(snapshot.Recoveries, mapRecovery(unit, operation))
if details {
snapshot.Recoveries = append(snapshot.Recoveries, mapRecovery(unit, operation))
}
}
if operation.FailureReason != "" && isRecoverableReason(operation.FailureReason) {
failures = append(failures, mapFailure(operation))
unitHasError = true
if details {
failures = append(failures, mapFailure(operation))
}
}
}
if len(failures) > 0 {
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,

View File

@@ -27,8 +27,14 @@ func TestBuildPartialUnitsAndCombinedRecovery(t *testing.T) {
{ModuleIndex: 1, ModuleKey: "grammar", ModuleInstance: "grammar_2", Section: contracts.SectionMetadata{Index: 0, StartSegmentID: 1, EndSegmentID: 2}, State: outcomes.WorkNotStarted},
{ModuleIndex: 0, ModuleKey: "grammar", ModuleInstance: "grammar_1", Section: contracts.SectionMetadata{Index: 0, StartSegmentID: 1, EndSegmentID: 2}, State: outcomes.WorkCompleted, CorrectionsApplied: 1, Operations: []outcomes.Operation{validation, proposal, inputFailure}},
}}
snapshot := Build(BuildInput{RunID: "run-1", Status: reporting.StatusCompleted, StartedAt: started, CompletedAt: &completed,
ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "grammar", InstanceName: "grammar_1"}, {ModuleKey: "grammar", InstanceName: "grammar_2"}}, RunOutput: output})
input := BuildInput{RunID: "run-1", Status: reporting.StatusCompleted, StartedAt: started, CompletedAt: &completed,
ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "grammar", InstanceName: "grammar_1"}, {ModuleKey: "grammar", InstanceName: "grammar_2"}}, RunOutput: output}
snapshot := Build(input)
input.SummaryOnly = true
summary := Build(input)
if !reflect.DeepEqual(snapshot.Summary, summary.Summary) || len(summary.Errors)+len(summary.Recoveries)+len(summary.WorkUnits) != 0 {
t.Fatalf("summary-only snapshot changed counts or retained details: %+v", summary)
}
if snapshot.Summary.WorkUnitsTotal == nil || *snapshot.Summary.WorkUnitsTotal != 2 || *snapshot.Summary.WorkUnitsWithErrors != 1 || *snapshot.Summary.Recoveries != 1 || *snapshot.Summary.ProposalResponsesWithFiltering != 1 || *snapshot.Summary.LLMOperations != 2 {
t.Fatalf("summary: %+v", snapshot.Summary)
}

View File

@@ -123,18 +123,6 @@ func GenerateCandidates(ctx context.Context, req Request) (Result, error) {
Counts: completion.Execution,
}
requestMetadata := map[string]any{
"stage_index": req.StageIndex, "stage_id": req.StageID,
"module_key": req.ModuleKey,
"module_instance": req.ModuleInstance,
"replacement_policy": req.ReplacementPolicy,
"section": req.Section,
"start_index": req.StartIndex,
"segment_ids": segmentIDs(req.WorkingTranscript),
"prompt": completion.Prompt,
}
requestMetadata["response_schema"] = responseschema.MustLookup(responseschema.CorrectionSetKey).DiagnosticsMap()
var writer InteractionDiagnosticsWriter
if req.DiagnosticsWriter != nil {
writer = req.DiagnosticsWriter
@@ -154,6 +142,17 @@ func GenerateCandidates(ctx context.Context, req Request) (Result, error) {
}
if writer != nil {
requestMetadata := map[string]any{
"stage_index": req.StageIndex, "stage_id": req.StageID,
"module_key": req.ModuleKey,
"module_instance": req.ModuleInstance,
"replacement_policy": req.ReplacementPolicy,
"section": req.Section,
"start_index": req.StartIndex,
"segment_ids": segmentIDs(req.WorkingTranscript),
"prompt": completion.Prompt,
}
requestMetadata["response_schema"] = responseschema.MustLookup(responseschema.CorrectionSetKey).DiagnosticsMap()
var diagnosticErr error
artifacts, diagnosticErr = writer.WriteInteraction(
stage,

View File

@@ -124,13 +124,7 @@ func (v *LLMBackedValidator) Validate(ctx context.Context, req Request) (Result,
var response LLMValidationResponse
var completion StructuredCompletionResponse
responseSchema := responseschema.MustLookup(responseschema.ValidatorDecisionSetKey)
promptMetadata := validatorPromptMetadata(v.validatorType)
stageInstance := req.ModuleInstance
if req.SectionIndex != nil {
stageInstance = fmt.Sprintf("%s:section-%04d", stageInstance, *req.SectionIndex)
}
stage := stagename.ValidatorBatch(stageInstance, v.name, batch.BatchIndex)
completion, err = req.LLMClient.CompleteStructured(ctx, StructuredCompletionRequest{
PromptID: promptMetadata.PromptID,
Variables: variables, Schema: responseschema.ValidatorDecisionSetKey,
@@ -142,10 +136,20 @@ func (v *LLMBackedValidator) Validate(ctx context.Context, req Request) (Result,
Counts: completion.Execution,
}
writeInteraction := func(responsePayload any, callErr error) InteractionArtifacts {
writeInteraction := func(reconciled *validationReconciliation, callErr error) InteractionArtifacts {
if req.DiagnosticsWriter == nil {
return InteractionArtifacts{}
}
stageInstance := req.ModuleInstance
if req.SectionIndex != nil {
stageInstance = fmt.Sprintf("%s:section-%04d", stageInstance, *req.SectionIndex)
}
stage := stagename.ValidatorBatch(stageInstance, v.name, batch.BatchIndex)
responseSchema := responseschema.MustLookup(responseschema.ValidatorDecisionSetKey)
responsePayload := map[string]any{"decoded": response, "final_content": completion.FinalContent, "model": completion.Model}
if reconciled != nil {
responsePayload["reconciliation"] = *reconciled
}
var diagnosticErr error
artifacts, diagnosticErr := req.DiagnosticsWriter.WriteInteraction(
stage,
@@ -168,7 +172,7 @@ func (v *LLMBackedValidator) Validate(ctx context.Context, req Request) (Result,
return artifacts
}
if err != nil {
writeInteraction(map[string]any{"decoded": response, "final_content": completion.FinalContent, "model": completion.Model}, err)
writeInteraction(nil, err)
var failure *structuredoutput.Failure
if errors.As(err, &failure) {
operation.FailureReason = failure.Reason
@@ -186,7 +190,7 @@ func (v *LLMBackedValidator) Validate(ctx context.Context, req Request) (Result,
}
reconciled := reconcileLLMResponse(batch.Items, response)
artifacts := writeInteraction(map[string]any{"decoded": response, "final_content": completion.FinalContent, "model": completion.Model, "reconciliation": reconciled}, nil)
artifacts := writeInteraction(&reconciled, nil)
if len(reconciled.UnavailableIndices) > 0 {
operation.FailureReason = outcomes.ReasonValidationDecisionsInvalid
operation.ProposalIndices = reconciled.UnavailableIndices