Connect processing preflight and output reporting
This commit is contained in:
@@ -344,6 +344,12 @@ retain already published output and primary failure details.
|
||||
|
||||
## Stage 8: Wire processing, common preflight, and reporting modes
|
||||
|
||||
**Completed.** Processing now acquires typed inputs, prepares outputs, and
|
||||
finalizes through CommandKit. The `executePipeline` handoff is intentionally
|
||||
implemented with the ordered runner work below; until then the application and
|
||||
unconverted tests do not build. The handler was checked with a controlled runner
|
||||
boundary; later readiness checks and assembled validation remain assigned below.
|
||||
|
||||
Own `internal/cli/process.go` and the new `internal/framework/preflight`.
|
||||
Use the [named preflight boundary](glossary-stages.md#empty-collections-and-audita-preflight--resolved)
|
||||
and the [CommandKit reporting decisions](commandkit.md#debug-reporting-and-artifact-capture--resolved).
|
||||
|
||||
@@ -1,357 +1,81 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/core/chunking"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/core/config"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/core/diagnostics"
|
||||
coreio "gitea.maximumdirect.net/eric/audita/v3/internal/core/io"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/core/normalization"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/core/outputschema"
|
||||
"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/core/schema"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/contracts"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/llm"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/modules"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/outcomes"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/processreport"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/runner"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/structuredoutput"
|
||||
commandkit "go.maximumdirect.net/commandkit"
|
||||
)
|
||||
|
||||
type processState struct {
|
||||
cfg config.Config
|
||||
runtime *llm.Runtime
|
||||
specs []contracts.ModuleRunSpec
|
||||
runID string
|
||||
started time.Time
|
||||
transcriptPath string
|
||||
glossaryPath string
|
||||
outputPath string
|
||||
selected selectedConfig
|
||||
overlay string
|
||||
ctx context.Context
|
||||
stdout io.Writer
|
||||
stderr io.Writer
|
||||
session *diagnostics.OutputSession
|
||||
debugDir *diagnostics.RunDirectory
|
||||
debugWarnings []string
|
||||
debugMu sync.Mutex
|
||||
runOutput *runner.RunOutput
|
||||
// processResult carries the runner's transcript and ordinary report facts.
|
||||
type processResult struct {
|
||||
Transcript *schema.Transcript
|
||||
Snapshot reporting.Snapshot
|
||||
}
|
||||
|
||||
func generateRunID() (string, error) {
|
||||
var random [12]byte
|
||||
if _, err := rand.Read(random[:]); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return "run-" + hex.EncodeToString(random[:]), nil
|
||||
}
|
||||
|
||||
func (s *processState) snapshot(status string, completed *time.Time, failures []processreport.RunFailure) reporting.Snapshot {
|
||||
return processreport.Build(processreport.BuildInput{RunID: s.runID, Status: status, StartedAt: s.started,
|
||||
CompletedAt: completed, TranscriptPath: s.transcriptPath, GlossaryPath: s.glossaryPath,
|
||||
OutputPath: s.outputPath, ModuleSpecs: s.specs, RunOutput: s.runOutput, RunFailures: failures})
|
||||
}
|
||||
|
||||
func (s *processState) bundle(snapshot reporting.Snapshot) (diagnostics.ReportBundle, error) {
|
||||
if !s.cfg.Reporting.Enabled {
|
||||
return diagnostics.ReportBundle{}, nil
|
||||
}
|
||||
secrets := llm.ConfiguredSecrets(&s.cfg)
|
||||
errorsYAML, err := reporting.YAML(snapshot.ErrorsDocument(), secrets...)
|
||||
func runProcess(inv commandkit.Invocation, input processInput) commandkit.Outcome {
|
||||
loaded, err := loadProcessInputs(inv.Context, input, inv.ReportingMode())
|
||||
if err != nil {
|
||||
return diagnostics.ReportBundle{}, err
|
||||
if inv.Context.Err() != nil || llm.IsAssetError(err) {
|
||||
return commandkit.Failure("Could not prepare processing.").WithCause(err)
|
||||
}
|
||||
return commandkit.Usage(err.Error()).WithCause(err)
|
||||
}
|
||||
recoveriesYAML, err := reporting.YAML(snapshot.RecoveriesDocument(), secrets...)
|
||||
prepared, err := prepareProcess(inv.Context, input, loaded, inv.Reporter().Output())
|
||||
if err != nil {
|
||||
return diagnostics.ReportBundle{}, err
|
||||
return commandkit.Failure(loaded.Runtime.RedactError(inv.Context, err)).WithCause(err)
|
||||
}
|
||||
return diagnostics.ReportBundle{Errors: errorsYAML, Recoveries: recoveriesYAML,
|
||||
Markdown: reporting.Markdown(snapshot, secrets...)}, nil
|
||||
}
|
||||
|
||||
func (s *processState) debugWrite(label string, write func(*diagnostics.RunDirectory) error) {
|
||||
if s.debugDir == nil {
|
||||
return
|
||||
defer prepared.Prepared.Output.Close()
|
||||
var warningsMu sync.Mutex
|
||||
warn := func(err error) {
|
||||
if err == nil {
|
||||
return
|
||||
}
|
||||
warningsMu.Lock()
|
||||
prepared.Prepared.Warnings = append(prepared.Prepared.Warnings, err)
|
||||
warningsMu.Unlock()
|
||||
}
|
||||
if err := write(s.debugDir); err != nil {
|
||||
s.reportDiagnosticError(fmt.Errorf("%s: %w", label, err))
|
||||
debug := prepared.Prepared.Debug
|
||||
if debug != nil {
|
||||
warn(debug.WriteSourceTranscript(loaded.Source, loaded.RawTranscript))
|
||||
}
|
||||
}
|
||||
|
||||
func (s *processState) startDebug() {
|
||||
if !s.cfg.Diagnostics.Debug {
|
||||
return
|
||||
}
|
||||
dir, err := diagnostics.NewRunDirectory(s.cfg.Diagnostics.WorkDir, s.runID, llm.ConfiguredSecrets(&s.cfg)...)
|
||||
if err != nil {
|
||||
s.reportDiagnosticError(fmt.Errorf("directory: %w", err))
|
||||
return
|
||||
}
|
||||
s.debugDir = dir
|
||||
s.debugWrite("invocation", func(d *diagnostics.RunDirectory) error {
|
||||
return d.WriteInvocationMetadata(diagnostics.InvocationMetadata{
|
||||
Operation: "process", TranscriptPath: s.transcriptPath, GlossaryPath: s.glossaryPath,
|
||||
OutputPath: s.outputPath, ConfigPath: s.selected.Path, ConfigSource: s.selected.Source,
|
||||
Overlay: s.overlay, ConfigVersion: &s.cfg.Version, TranscriptDescription: s.cfg.Context.Description,
|
||||
Modules: append([]string(nil), s.cfg.Pipeline.Modules...), RunID: s.runID, StartedAt: s.started,
|
||||
})
|
||||
cfg := &loaded.Configuration.Config
|
||||
normalizer := normalization.NewNormalizer(normalization.NormalizationConfig{
|
||||
MaxSegmentGap: cfg.Normalization.MaxSegmentGap, EllipsisGap: cfg.Normalization.EllipsisGap,
|
||||
MaxSegmentDuration: cfg.Normalization.MaxSegmentDuration, MaxSegmentTokens: cfg.Normalization.MaxSegmentTokens,
|
||||
})
|
||||
var effective bytes.Buffer
|
||||
if err := s.selected.Result.WriteEffective(&effective); err == nil {
|
||||
s.debugWrite("effective settings", func(d *diagnostics.RunDirectory) error { return d.WriteEffectiveConfig(effective.Bytes()) })
|
||||
} else {
|
||||
s.reportDiagnosticError(fmt.Errorf("effective settings: %w", err))
|
||||
normalized, summary := normalizer.Normalize(sourceToCanonicalTranscript(loaded.Source))
|
||||
if debug != nil {
|
||||
warn(debug.WriteNormalizedTranscript(normalized))
|
||||
warn(debug.WriteNormalizationSummary(summary))
|
||||
}
|
||||
if origins, err := llm.ExplanationYAML(s.selected.Result); err == nil {
|
||||
s.debugWrite("configuration origins", func(d *diagnostics.RunDirectory) error {
|
||||
return d.WriteYAMLArtifact(diagnostics.ArtifactConfigOrigins, origins)
|
||||
})
|
||||
} else {
|
||||
s.reportDiagnosticError(fmt.Errorf("configuration origins: %w", err))
|
||||
result, runErr := executePipeline(inv.Context, loaded, prepared, normalized, warn)
|
||||
if runErr != nil {
|
||||
return finishProcess(inv, input, loaded, prepared, result.Snapshot, nil, runErr)
|
||||
}
|
||||
if profiles, err := s.runtime.SelectedProfilesYAML(); err == nil {
|
||||
s.debugWrite("selected profiles", func(d *diagnostics.RunDirectory) error {
|
||||
return d.WriteYAMLArtifact(diagnostics.ArtifactSelectedProfiles, profiles)
|
||||
})
|
||||
} else {
|
||||
s.reportDiagnosticError(fmt.Errorf("selected profiles: %w", err))
|
||||
encoder, err := outputschema.Resolve(cfg.Output.Schema)
|
||||
if err != nil {
|
||||
return finishProcess(inv, input, loaded, prepared, result.Snapshot, nil, err)
|
||||
}
|
||||
if result.Transcript == nil {
|
||||
return finishProcess(inv, input, loaded, prepared, result.Snapshot, nil, fmt.Errorf("processing returned no transcript"))
|
||||
}
|
||||
payload, err := encoder.Encoder(result.Transcript)
|
||||
if err != nil {
|
||||
return finishProcess(inv, input, loaded, prepared, result.Snapshot, nil, err)
|
||||
}
|
||||
return finishProcess(inv, input, loaded, prepared, result.Snapshot, payload, nil)
|
||||
}
|
||||
|
||||
func (s *processState) debugPath() string {
|
||||
if s.debugDir == nil {
|
||||
// DebugPath returns the retained artifact directory, if its preparation succeeded.
|
||||
func (p *preparedProcess) DebugPath() string {
|
||||
if p == nil || p.Prepared == nil || p.Prepared.Debug == nil {
|
||||
return ""
|
||||
}
|
||||
return s.debugDir.Path()
|
||||
}
|
||||
|
||||
func (s *processState) execute() int {
|
||||
transcriptBytes, err := coreio.ReadRequiredFile(s.transcriptPath, "transcript")
|
||||
if err != nil {
|
||||
return s.fail("input", outcomes.ReasonInternalFailure, err)
|
||||
}
|
||||
glossaryBytes, err := coreio.ReadRequiredFile(s.glossaryPath, "glossary")
|
||||
if err != nil {
|
||||
return s.fail("input", outcomes.ReasonInternalFailure, err)
|
||||
}
|
||||
source, err := schema.ParseSourceTranscriptJSON(transcriptBytes)
|
||||
if err != nil {
|
||||
return s.fail("input", outcomes.ReasonInputInvalid, err)
|
||||
}
|
||||
glossary, err := schema.ParseGlossaryYAML(glossaryBytes)
|
||||
if err != nil {
|
||||
return s.fail("input", outcomes.ReasonInputInvalid, err)
|
||||
}
|
||||
s.debugWrite("source transcript", func(d *diagnostics.RunDirectory) error { return d.WriteSourceTranscript(source, transcriptBytes) })
|
||||
normalizer := normalization.NewNormalizer(normalization.NormalizationConfig{
|
||||
MaxSegmentGap: s.cfg.Normalization.MaxSegmentGap, EllipsisGap: s.cfg.Normalization.EllipsisGap,
|
||||
MaxSegmentDuration: s.cfg.Normalization.MaxSegmentDuration, MaxSegmentTokens: s.cfg.Normalization.MaxSegmentTokens,
|
||||
})
|
||||
normalized, normSummary := normalizer.Normalize(sourceToCanonicalTranscript(source))
|
||||
s.debugWrite("normalized transcript", func(d *diagnostics.RunDirectory) error { return d.WriteNormalizedTranscript(normalized) })
|
||||
s.debugWrite("normalization summary", func(d *diagnostics.RunDirectory) error { return d.WriteNormalizationSummary(normSummary) })
|
||||
if err := s.ctx.Err(); err != nil {
|
||||
return s.fail("processing", outcomes.ReasonCanceled, err)
|
||||
}
|
||||
observations := llm.NewObservationCollector()
|
||||
proposalClient, err := s.runtime.NewRoleClient("proposal", observations.Hooks())
|
||||
if err != nil {
|
||||
return s.fail("processing", outcomes.ReasonSetupFailure, err)
|
||||
}
|
||||
validationClient, err := s.runtime.NewRoleClient("validation", observations.Hooks())
|
||||
if err != nil {
|
||||
return s.fail("processing", outcomes.ReasonSetupFailure, err)
|
||||
}
|
||||
factory := modules.NewFactory(modules.Dependencies{Config: &s.cfg, Glossary: glossary, DiagnosticsDir: s.debugPath()})
|
||||
result, runErr := runner.New(factory).Run(s.ctx, runner.RunInput{
|
||||
Config: &s.cfg, Transcript: normalized, Glossary: glossary, ModuleSpecs: s.specs,
|
||||
ProposalLLMClient: proposalClient, ProposalDiagnosticsDir: s.debugPath(),
|
||||
ValidationLLMClient: validationClient, ValidationDiagnosticsDir: s.debugPath(),
|
||||
CaptureCandidates: s.debugDir != nil,
|
||||
ReportDiagnosticError: s.reportDiagnosticError,
|
||||
})
|
||||
result.Observations = observations.Snapshot()
|
||||
result.ObservationsAvailable = true
|
||||
result.LLMSummary = llm.Summarize(result.Observations)
|
||||
s.runOutput = &result
|
||||
s.writeExecutionEvidence()
|
||||
if len(result.Sections) > 0 || runErr == nil {
|
||||
s.writeChunkSummary(result.Sections)
|
||||
}
|
||||
if runErr != nil {
|
||||
return s.fail("processing", fatalReason(s.ctx, runErr), runErr)
|
||||
}
|
||||
encoder, err := outputschema.Resolve(s.cfg.Output.Schema)
|
||||
if err != nil {
|
||||
return s.fail("encoding", outcomes.ReasonSetupFailure, err)
|
||||
}
|
||||
outputBytes, err := encoder.Encoder(result.FinalTranscript)
|
||||
if err != nil {
|
||||
return s.fail("encoding", outcomes.ReasonInternalFailure, err)
|
||||
}
|
||||
return s.finish(outputBytes)
|
||||
}
|
||||
|
||||
func (s *processState) writeChunkSummary(inventory []contracts.SectionMetadata) {
|
||||
if s.debugDir == nil {
|
||||
return
|
||||
}
|
||||
sections := make([]chunking.Section, len(inventory))
|
||||
for i, metadata := range inventory {
|
||||
sections[i] = chunking.Section{Index: metadata.Index, StartSegmentID: metadata.StartSegmentID,
|
||||
EndSegmentID: metadata.EndSegmentID, EstimatedTokens: metadata.EstimatedTokens,
|
||||
Segments: make([]schema.Segment, metadata.SegmentCount)}
|
||||
}
|
||||
config := chunking.ChunkingConfig{TargetSections: s.cfg.Chunking.TargetSections,
|
||||
MaxSectionTokens: s.cfg.Chunking.MaxSectionTokens, MinSectionTokens: s.cfg.Chunking.MinSectionTokens}
|
||||
summary := chunking.ComputeDetailedSummary(sections, config)
|
||||
s.debugWrite("chunking summary", func(d *diagnostics.RunDirectory) error { return d.WriteChunkingSummary(&summary) })
|
||||
}
|
||||
|
||||
func (s *processState) writeExecutionEvidence() {
|
||||
if s.debugDir == nil || s.runOutput == nil {
|
||||
return
|
||||
}
|
||||
output := s.runOutput
|
||||
s.debugWrite("model execution", func(d *diagnostics.RunDirectory) error {
|
||||
return d.WriteJSONArtifact(diagnostics.ArtifactLLMExecution, processreport.BuildLLMExecution(output.Observations))
|
||||
})
|
||||
s.debugWrite("utilization", func(d *diagnostics.RunDirectory) error {
|
||||
return d.WriteJSONArtifact(diagnostics.ArtifactUtilizationSummary, processreport.BuildUtilization(output, s.cfg.Backends))
|
||||
})
|
||||
s.debugWrite("correction ledger", func(d *diagnostics.RunDirectory) error {
|
||||
return d.WriteJSONArtifact(diagnostics.ArtifactCorrectionLedger, processreport.BuildCorrectionLedger(processreport.CorrectionLedgerInput{
|
||||
RunDirectoryPath: d.Path(), RunOutput: output}))
|
||||
})
|
||||
}
|
||||
|
||||
func fatalReason(ctx context.Context, err error) outcomes.ReasonCode {
|
||||
if ctx.Err() != nil || errors.Is(err, context.Canceled) {
|
||||
return outcomes.ReasonCanceled
|
||||
}
|
||||
var failure *structuredoutput.Failure
|
||||
if errors.As(err, &failure) && failure.Reason != "" {
|
||||
return failure.Reason
|
||||
}
|
||||
return outcomes.ReasonInternalFailure
|
||||
}
|
||||
|
||||
func (s *processState) fail(phase string, reason outcomes.ReasonCode, cause error) int {
|
||||
s.debugWrite("error log", func(d *diagnostics.RunDirectory) error {
|
||||
return d.WriteErrorLog(phase + ": " + llm.RedactError(cause, s.cfg))
|
||||
})
|
||||
failures := []processreport.RunFailure{{Phase: phase, Reason: reason}}
|
||||
completed := time.Now().UTC()
|
||||
base, err := s.bundle(s.snapshot(reporting.StatusFailed, &completed, failures))
|
||||
if err != nil {
|
||||
fmt.Fprintf(s.stderr, "audita process: %s\n", llm.RedactError(err, s.cfg))
|
||||
}
|
||||
writeFailures := s.session.Finalize(nil, base, func(extra []diagnostics.FileFailure) diagnostics.ReportBundle {
|
||||
all := append(append([]processreport.RunFailure(nil), failures...), mapWriteFailures(extra)...)
|
||||
bundle, renderErr := s.bundle(s.snapshot(reporting.StatusFailed, &completed, all))
|
||||
if renderErr != nil {
|
||||
return base
|
||||
}
|
||||
return bundle
|
||||
})
|
||||
fmt.Fprintf(s.stderr, "audita process: %s\n", llm.RedactError(cause, s.cfg))
|
||||
s.printFailures(writeFailures)
|
||||
s.printDebugWarnings()
|
||||
if s.debugDir != nil {
|
||||
fmt.Fprintf(s.stderr, "audita process: diagnostics: %s\n", s.safePath(s.debugDir.Path()))
|
||||
}
|
||||
return 1
|
||||
}
|
||||
|
||||
func (s *processState) finish(outputBytes []byte) int {
|
||||
completed := time.Now().UTC()
|
||||
snapshot := s.snapshot(reporting.StatusCompleted, &completed, nil)
|
||||
complete, err := s.bundle(snapshot)
|
||||
if err != nil {
|
||||
return s.fail("report_write", outcomes.ReasonInternalFailure, err)
|
||||
}
|
||||
if s.session.TranscriptPath() == "" {
|
||||
if _, err := s.stdout.Write(outputBytes); err != nil {
|
||||
return s.fail("transcript_write", outcomes.ReasonInternalFailure, err)
|
||||
}
|
||||
outputBytes = nil
|
||||
}
|
||||
writeFailures := s.session.Finalize(outputBytes, complete, func(extra []diagnostics.FileFailure) diagnostics.ReportBundle {
|
||||
failed, renderErr := s.bundle(s.snapshot(reporting.StatusFailed, &completed, mapWriteFailures(extra)))
|
||||
if renderErr != nil {
|
||||
return diagnostics.ReportBundle{}
|
||||
}
|
||||
return failed
|
||||
})
|
||||
if len(writeFailures) > 0 {
|
||||
s.printFailures(writeFailures)
|
||||
s.printDebugWarnings()
|
||||
if s.debugDir != nil {
|
||||
fmt.Fprintf(s.stderr, "audita process: diagnostics: %s\n", s.safePath(s.debugDir.Path()))
|
||||
}
|
||||
return 1
|
||||
}
|
||||
fmt.Fprintln(s.stderr, reporting.CompletionLine(snapshot))
|
||||
for _, path := range s.session.ReportPaths() {
|
||||
fmt.Fprintf(s.stderr, "Report: %s\n", s.safePath(path))
|
||||
}
|
||||
if s.debugDir != nil {
|
||||
fmt.Fprintf(s.stderr, "Debug: %s\n", s.safePath(s.debugDir.Path()))
|
||||
}
|
||||
s.printDebugWarnings()
|
||||
return 0
|
||||
}
|
||||
|
||||
func mapWriteFailures(failures []diagnostics.FileFailure) []processreport.RunFailure {
|
||||
out := make([]processreport.RunFailure, len(failures))
|
||||
for i, failure := range failures {
|
||||
phase := "report_write"
|
||||
if failure.Kind == "transcript write" {
|
||||
phase = "transcript_write"
|
||||
}
|
||||
out[i] = processreport.RunFailure{Phase: phase, Reason: outcomes.ReasonInternalFailure, OutputPath: failure.Path}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (s *processState) printFailures(failures []diagnostics.FileFailure) {
|
||||
for _, failure := range failures {
|
||||
fmt.Fprintf(s.stderr, "audita process: %s\n", llm.RedactError(failure, s.cfg))
|
||||
}
|
||||
}
|
||||
|
||||
func (s *processState) printDebugWarnings() {
|
||||
s.debugMu.Lock()
|
||||
defer s.debugMu.Unlock()
|
||||
for _, warning := range s.debugWarnings {
|
||||
fmt.Fprintf(s.stderr, "audita process: %s\n", warning)
|
||||
}
|
||||
}
|
||||
|
||||
// safePath sanitizes presentation only; filesystem destinations remain literal.
|
||||
func (s *processState) safePath(path string) string {
|
||||
return redaction.New(llm.ConfiguredSecrets(&s.cfg)).String(path)
|
||||
}
|
||||
|
||||
func (s *processState) reportDiagnosticError(err error) {
|
||||
if err == nil {
|
||||
return
|
||||
}
|
||||
message := "debug " + llm.RedactError(err, s.cfg)
|
||||
s.debugMu.Lock()
|
||||
defer s.debugMu.Unlock()
|
||||
s.debugWarnings = append(s.debugWarnings, message)
|
||||
return p.Prepared.Debug.Path()
|
||||
}
|
||||
|
||||
@@ -79,7 +79,7 @@ func processCommand() commandkit.Command {
|
||||
in.Overrides = append(in.Overrides, ck.Override{Path: ck.Keys("reporting", "enabled"), Value: v.Value, Flag: "--reporting"})
|
||||
}
|
||||
}))
|
||||
return commandkit.NewCommand(commandkit.CommandSpec[processInput]{Name: "process", Summary: "Polish a transcript.", Flags: flags,
|
||||
return commandkit.NewCommand(commandkit.CommandSpec[processInput]{Name: "process", Summary: "Polish a transcript.", Flags: flags, Run: runProcess,
|
||||
Arguments: []commandkit.Argument[processInput]{commandkit.ValueArgument(commandkit.ArgumentMeta{Name: "transcript", Usage: "Transcript JSON file."}, true, stringValue, func(in *processInput, v commandkit.Supplied[string]) { in.Transcript = v.Value })},
|
||||
})
|
||||
}
|
||||
|
||||
58
internal/cli/process_command_integration_test.go
Normal file
58
internal/cli/process_command_integration_test.go
Normal file
@@ -0,0 +1,58 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
commandkit "go.maximumdirect.net/commandkit"
|
||||
)
|
||||
|
||||
func TestProcessUsesResolvedReportingMode(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
for name, contents := range map[string]string{
|
||||
"config.yml": "version: 3\npipeline:\n stages: []\nreporting:\n enabled: false\n",
|
||||
"transcript.json": `[{"speaker":"Alice","start":0,"end":1,"text":"hello"}]`,
|
||||
"blocked": "file",
|
||||
} {
|
||||
if err := os.WriteFile(filepath.Join(dir, name), []byte(contents), 0644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
app, err := buildApplication()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
flags []string
|
||||
status int
|
||||
stdout bool
|
||||
info bool
|
||||
warning bool
|
||||
}{
|
||||
{"ordinary", nil, 0, true, false, false},
|
||||
{"verbose", []string{"--verbose"}, 0, true, true, false},
|
||||
{"quiet stdout", []string{"--quiet"}, 2, false, false, false},
|
||||
{"quiet file", []string{"--quiet", "--output", filepath.Join(dir, "quiet.json")}, 0, false, false, false},
|
||||
{"verbose after quiet", []string{"--quiet", "--verbose"}, 0, true, true, false},
|
||||
{"quiet after verbose", []string{"--verbose", "--quiet", "--output", filepath.Join(dir, "ordered.json")}, 0, false, false, false},
|
||||
{"debug disabled", []string{"--debug=false", "--work-dir", filepath.Join(dir, "blocked")}, 0, true, false, false},
|
||||
{"debug failure", []string{"--debug", "--work-dir", filepath.Join(dir, "blocked")}, 0, true, true, true},
|
||||
{"debug quiet", []string{"--debug", "--quiet", "--output", filepath.Join(dir, "debug-quiet.json"), "--work-dir", filepath.Join(dir, "blocked")}, 0, false, false, false},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
var out, stderr bytes.Buffer
|
||||
args := []string{"process", "--config", filepath.Join(dir, "config.yml")}
|
||||
args = append(args, tc.flags...)
|
||||
args = append(args, filepath.Join(dir, "transcript.json"))
|
||||
result := app.Execute(context.Background(), args, commandkit.Streams{Out: &out, Err: &stderr})
|
||||
if result.Status != tc.status || (out.Len() > 0) != tc.stdout || strings.Contains(stderr.String(), "Completed:") != tc.info || strings.Contains(stderr.String(), "Debug artifact:") != tc.warning {
|
||||
t.Fatalf("status=%d out=%q err=%q", result.Status, out.String(), stderr.String())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
139
internal/cli/process_inputs.go
Normal file
139
internal/cli/process_inputs.go
Normal file
@@ -0,0 +1,139 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"io"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/core/config"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/core/diagnostics"
|
||||
coreio "gitea.maximumdirect.net/eric/audita/v3/internal/core/io"
|
||||
"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"
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/preflight"
|
||||
"go.maximumdirect.net/ckconfig"
|
||||
commandkit "go.maximumdirect.net/commandkit"
|
||||
ck "go.maximumdirect.net/configkit"
|
||||
)
|
||||
|
||||
type processInputs struct {
|
||||
Configuration *ck.Result[config.Config]
|
||||
Runtime *llm.Runtime
|
||||
Source *schema.SourceTranscript
|
||||
RawTranscript []byte
|
||||
ProtectedPaths []string
|
||||
}
|
||||
|
||||
func loadProcessInputs(ctx context.Context, input processInput, mode commandkit.ReportingMode) (*processInputs, error) {
|
||||
if mode == commandkit.ReportingModeQuiet && input.Output.Value == "" {
|
||||
return nil, fmt.Errorf("quiet processing requires an output file or /dev/null")
|
||||
}
|
||||
selection := ckconfig.Selection{Config: input.Config, Overlay: input.Overlay, Overrides: input.Overrides}
|
||||
loaded, err := loadApplication(ctx, selection, struct{}{}, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
runtime, err := llm.NewRuntime(ctx, loaded, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
raw, err := coreio.ReadRequiredFile(input.Transcript, "transcript")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
source, err := schema.ParseSourceTranscriptJSON(raw)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
protected := append([]string(nil), runtime.SourcePaths...)
|
||||
for _, entry := range loaded.Sources() {
|
||||
if entry.Kind == ck.SourceKindFile && entry.Path != "" {
|
||||
protected = append(protected, entry.Path)
|
||||
}
|
||||
}
|
||||
return &processInputs{Configuration: loaded, Runtime: runtime, Source: source, RawTranscript: raw, ProtectedPaths: protected}, nil
|
||||
}
|
||||
|
||||
// preparedProcess groups the output resources and the facts needed for reports.
|
||||
type preparedProcess struct {
|
||||
Prepared *preflight.Prepared
|
||||
RunID string
|
||||
Started time.Time
|
||||
}
|
||||
|
||||
func prepareProcess(ctx context.Context, input processInput, loaded *processInputs, stdout io.Writer) (*preparedProcess, error) {
|
||||
cfg := &loaded.Configuration.Config
|
||||
reportDir := ""
|
||||
reportDirSet := false
|
||||
for _, override := range input.Overrides {
|
||||
if override.Path.String() == "/reporting/stdout_dir" {
|
||||
reportDirSet = true
|
||||
}
|
||||
}
|
||||
if cfg.Reporting.Enabled {
|
||||
reportDir = cfg.Reporting.StdoutDir
|
||||
if input.Output.Value != "" && !reportDirSet {
|
||||
reportDir = filepath.Dir(input.Output.Value)
|
||||
}
|
||||
}
|
||||
runID, err := generateRunID()
|
||||
if err != nil {
|
||||
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 _, item := range cfg.Pipeline.Stages {
|
||||
snapshot.StageNames = append(snapshot.StageNames, item.Module)
|
||||
}
|
||||
secrets := loaded.Runtime.ConfiguredSecrets(ctx)
|
||||
initial, err := renderReportBundle(snapshot, secrets...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
snapshot.Status = reporting.StatusFailed
|
||||
failed, err := renderReportBundle(snapshot, secrets...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var debug *preflight.DebugInput
|
||||
var warnings []error
|
||||
if input.Debug.Value {
|
||||
var effective bytes.Buffer
|
||||
if err := loaded.Configuration.WriteEffective(&effective); err != nil {
|
||||
warnings = append(warnings, err)
|
||||
}
|
||||
debug = &preflight.DebugInput{WorkDir: cfg.Diagnostics.WorkDir, RunID: runID, Secrets: secrets, EffectiveConfig: effective.Bytes(), Metadata: diagnostics.InvocationMetadata{Operation: "process", TranscriptPath: input.Transcript, OutputPath: input.Output.Value, Overlay: input.Overlay.Value, ConfigVersion: &cfg.Version, RunID: runID, StartedAt: started}}
|
||||
}
|
||||
prepared, err := preflight.Prepare(ctx, preflight.Input{Config: cfg, Transcript: loaded.Source, Output: diagnostics.OutputSelection{InputTranscriptPath: input.Transcript, TranscriptPath: input.Output.Value, ReportDir: reportDir, ReportingEnabled: cfg.Reporting.Enabled, ProtectedPaths: loaded.ProtectedPaths, Stdout: stdout}, InitialReports: initial, FailedReports: failed, Debug: debug})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
prepared.Warnings = append(warnings, prepared.Warnings...)
|
||||
return &preparedProcess{Prepared: prepared, RunID: runID, Started: started}, nil
|
||||
}
|
||||
|
||||
func generateRunID() (string, error) {
|
||||
var random [12]byte
|
||||
if _, err := rand.Read(random[:]); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return "run-" + hex.EncodeToString(random[:]), nil
|
||||
}
|
||||
|
||||
func renderReportBundle(value reporting.Snapshot, secrets ...string) (diagnostics.ReportBundle, error) {
|
||||
errorsYAML, err := reporting.YAML(value.ErrorsDocument(), secrets...)
|
||||
if err != nil {
|
||||
return diagnostics.ReportBundle{}, err
|
||||
}
|
||||
recoveriesYAML, err := reporting.YAML(value.RecoveriesDocument(), secrets...)
|
||||
if err != nil {
|
||||
return diagnostics.ReportBundle{}, err
|
||||
}
|
||||
return diagnostics.ReportBundle{Errors: errorsYAML, Recoveries: recoveriesYAML, Markdown: reporting.Markdown(value, secrets...)}, nil
|
||||
}
|
||||
78
internal/cli/process_inputs_test.go
Normal file
78
internal/cli/process_inputs_test.go
Normal file
@@ -0,0 +1,78 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
commandkit "go.maximumdirect.net/commandkit"
|
||||
ck "go.maximumdirect.net/configkit"
|
||||
)
|
||||
|
||||
func TestQuietDestinationCheckPrecedesInputAcquisition(t *testing.T) {
|
||||
_, err := loadProcessInputs(context.Background(), processInput{Transcript: "missing.json"}, commandkit.ReportingModeQuiet)
|
||||
if err == nil || !strings.Contains(err.Error(), "/dev/null") {
|
||||
t.Fatalf("error=%v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProcessAcquisitionCapturesConsumedFiles(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
for name, contents := range map[string]string{
|
||||
"config.yml": "version: 3\npipeline:\n stages: []\n",
|
||||
"transcript.json": `[{"speaker":"Alice","start":0,"end":1,"text":"hello"}]`,
|
||||
"prompts.yml": "prompts:\n - id: extra\n messages:\n - role: user\n content_file: text.txt\n output:\n format: text\n validation_mode: none\n",
|
||||
"text.txt": "example",
|
||||
} {
|
||||
if err := os.WriteFile(filepath.Join(dir, name), []byte(contents), 0644); 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")}
|
||||
result, err := loadProcessInputs(context.Background(), input, commandkit.ReportingModeOrdinary)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, name := range []string{"config.yml", "prompts.yml", "text.txt"} {
|
||||
if !slices.Contains(result.ProtectedPaths, filepath.Join(dir, name)) {
|
||||
t.Errorf("missing %s in %v", name, result.ProtectedPaths)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestPrepareProcessInitializesReportsAndToleratesDebugLoss(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
for name, contents := range map[string]string{
|
||||
"config.yml": "version: 3\npipeline:\n stages: []\n",
|
||||
"transcript.json": `[{"speaker":"Alice","start":0,"end":1,"text":"hello"}]`,
|
||||
"blocked": "not a directory",
|
||||
} {
|
||||
if err := os.WriteFile(filepath.Join(dir, name), []byte(contents), 0644); 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}, Debug: commandkit.Supplied[bool]{Value: true, Set: true}, Overrides: []ck.Override{{Path: ck.Keys("diagnostics", "work_dir"), Value: filepath.Join(dir, "blocked"), Flag: "--work-dir"}}}
|
||||
loaded, err := loadProcessInputs(context.Background(), input, commandkit.ReportingModeOrdinary)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
prepared, err := prepareProcess(context.Background(), input, loaded, io.Discard)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer prepared.Prepared.Output.Close()
|
||||
if len(prepared.Prepared.Warnings) == 0 {
|
||||
t.Fatal("debug loss was not reported")
|
||||
}
|
||||
report, err := os.ReadFile(filepath.Join(dir, "errors.yml"))
|
||||
if err != nil || !strings.Contains(string(report), "in_progress") {
|
||||
t.Fatalf("report=%q err=%v", report, err)
|
||||
}
|
||||
if _, err := os.Stat(input.Output.Value); !os.IsNotExist(err) {
|
||||
t.Fatalf("output published early: %v", err)
|
||||
}
|
||||
}
|
||||
65
internal/cli/process_output.go
Normal file
65
internal/cli/process_output.go
Normal file
@@ -0,0 +1,65 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"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"
|
||||
commandkit "go.maximumdirect.net/commandkit"
|
||||
)
|
||||
|
||||
// finishProcess delivers the transcript and reports before emitting success information.
|
||||
func finishProcess(inv commandkit.Invocation, input processInput, loaded *processInputs, prepared *preparedProcess, snapshot reporting.Snapshot, transcript []byte, primary error) commandkit.Outcome {
|
||||
reporter := inv.Reporter()
|
||||
secrets := loaded.Runtime.ConfiguredSecrets(inv.Context)
|
||||
redact := func(err error) string { return loaded.Runtime.RedactError(inv.Context, err) }
|
||||
for _, warning := range prepared.Prepared.Warnings {
|
||||
_ = reporter.Warn("Debug artifact: " + redact(warning))
|
||||
}
|
||||
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 {
|
||||
snapshot.Status = reporting.StatusFailed
|
||||
if len(snapshot.Errors) == 0 {
|
||||
snapshot.Errors = []reporting.ErrorRecord{{Scope: "run", Reason: "processing_failure", Message: "Processing failed"}}
|
||||
}
|
||||
} else {
|
||||
snapshot.Status = reporting.StatusCompleted
|
||||
}
|
||||
reports, err := renderReportBundle(snapshot, secrets...)
|
||||
if err != nil {
|
||||
if primary == nil {
|
||||
primary = err
|
||||
}
|
||||
snapshot.Status = reporting.StatusFailed
|
||||
}
|
||||
failed := func(failures []diagnostics.FileFailure) diagnostics.ReportBundle {
|
||||
copySnapshot := snapshot
|
||||
copySnapshot.Status = reporting.StatusFailed
|
||||
copySnapshot.Errors = append([]reporting.ErrorRecord(nil), snapshot.Errors...)
|
||||
for _, failure := range failures {
|
||||
copySnapshot.Errors = append(copySnapshot.Errors, reporting.ErrorRecord{Scope: "run", Reason: "output_failure", Message: "Required output delivery failed", OutputPath: failure.Path})
|
||||
}
|
||||
result, bundleErr := renderReportBundle(copySnapshot, secrets...)
|
||||
if bundleErr != nil {
|
||||
return reports
|
||||
}
|
||||
return result
|
||||
}
|
||||
result := prepared.Prepared.Output.Finish(primary, transcript, reports, failed)
|
||||
if result != nil {
|
||||
return commandkit.Failure(redact(result)).WithCause(result)
|
||||
}
|
||||
safe := redaction.New(secrets)
|
||||
_ = reporter.AtEnd().Info(reporting.CompletionLine(snapshot))
|
||||
for _, path := range prepared.Prepared.Output.ReportPaths() {
|
||||
_ = reporter.AtEnd().Info(fmt.Sprintf("Report: %s", safe.String(path)))
|
||||
}
|
||||
if prepared.Prepared.Debug != nil {
|
||||
_ = reporter.AtEnd().Info(fmt.Sprintf("Diagnostics: %s", safe.String(prepared.Prepared.Debug.Path())))
|
||||
}
|
||||
return commandkit.Success()
|
||||
}
|
||||
112
internal/cli/process_output_test.go
Normal file
112
internal/cli/process_output_test.go
Normal file
@@ -0,0 +1,112 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"gitea.maximumdirect.net/eric/audita/v3/internal/core/reporting"
|
||||
commandkit "go.maximumdirect.net/commandkit"
|
||||
)
|
||||
|
||||
func TestPreparedDeliveryFollowsReportingMode(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
for name, contents := range map[string]string{
|
||||
"config.yml": "version: 3\npipeline:\n stages: []\nreporting:\n enabled: false\n",
|
||||
"transcript.json": `[{"speaker":"Alice","start":0,"end":1,"text":"hello"}]`,
|
||||
} {
|
||||
if err := os.WriteFile(filepath.Join(dir, name), []byte(contents), 0644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
flags []string
|
||||
file bool
|
||||
wantInfo bool
|
||||
}{
|
||||
{"ordinary", nil, false, false}, {"verbose", []string{"--verbose"}, false, true}, {"quiet file", []string{"--quiet"}, true, false},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
input := processInput{Config: commandkit.Supplied[string]{Value: filepath.Join(dir, "config.yml"), Set: true}, Transcript: filepath.Join(dir, "transcript.json")}
|
||||
if tc.file {
|
||||
input.Output = commandkit.Supplied[string]{Value: filepath.Join(dir, "result.json"), Set: true}
|
||||
}
|
||||
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 {
|
||||
return commandkit.Usage(err.Error())
|
||||
}
|
||||
prepared, err := prepareProcess(inv.Context, input, loaded, inv.Reporter().Output())
|
||||
if err != nil {
|
||||
return commandkit.Failure(err.Error())
|
||||
}
|
||||
return finishProcess(inv, input, loaded, prepared, reporting.Snapshot{}, []byte("transcript bytes"), nil)
|
||||
}})
|
||||
app, err := commandkit.New(commandkit.Identity{Name: "test"}, cmd)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var out, stderr bytes.Buffer
|
||||
args := append([]string{"deliver"}, tc.flags...)
|
||||
outcome := app.Execute(context.Background(), args, commandkit.Streams{Out: &out, Err: &stderr})
|
||||
if outcome.Status != 0 {
|
||||
t.Fatalf("outcome=%+v err=%s", outcome, stderr.String())
|
||||
}
|
||||
if tc.file {
|
||||
data, err := os.ReadFile(input.Output.Value)
|
||||
if err != nil || string(data) != "transcript bytes" {
|
||||
t.Fatalf("file=%q err=%v", data, err)
|
||||
}
|
||||
if out.Len() != 0 {
|
||||
t.Fatalf("stdout=%q", out.String())
|
||||
}
|
||||
} else if out.String() != "transcript bytes" {
|
||||
t.Fatalf("stdout=%q", out.String())
|
||||
}
|
||||
if strings.Contains(stderr.String(), "Completed:") != tc.wantInfo {
|
||||
t.Fatalf("stderr=%q", stderr.String())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeliveryFailureSuppressesCompletionSummary(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
for name, contents := range map[string]string{
|
||||
"config.yml": "version: 3\npipeline:\n stages: []\nreporting:\n enabled: false\n",
|
||||
"transcript.json": `[{"speaker":"Alice","start":0,"end":1,"text":"hello"}]`,
|
||||
} {
|
||||
if err := os.WriteFile(filepath.Join(dir, name), []byte(contents), 0644); 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")}
|
||||
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 {
|
||||
return commandkit.Failure(err.Error())
|
||||
}
|
||||
prepared, err := prepareProcess(inv.Context, input, loaded, inv.Reporter().Output())
|
||||
if err != nil {
|
||||
return commandkit.Failure(err.Error())
|
||||
}
|
||||
return finishProcess(inv, input, loaded, prepared, reporting.Snapshot{}, []byte("transcript bytes"), nil)
|
||||
}})
|
||||
app, err := commandkit.New(commandkit.Identity{Name: "test"}, cmd)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var stderr bytes.Buffer
|
||||
outcome := app.Execute(context.Background(), []string{"deliver", "--verbose"}, commandkit.Streams{Out: shortProcessWriter{}, Err: &stderr})
|
||||
if outcome.Status == 0 || strings.Contains(stderr.String(), "Completed:") {
|
||||
t.Fatalf("status=%d stderr=%q", outcome.Status, stderr.String())
|
||||
}
|
||||
}
|
||||
|
||||
type shortProcessWriter struct{}
|
||||
|
||||
func (shortProcessWriter) Write(data []byte) (int, error) { return len(data) / 2, nil }
|
||||
76
internal/framework/preflight/preflight.go
Normal file
76
internal/framework/preflight/preflight.go
Normal file
@@ -0,0 +1,76 @@
|
||||
// Package preflight checks resolved processing inputs and prepares output resources.
|
||||
package preflight
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"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/schema"
|
||||
)
|
||||
|
||||
// Input contains already acquired application and transcript inputs.
|
||||
type Input struct {
|
||||
Config *config.Config
|
||||
Transcript *schema.SourceTranscript
|
||||
Output diagnostics.OutputSelection
|
||||
InitialReports diagnostics.ReportBundle
|
||||
FailedReports diagnostics.ReportBundle
|
||||
Debug *DebugInput
|
||||
}
|
||||
|
||||
type DebugInput struct {
|
||||
WorkDir string
|
||||
RunID string
|
||||
Metadata diagnostics.InvocationMetadata
|
||||
EffectiveConfig []byte
|
||||
Secrets []string
|
||||
}
|
||||
|
||||
type Prepared struct {
|
||||
Output *diagnostics.OutputSession
|
||||
Debug *diagnostics.RunDirectory
|
||||
Warnings []error
|
||||
}
|
||||
|
||||
// Prepare validates invocation inputs and collisions before acquiring outputs.
|
||||
func Prepare(ctx context.Context, input Input) (*Prepared, error) {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if input.Config == nil {
|
||||
return nil, fmt.Errorf("application configuration is required")
|
||||
}
|
||||
if input.Transcript == nil {
|
||||
return nil, fmt.Errorf("transcript is required")
|
||||
}
|
||||
session, err := diagnostics.SelectOutput(input.Output)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := session.Prepare(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
_ = session.Close()
|
||||
return nil, err
|
||||
}
|
||||
if err := session.Initialize(input.InitialReports, input.FailedReports); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, session.Finish(err, nil, input.FailedReports, func([]diagnostics.FileFailure) diagnostics.ReportBundle { return input.FailedReports })
|
||||
}
|
||||
result := &Prepared{Output: session}
|
||||
if input.Debug != nil {
|
||||
result.Debug, result.Warnings = session.PrepareDebug(input.Debug.WorkDir, input.Debug.RunID, input.Debug.Metadata, input.Debug.EffectiveConfig, input.Debug.Secrets...)
|
||||
}
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, session.Finish(err, nil, input.FailedReports, func([]diagnostics.FileFailure) diagnostics.ReportBundle { return input.FailedReports })
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
48
internal/framework/preflight/preflight_test.go
Normal file
48
internal/framework/preflight/preflight_test.go
Normal file
@@ -0,0 +1,48 @@
|
||||
package preflight
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"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/schema"
|
||||
)
|
||||
|
||||
func TestReadinessAndCollisionPrecedeOutputPreparation(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
output := filepath.Join(dir, "output.json")
|
||||
if _, err := Prepare(context.Background(), Input{Config: &config.Config{}, Output: diagnostics.OutputSelection{TranscriptPath: output}}); err == nil {
|
||||
t.Fatal("missing transcript accepted")
|
||||
}
|
||||
if _, err := os.Stat(output); !os.IsNotExist(err) {
|
||||
t.Fatalf("output created: %v", err)
|
||||
}
|
||||
if _, err := Prepare(context.Background(), Input{Config: &config.Config{}, Transcript: &schema.SourceTranscript{}, Output: diagnostics.OutputSelection{TranscriptPath: output, ProtectedPaths: []string{output}}}); err == nil {
|
||||
t.Fatal("collision accepted")
|
||||
}
|
||||
entries, err := os.ReadDir(dir)
|
||||
if err != nil || len(entries) != 0 {
|
||||
t.Fatalf("prepared before check: %v %v", entries, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOptionalDebugFailureDoesNotPreventPreparation(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
blocked := filepath.Join(dir, "file")
|
||||
if err := os.WriteFile(blocked, []byte("x"), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
result, err := Prepare(context.Background(), Input{Config: &config.Config{}, Transcript: &schema.SourceTranscript{}, Output: diagnostics.OutputSelection{TranscriptPath: filepath.Join(dir, "output")}, Debug: &DebugInput{WorkDir: blocked, RunID: "run"}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(result.Warnings) == 0 || result.Debug != nil {
|
||||
t.Fatalf("debug=%v warnings=%v", result.Debug, result.Warnings)
|
||||
}
|
||||
if err := result.Output.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user