From 852e79405ff003080c072b28313d63d5abad57d5 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Tue, 10 Nov 2026 17:03:10 +0000 Subject: [PATCH] Prepare and finalize output resources safely --- docs/roadmap/implementation.md | 5 + internal/core/diagnostics/output_session.go | 230 +++++++++++++----- .../core/diagnostics/output_session_test.go | 181 +++++++++++++- 3 files changed, 348 insertions(+), 68 deletions(-) diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 336ee73..59ead6f 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -305,6 +305,11 @@ status handoff without duplicating the upstream command grammar suite. ## Stage 7: Prepare and finalize output resources +**Completed.** Output selection and resource preparation are separate; the +session owns temporary transcript cleanup, report finalization, and optional +debug preparation. Preflight and the process handler must connect these APIs in +the next assignment. Focused diagnostics tests passed. + Own `internal/core/diagnostics/output_session.go`, file/path helpers, and focused tests. Use the [output preparation](commandkit.md#output-resource-preparation-during-preflight--resolved) and [failure](commandkit.md#output-failure-behavior--resolved) contracts. diff --git a/internal/core/diagnostics/output_session.go b/internal/core/diagnostics/output_session.go index cb63e10..936fd6a 100644 --- a/internal/core/diagnostics/output_session.go +++ b/internal/core/diagnostics/output_session.go @@ -3,6 +3,7 @@ package diagnostics import ( "errors" "fmt" + "io" "os" "path/filepath" "strings" @@ -23,6 +24,7 @@ type OutputSelection struct { ReportDir string ReportingEnabled bool ProtectedPaths []string + Stdout io.Writer } type ReportBundle struct { @@ -43,14 +45,20 @@ func (f FileFailure) Unwrap() error { return f.Err } // OutputSession owns checked destinations and per-file publication. Filesystem // failures remain typed so CLI can present safe, redacted diagnostics. type OutputSession struct { - transcript string - reports [3]string - enabled bool - replace func(string, []byte) error + transcript string + reports [3]string + enabled bool + replace func(string, []byte) error + temp *os.File + tempPath string + publishPath string + direct bool + stdout io.Writer } -func PrepareOutput(selection OutputSelection) (*OutputSession, error) { - session := &OutputSession{enabled: selection.ReportingEnabled, replace: atomicReplace} +// SelectOutput checks active destinations before any output resources are opened. +func SelectOutput(selection OutputSelection) (*OutputSession, error) { + session := &OutputSession{enabled: selection.ReportingEnabled, replace: atomicReplace, stdout: selection.Stdout} active := make([]string, 0, 4) if selection.TranscriptPath != "" { session.transcript = selection.TranscriptPath @@ -113,14 +121,77 @@ func PrepareOutput(selection OutputSelection) (*OutputSession, error) { } } } - if session.transcript != "" { - if err := preflightTranscript(session.transcript); err != nil { - return nil, err - } + return session, nil +} + +// PrepareOutput validates destinations and acquires the transcript resource. +func PrepareOutput(selection OutputSelection) (*OutputSession, error) { + session, err := SelectOutput(selection) + if err != nil { + return nil, err + } + if err := session.Prepare(); err != nil { + return nil, err } return session, nil } +// Prepare opens a sibling temporary file for regular transcript output. +func (s *OutputSession) Prepare() error { + if s == nil { + return fmt.Errorf("output session is required") + } + if s.transcript == "" || s.temp != nil || s.direct { + return nil + } + info, err := os.Stat(s.transcript) + if err == nil && info.IsDir() { + return fmt.Errorf("transcript output %q is a directory", s.transcript) + } + if err == nil && !info.Mode().IsRegular() { + s.direct = true + return nil + } + if err != nil && !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("transcript output %q: %w", s.transcript, err) + } + s.publishPath, err = canonicalDestination(s.transcript) + if err != nil { + return err + } + s.temp, err = os.CreateTemp(filepath.Dir(s.publishPath), ".audita-output-*") + if err != nil { + return fmt.Errorf("prepare transcript output %q: %w", s.transcript, err) + } + s.tempPath = s.temp.Name() + if info != nil { + if err := s.temp.Chmod(info.Mode().Perm()); err != nil { + _ = s.Close() + return err + } + } + return nil +} + +// Close releases unpublished output resources; caller-owned streams remain open. +func (s *OutputSession) Close() error { + if s == nil { + return nil + } + var errs []error + if s.temp != nil { + errs = append(errs, s.temp.Close()) + s.temp = nil + } + if s.tempPath != "" { + if err := os.Remove(s.tempPath); err != nil && !errors.Is(err, os.ErrNotExist) { + errs = append(errs, err) + } + s.tempPath = "" + } + return errors.Join(errs...) +} + func (s *OutputSession) ReportPaths() []string { if s == nil || !s.enabled { return nil @@ -135,23 +206,44 @@ func (s *OutputSession) TranscriptPath() string { return s.transcript } -// Initialize writes all report destinations before input/model processing. On -// failure it makes one best-effort pass to mark already-created reports failed. +// Initialize publishes each initial report and attempts failed-status reports if any fail. func (s *OutputSession) Initialize(initial, failed ReportBundle) error { if s == nil || !s.enabled { return nil } - created := make([]int, 0, len(s.reports)) + var failures []error for i, payload := range bundleParts(initial) { if err := s.replace(s.reports[i], payload); err != nil { - for _, prior := range created { - _ = s.replace(s.reports[prior], bundleParts(failed)[prior]) - } - return FileFailure{Path: s.reports[i], Kind: "report initialization", Err: err} + failures = append(failures, FileFailure{Path: s.reports[i], Kind: "report initialization", Err: err}) } - created = append(created, i) } - return nil + if len(failures) != 0 { + for i, payload := range bundleParts(failed) { + if err := s.replace(s.reports[i], payload); err != nil { + failures = append(failures, FileFailure{Path: s.reports[i], Kind: "failed report initialization", Err: err}) + } + } + failures = append(failures, s.Close()) + } + return errors.Join(failures...) +} + +// PrepareDebug creates optional initial diagnostics and returns any nonfatal problems. +func (s *OutputSession) PrepareDebug(workDir, runID string, metadata InvocationMetadata, effective []byte, secrets ...string) (*RunDirectory, []error) { + run, err := NewRunDirectory(workDir, runID, secrets...) + if err != nil { + return nil, []error{err} + } + var problems []error + if err := run.WriteInvocationMetadata(metadata); err != nil { + problems = append(problems, err) + } + if effective != nil { + if err := run.WriteEffectiveConfig(effective); err != nil { + problems = append(problems, err) + } + } + return run, problems } // Finalize publishes the transcript first, then errors, recoveries, and Markdown. @@ -162,85 +254,91 @@ func (s *OutputSession) Finalize(transcript []byte, reports ReportBundle, failed return []FileFailure{{Kind: "output finalization", Err: fmt.Errorf("output session is required")}} } failures := make([]FileFailure, 0) - if transcript != nil && s.transcript != "" { + if transcript != nil { if err := s.writeTranscript(transcript); err != nil { failures = append(failures, FileFailure{Path: s.transcript, Kind: "transcript write", Err: err}) } } + if err := s.Close(); err != nil { + failures = append(failures, FileFailure{Path: s.transcript, Kind: "output cleanup", Err: err}) + } if s.enabled { for i, payload := range bundleParts(reports) { if err := s.replace(s.reports[i], payload); err != nil { failures = append(failures, FileFailure{Path: s.reports[i], Kind: "report write", Err: err}) } } - if len(failures) > 0 { + if len(failures) > 0 && failed != nil { for i, payload := range bundleParts(failed(failures)) { - _ = s.replace(s.reports[i], payload) + if err := s.replace(s.reports[i], payload); err != nil { + failures = append(failures, FileFailure{Path: s.reports[i], Kind: "failed report write", Err: err}) + } } } } return failures } -// Stream destinations use ordinary writes; replacing their directory entry would -// destroy the device or pipe. Regular files retain per-file replacement. +// Stream destinations are opened only when delivering the transcript. +// Finish retains an earlier processing error while reporting delivery failures. +func (s *OutputSession) Finish(primary error, transcript []byte, reports ReportBundle, failed func([]FileFailure) ReportBundle) error { + if primary != nil { + transcript = nil + } + failures := s.Finalize(transcript, reports, failed) + errs := make([]error, 0, len(failures)+1) + errs = append(errs, primary) + for _, failure := range failures { + errs = append(errs, failure) + } + return errors.Join(errs...) +} + func (s *OutputSession) writeTranscript(payload []byte) error { - info, err := os.Stat(s.transcript) - if err != nil && !errors.Is(err, os.ErrNotExist) { + if s.transcript == "" { + if s.stdout == nil { + return fmt.Errorf("transcript stdout writer is required") + } + n, err := s.stdout.Write(payload) + if err == nil && n != len(payload) { + return io.ErrShortWrite + } return err } - if err == nil && !info.Mode().IsRegular() { + if s.direct { file, err := os.OpenFile(s.transcript, os.O_WRONLY, 0) if err != nil { return err } - _, writeErr := file.Write(payload) + n, writeErr := file.Write(payload) + if writeErr == nil && n != len(payload) { + writeErr = io.ErrShortWrite + } return errors.Join(writeErr, file.Close()) } - return s.replace(s.transcript, payload) + if s.temp == nil { + return fmt.Errorf("transcript output is not prepared") + } + n, writeErr := s.temp.Write(payload) + if writeErr == nil && n != len(payload) { + writeErr = io.ErrShortWrite + } + closeErr := s.temp.Close() + s.temp = nil + if err := errors.Join(writeErr, closeErr); err != nil { + return err + } + if err := os.Rename(s.tempPath, s.publishPath); err != nil { + return err + } + s.tempPath = "" + return nil } func bundleParts(bundle ReportBundle) [3][]byte { return [3][]byte{bundle.Errors, bundle.Recoveries, bundle.Markdown} } -func preflightTranscript(path string) error { - info, err := os.Stat(path) - if err == nil { - if info.IsDir() { - return fmt.Errorf("transcript output %q is a directory", path) - } - if !info.Mode().IsRegular() { - // Opening a pipe here can block before its reader is ready. - return nil - } - file, openErr := os.OpenFile(path, os.O_WRONLY, 0) - if openErr != nil { - return fmt.Errorf("transcript output %q cannot be opened: %w", path, openErr) - } - if closeErr := file.Close(); closeErr != nil { - return fmt.Errorf("transcript output %q cannot be closed: %w", path, closeErr) - } - } else if !errors.Is(err, os.ErrNotExist) { - return fmt.Errorf("transcript output %q: %w", path, err) - } - parent := filepath.Dir(path) - probe, err := os.CreateTemp(parent, ".audita-probe-*") - if err != nil { - return fmt.Errorf("transcript output parent %q: %w", parent, err) - } - name := probe.Name() - closeErr := probe.Close() - removeErr := os.Remove(name) - if closeErr != nil { - return fmt.Errorf("transcript output probe close: %w", closeErr) - } - if removeErr != nil { - return fmt.Errorf("transcript output probe removal: %w", removeErr) - } - return nil -} - func sameDestination(a, b string) (bool, error) { ca, err := canonicalDestination(a) if err != nil { diff --git a/internal/core/diagnostics/output_session_test.go b/internal/core/diagnostics/output_session_test.go index 4817d37..e8ced65 100644 --- a/internal/core/diagnostics/output_session_test.go +++ b/internal/core/diagnostics/output_session_test.go @@ -1,8 +1,11 @@ package diagnostics import ( + "bytes" "errors" + "io" "os" + "os/exec" "path/filepath" "strings" "testing" @@ -191,8 +194,8 @@ func TestInitializeFailureMarksPriorReportFailedBeforeModelWork(t *testing.T) { if err != nil || string(got) != "failed" { t.Fatalf("prior report not failed: %s %v", got, err) } - if _, err := os.Stat(filepath.Join(dir, MarkdownFile)); !errors.Is(err, os.ErrNotExist) { - t.Fatalf("later report created: %v", err) + if got, err := os.ReadFile(filepath.Join(dir, MarkdownFile)); err != nil || string(got) != "failed" { + t.Fatalf("later report not finalized: %q %v", got, err) } } @@ -255,3 +258,177 @@ func TestDebugRunDirectoryUsesSuppliedIDAndIsRetained(t *testing.T) { t.Fatalf("debug directory removed: %v", err) } } + +func TestSelectionAndAbortPreserveTranscript(t *testing.T) { + dir := t.TempDir() + output := filepath.Join(dir, "output.json") + if err := os.WriteFile(output, []byte("original"), 0600); err != nil { + t.Fatal(err) + } + session, err := SelectOutput(OutputSelection{TranscriptPath: output}) + if err != nil { + t.Fatal(err) + } + if entries, err := os.ReadDir(dir); err != nil || len(entries) != 1 { + t.Fatalf("selection changed directory: %v %v", entries, err) + } + if err := session.Prepare(); err != nil { + t.Fatal(err) + } + if session.temp == nil { + t.Fatal("temporary output not open") + } + temp := session.tempPath + if err := session.Close(); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(temp); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("unpublished file remains: %v", err) + } + if data, err := os.ReadFile(output); err != nil || string(data) != "original" { + t.Fatalf("output changed: %q %v", data, err) + } +} + +func TestStdoutAndDeviceDelivery(t *testing.T) { + var stdout bytes.Buffer + session, err := PrepareOutput(OutputSelection{Stdout: &stdout}) + if err != nil { + t.Fatal(err) + } + if failures := session.Finalize([]byte("transcript"), ReportBundle{}, nil); len(failures) != 0 { + t.Fatal(failures) + } + if stdout.String() != "transcript" { + t.Fatalf("stdout=%q", stdout.String()) + } + session, err = PrepareOutput(OutputSelection{TranscriptPath: os.DevNull}) + if err != nil { + t.Fatal(err) + } + if !session.direct || session.temp != nil { + t.Fatal("device was not deferred") + } + if failures := session.Finalize([]byte("discard"), ReportBundle{}, nil); len(failures) != 0 { + t.Fatal(failures) + } +} + +func TestFinishRetainsPrimaryFailureAndAttemptsReports(t *testing.T) { + dir := t.TempDir() + output := filepath.Join(dir, "transcript.json") + if err := os.WriteFile(output, []byte("original"), 0600); err != nil { + t.Fatal(err) + } + session, err := PrepareOutput(OutputSelection{TranscriptPath: output, ReportDir: dir, ReportingEnabled: true}) + if err != nil { + t.Fatal(err) + } + originalReplace := session.replace + session.replace = func(path string, data []byte) error { + if filepath.Base(path) == ErrorsFile { + return os.ErrPermission + } + return originalReplace(path, data) + } + primary := errors.New("processing failed") + bundle := ReportBundle{Errors: []byte("failed"), Recoveries: []byte("failed"), Markdown: []byte("failed")} + result := session.Finish(primary, []byte("unwanted"), bundle, func([]FileFailure) ReportBundle { return bundle }) + if !errors.Is(result, primary) || !errors.Is(result, os.ErrPermission) { + t.Fatalf("combined error: %v", result) + } + if data, err := os.ReadFile(output); err != nil || string(data) != "original" { + t.Fatalf("output=%q err=%v", data, err) + } + if data, err := os.ReadFile(filepath.Join(dir, MarkdownFile)); err != nil || string(data) != "failed" { + t.Fatalf("remaining report=%q err=%v", data, err) + } +} + +func TestOptionalDebugPreparationReportsProblems(t *testing.T) { + dir := t.TempDir() + session, err := SelectOutput(OutputSelection{}) + if err != nil { + t.Fatal(err) + } + file := filepath.Join(dir, "not-directory") + if err := os.WriteFile(file, []byte("x"), 0600); err != nil { + t.Fatal(err) + } + if run, problems := session.PrepareDebug(file, "run", InvocationMetadata{}, nil); run != nil || len(problems) == 0 { + t.Fatalf("run=%v problems=%v", run, problems) + } + run, problems := session.PrepareDebug(filepath.Join(dir, "debug"), "run", InvocationMetadata{Operation: "process"}, []byte("secret: sensitive\n"), "sensitive") + if len(problems) != 0 { + t.Fatal(problems) + } + contents, err := os.ReadFile(filepath.Join(run.Path(), ArtifactEffectiveConfig)) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(contents), "sensitive") { + t.Fatal("debug artifact contains secret") + } +} + +type shortOutput struct{} + +func (shortOutput) Write(b []byte) (int, error) { return len(b) / 2, nil } + +func TestShortStdoutWriteIsFailure(t *testing.T) { + session, err := PrepareOutput(OutputSelection{Stdout: shortOutput{}}) + if err != nil { + t.Fatal(err) + } + failures := session.Finalize([]byte("transcript"), ReportBundle{}, nil) + if len(failures) != 1 || !errors.Is(failures[0], io.ErrShortWrite) { + t.Fatalf("failures=%v", failures) + } +} + +func TestTranscriptWriteFailureCleansTemporaryResource(t *testing.T) { + dir := t.TempDir() + output := filepath.Join(dir, "out.json") + if err := os.WriteFile(output, []byte("original"), 0600); err != nil { + t.Fatal(err) + } + session, err := PrepareOutput(OutputSelection{TranscriptPath: output}) + if err != nil { + t.Fatal(err) + } + temp := session.tempPath + if err := session.temp.Close(); err != nil { + t.Fatal(err) + } + failures := session.Finalize([]byte("new"), ReportBundle{}, nil) + if len(failures) == 0 { + t.Fatal("closed output accepted") + } + if _, err := os.Stat(temp); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("temporary output remains: %v", err) + } + if data, err := os.ReadFile(output); err != nil || string(data) != "original" { + t.Fatalf("output=%q err=%v", data, err) + } +} + +func TestNamedPipeIsNotOpenedDuringPreparation(t *testing.T) { + tool, err := exec.LookPath("mkfifo") + if err != nil { + t.Skip("mkfifo is unavailable") + } + pipe := filepath.Join(t.TempDir(), "output-pipe") + if output, err := exec.Command(tool, pipe).CombinedOutput(); err != nil { + t.Fatalf("mkfifo: %v %s", err, output) + } + session, err := PrepareOutput(OutputSelection{TranscriptPath: pipe}) + if err != nil { + t.Fatal(err) + } + if !session.direct || session.temp != nil { + t.Fatal("pipe was not deferred") + } + if err := session.Close(); err != nil { + t.Fatal(err) + } +}