From bbac312b3200edc35fe95bf4fa874db28e914b95 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Tue, 10 Nov 2026 20:28:41 +0000 Subject: [PATCH] Honor cancellation before transcript delivery --- docs/roadmap/remediation.md | 2 + internal/cli/process.go | 3 ++ internal/cli/process_output.go | 37 ++++++++++++---- internal/cli/process_output_test.go | 65 +++++++++++++++++++++++++++++ 4 files changed, 99 insertions(+), 8 deletions(-) diff --git a/docs/roadmap/remediation.md b/docs/roadmap/remediation.md index 213b177..eaca97a 100644 --- a/docs/roadmap/remediation.md +++ b/docs/roadmap/remediation.md @@ -322,6 +322,8 @@ leave reports. Historical references continue to describe their own versions. ## R9 — P2: Honor Cancellation in Normalization-Only Processing +**Status:** Completed. Processing checks cancellation after normalization and before delivery. A controlled empty-pipeline test preserves the existing transcript, finalizes both YAML reports as failed, and retains the cancellation cause. + **Category** Cancellation regression in Stages 8 and 12. diff --git a/internal/cli/process.go b/internal/cli/process.go index d4f864d..21b61ad 100644 --- a/internal/cli/process.go +++ b/internal/cli/process.go @@ -62,6 +62,9 @@ func runProcess(inv commandkit.Invocation, input processInput) commandkit.Outcom warn(debug.WriteNormalizedTranscript(normalized)) warn(debug.WriteNormalizationSummary(summary)) } + if err := inv.Context.Err(); err != nil { + return finishProcess(inv, input, loaded, prepared, reporting.Snapshot{}, nil, err) + } result, runErr := executePipeline(inv.Context, loaded, prepared, normalized, warn) if runErr != nil { return finishProcess(inv, input, loaded, prepared, result.Snapshot, nil, runErr) diff --git a/internal/cli/process_output.go b/internal/cli/process_output.go index 4af7e7a..494451f 100644 --- a/internal/cli/process_output.go +++ b/internal/cli/process_output.go @@ -1,12 +1,15 @@ package cli import ( + "context" + "errors" "fmt" "time" "gitea.maximumdirect.net/eric/audita/v3/internal/core/diagnostics" "gitea.maximumdirect.net/eric/audita/v3/internal/core/redaction" "gitea.maximumdirect.net/eric/audita/v3/internal/core/reporting" + "gitea.maximumdirect.net/eric/audita/v3/internal/framework/outcomes" commandkit "go.maximumdirect.net/commandkit" ) @@ -21,15 +24,12 @@ func finishProcess(inv commandkit.Invocation, input processInput, loaded *proces completed := time.Now().UTC() snapshot.RunID, snapshot.StartedAt, snapshot.CompletedAt = prepared.RunID, prepared.Started, &completed snapshot.TranscriptPath, snapshot.OutputPath = input.Transcript, input.Output.Value + if primary == nil { + primary = inv.Context.Err() + } + snapshot.Status = reporting.StatusCompleted if primary != nil { - snapshot.Status = reporting.StatusFailed - snapshot = failedOutputSnapshot(snapshot, nil) - if *snapshot.Summary.RunErrors == 0 { - snapshot.Errors = append(snapshot.Errors, reporting.ErrorRecord{Scope: "run", Reason: "processing_failure", Message: "Processing failed"}) - snapshot = failedOutputSnapshot(snapshot, nil) - } - } else { - snapshot.Status = reporting.StatusCompleted + snapshot = failedProcessingSnapshot(snapshot, primary, inv.Context.Err()) } reports, err := renderReportBundle(snapshot, secrets...) if err != nil { @@ -45,6 +45,11 @@ func finishProcess(inv commandkit.Invocation, input processInput, loaded *proces } return result } + if canceled := inv.Context.Err(); primary == nil && canceled != nil { + primary = canceled + snapshot = failedProcessingSnapshot(snapshot, canceled, canceled) + reports = failed(nil) + } result := prepared.Prepared.Output.Finish(primary, transcript, reports, failed) if result != nil { return commandkit.Failure(redact(result)).WithCause(result) @@ -60,6 +65,22 @@ func finishProcess(inv commandkit.Invocation, input processInput, loaded *proces return commandkit.Success() } +func failedProcessingSnapshot(snapshot reporting.Snapshot, primary, contextErr error) reporting.Snapshot { + for _, record := range snapshot.Errors { + if record.Scope == "run" { + return failedOutputSnapshot(snapshot, nil) + } + } + record := reporting.ErrorRecord{Scope: "run", Reason: "processing_failure", Message: "Processing failed"} + if errors.Is(primary, context.Canceled) || (contextErr != nil && errors.Is(primary, contextErr)) { + record.Phase = "processing" + record.Reason = string(outcomes.ReasonCanceled) + record.Message = "Processing was canceled." + } + snapshot.Errors = append(snapshot.Errors, record) + return failedOutputSnapshot(snapshot, nil) +} + // failedOutputSnapshot derives the count from the same run records sent to each report. func failedOutputSnapshot(snapshot reporting.Snapshot, failures []diagnostics.FileFailure) reporting.Snapshot { snapshot.Status = reporting.StatusFailed diff --git a/internal/cli/process_output_test.go b/internal/cli/process_output_test.go index 3466122..438d56e 100644 --- a/internal/cli/process_output_test.go +++ b/internal/cli/process_output_test.go @@ -3,6 +3,7 @@ package cli import ( "bytes" "context" + "errors" "os" "path/filepath" "strings" @@ -127,6 +128,70 @@ func TestDeliveryFailureSuppressesCompletionSummary(t *testing.T) { } +func TestCanceledEmptyPipelineLeavesTranscriptAndFinalizesReports(t *testing.T) { + dir := t.TempDir() + for name, contents := range map[string]string{ + "config.yml": "version: 3\npipeline: {stages: []}\n", + "transcript.json": `[{"speaker":"Alice","start":0,"end":1,"text":"hello"}]`, + "output.json": "previous transcript", + } { + if err := os.WriteFile(filepath.Join(dir, name), []byte(contents), 0600); err != nil { + t.Fatal(err) + } + } + input := processInput{ + Config: commandkit.Supplied[string]{Value: filepath.Join(dir, "config.yml"), Set: true}, + Transcript: filepath.Join(dir, "transcript.json"), + Output: commandkit.Supplied[string]{Value: filepath.Join(dir, "output.json"), Set: true}, + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cmd := commandkit.NewCommand(commandkit.CommandSpec[struct{}]{Name: "deliver", Run: func(inv commandkit.Invocation, _ struct{}) commandkit.Outcome { + loaded, err := loadProcessInputs(inv.Context, input, inv.ReportingMode()) + if err != nil { + t.Fatal(err) + } + prepared, err := prepareProcess(inv.Context, input, loaded, inv.Reporter().Output()) + if err != nil { + t.Fatal(err) + } + defer prepared.Prepared.Output.Close() + result, err := executePipeline(inv.Context, loaded, prepared, sourceToCanonicalTranscript(loaded.Source), func(error) {}) + if err != nil { + t.Fatal(err) + } + cancel() + return finishProcess(inv, input, loaded, prepared, result.Snapshot, []byte("new transcript"), nil) + }}) + app, err := commandkit.New(commandkit.Identity{Name: "test"}, cmd) + if err != nil { + t.Fatal(err) + } + var stdout, stderr bytes.Buffer + outcome := app.Execute(ctx, []string{"deliver", "--verbose"}, commandkit.Streams{Out: &stdout, Err: &stderr}) + if outcome.Status == 0 || !errors.Is(outcome.Cause, context.Canceled) || strings.Contains(stderr.String(), "Completed:") { + t.Fatalf("outcome=%+v stderr=%q", outcome, stderr.String()) + } + if data, err := os.ReadFile(input.Output.Value); err != nil || string(data) != "previous transcript" { + t.Fatalf("transcript changed after cancellation: %q, %v", data, err) + } + var errorsDoc reporting.ErrorsDocument + var recoveries reporting.RecoveriesDocument + for name, target := range map[string]any{"errors.yml": &errorsDoc, "recoveries.yml": &recoveries} { + data, err := os.ReadFile(filepath.Join(dir, name)) + if err != nil { + t.Fatal(err) + } + if err := yaml.Unmarshal(data, target); err != nil { + t.Fatal(err) + } + } + if errorsDoc.Status != reporting.StatusFailed || recoveries.Status != reporting.StatusFailed || + len(errorsDoc.Errors) != 1 || errorsDoc.Errors[0].Reason != "canceled" { + t.Fatalf("reports did not capture cancellation: %+v %+v", errorsDoc, recoveries) + } +} + type shortProcessWriter struct{} func (shortProcessWriter) Write(data []byte) (int, error) { return len(data) / 2, nil }