Prepare and finalize output resources safely

2026-11-10 17:03:10 +00:00
parent 8e1e073731
commit 852e79405f
3 changed files with 348 additions and 68 deletions
--- 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.
--- 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 {
--- 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)
+	}
+}