eric/audita
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity

Compose resilient CLI processing and report lifecycle

Browse Source
This commit is contained in:
Eric Rakestraw
2026-09-24 17:48:52 +00:00
parent 87b8d9ad5c
commit 2f036b0356
11 changed files with 676 additions and 777 deletions
Download Patch File Download Diff File Expand all files Collapse all files

431
internal/cli/process.go Normal file
View File

@@ -0,0 +1,431 @@
package cli
import (
"bytes"
"context"
"crypto/rand"
"encoding/hex"
"errors"
"fmt"
"io"
"path/filepath"
"strings"
"time"
"gitea.maximumdirect.net/eric/audita/internal/core/chunking"
"gitea.maximumdirect.net/eric/audita/internal/core/config"
"gitea.maximumdirect.net/eric/audita/internal/core/diagnostics"
coreio "gitea.maximumdirect.net/eric/audita/internal/core/io"
"gitea.maximumdirect.net/eric/audita/internal/core/normalization"
"gitea.maximumdirect.net/eric/audita/internal/core/outputschema"
"gitea.maximumdirect.net/eric/audita/internal/core/reporting"
"gitea.maximumdirect.net/eric/audita/internal/core/schema"
"gitea.maximumdirect.net/eric/audita/internal/framework/contracts"
"gitea.maximumdirect.net/eric/audita/internal/framework/llm"
"gitea.maximumdirect.net/eric/audita/internal/framework/modules"
"gitea.maximumdirect.net/eric/audita/internal/framework/outcomes"
"gitea.maximumdirect.net/eric/audita/internal/framework/processreport"
"gitea.maximumdirect.net/eric/audita/internal/framework/runner"
"gitea.maximumdirect.net/eric/audita/internal/framework/structuredoutput"
)
func runProcess(ctx context.Context, args []string, stdout, stderr io.Writer) int {
input, help, err := parseConfigInput("process", args, stderr)
if err != nil {
fmt.Fprintf(stderr, "audita process: %s\n", configInputError(err))
return 2
}
if help {
writeProcessUsage(stdout)
return 0
}
selected, err := loadConfigInput(input, "process")
if err != nil {
fmt.Fprintf(stderr, "audita process: %s\n", configInputError(err))
return 2
}
cfg := selected.Result.Config
reportDir, err := selectedReportDir(cfg, input.OutputPath)
if err != nil {
fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg))
return 2
}
if cfg.Diagnostics.Debug && strings.TrimSpace(cfg.Diagnostics.WorkDir) == "" {
fmt.Fprintln(stderr, "audita process: debug work directory is required")
return 2
}
if err := ctx.Err(); err != nil {
fmt.Fprintf(stderr, "audita process: %s\n", err)
return 1
}
runtime, err := llm.NewRuntime(ctx, cfg, selected.Path, nil)
if err != nil {
fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg))
if ctx.Err() != nil || llm.IsAssetError(err) {
return 1
}
return 2
}
specs, err := contracts.ResolveModuleRunSpecs(cfg.Pipeline.Modules)
if err != nil {
fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg))
return 2
}
runID, err := generateRunID()
if err != nil {
fmt.Fprintf(stderr, "audita process: run ID: %s\n", llm.RedactError(err, cfg))
return 1
}
started := time.Now().UTC()
state := &processState{cfg: cfg, runtime: runtime, specs: specs, runID: runID, started: started,
transcriptPath: input.TranscriptPath, glossaryPath: input.GlossaryPath, outputPath: input.OutputPath,
selected: selected, overlay: input.Overlay, stderr: stderr, stdout: stdout, ctx: ctx}
protected := []string{input.TranscriptPath, input.GlossaryPath}
for _, source := range selected.Result.Sources() {
if source.Path != "" {
protected = append(protected, source.Path)
}
}
protected = append(protected, selectedProfilePaths(cfg, selected.Path)...)
session, err := diagnostics.PrepareOutput(diagnostics.OutputSelection{TranscriptPath: input.OutputPath,
ReportDir: reportDir, ReportingEnabled: cfg.Reporting.Enabled, ProtectedPaths: protected})
if err != nil {
fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg))
return 2
}
state.session = session
initial := state.snapshot(reporting.StatusInProgress, nil, nil)
initialBundle, err := state.bundle(initial)
if err != nil {
fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg))
return 1
}
initFailure := []processreport.RunFailure{{Phase: "initialization", Reason: outcomes.ReasonInternalFailure}}
failedInit, err := state.bundle(state.snapshot(reporting.StatusFailed, nil, initFailure))
if err != nil {
fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg))
return 1
}
if err := session.Initialize(initialBundle, failedInit); err != nil {
fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, cfg))
return 1
}
state.startDebug()
return state.execute()
}
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
runOutput *runner.RunOutput
}
func generateRunID() (string, error) {
var random [12]byte
if _, err := rand.Read(random[:]); err != nil {
return "", err
}
return "run-" + hex.EncodeToString(random[:]), nil
}
func selectedProfilePaths(cfg config.Config, configPath string) []string {
dir := cfg.ProfilesDir
if !filepath.IsAbs(dir) {
dir = filepath.Join(filepath.Dir(configPath), dir)
}
ids := []string{cfg.LLM.ProposalProfile, cfg.LLM.ValidationProfile}
paths := make([]string, 0, len(ids)*2)
for _, id := range ids {
paths = append(paths, filepath.Join(dir, id+".yaml"), filepath.Join(dir, id+".yml"))
}
return paths
}
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...)
if err != nil {
return diagnostics.ReportBundle{}, err
}
recoveriesYAML, err := reporting.YAML(snapshot.RecoveriesDocument(), secrets...)
if err != nil {
return diagnostics.ReportBundle{}, 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
}
if err := write(s.debugDir); err != nil {
s.debugWarnings = append(s.debugWarnings, "debug "+label+": "+llm.RedactError(err, s.cfg))
}
}
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.debugWarnings = append(s.debugWarnings, "debug directory: "+llm.RedactError(err, s.cfg))
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,
})
})
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.debugWarnings = append(s.debugWarnings, "debug effective settings: "+llm.RedactError(err, s.cfg))
}
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)
})
}
if profiles, err := s.runtime.SelectedProfilesYAML(); err == nil {
s.debugWrite("selected profiles", func(d *diagnostics.RunDirectory) error {
return d.WriteYAMLArtifact(diagnostics.ArtifactSelectedProfiles, profiles)
})
}
}
func (s *processState) debugPath() string {
if s.debugDir == 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,
})
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.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.outputPath == "" {
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.debugDir.Path())
}
return 1
}
fmt.Fprintln(s.stderr, reporting.CompletionLine(snapshot))
for _, path := range s.session.ReportPaths() {
fmt.Fprintf(s.stderr, "Report: %s\n", path)
}
if s.debugDir != nil {
fmt.Fprintf(s.stderr, "Debug: %s\n", 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() {
for _, warning := range s.debugWarnings {
fmt.Fprintf(s.stderr, "audita process: %s\n", warning)
}
}

353
internal/cli/run.go
View File

@@ -1,229 +1,14 @@
package cli
import (
"bytes"
"context"
"fmt"
"io"
"strings"
"time"
ck "gitea.maximumdirect.net/eric/ConfigKit"
"gitea.maximumdirect.net/eric/audita/internal/core/chunking"
"gitea.maximumdirect.net/eric/audita/internal/core/config"
"gitea.maximumdirect.net/eric/audita/internal/core/diagnostics"
coreio "gitea.maximumdirect.net/eric/audita/internal/core/io"
"gitea.maximumdirect.net/eric/audita/internal/core/normalization"
"gitea.maximumdirect.net/eric/audita/internal/core/outputschema"
"gitea.maximumdirect.net/eric/audita/internal/core/schema"
"gitea.maximumdirect.net/eric/audita/internal/framework/contracts"
"gitea.maximumdirect.net/eric/audita/internal/framework/llm"
"gitea.maximumdirect.net/eric/audita/internal/framework/modules"
"gitea.maximumdirect.net/eric/audita/internal/framework/processreport"
"gitea.maximumdirect.net/eric/audita/internal/framework/runner"
)
type processInvocation struct {
TranscriptPath string
GlossaryPath string
OutputPath string
ReportDir string
RunID string
Config config.Config
ConfigResult *ck.Result[config.Config]
LLMRuntime *llm.Runtime
ConfigPath string
ConfigSource string
Overlay string
ConfigVersion *int
}
var processRunner = func(ctx context.Context, inv processInvocation, stdout, stderr io.Writer) (*normalization.NormalizationSummary, *chunking.Summary, *runner.RunOutput, *diagnostics.RunDirectory, error) {
var runDir *diagnostics.RunDirectory
if inv.Config.Diagnostics.Debug {
var err error
runDir, err = diagnostics.NewRunDirectory(inv.Config.Diagnostics.WorkDir, inv.RunID, llm.ConfiguredSecrets(&inv.Config)...)
if err != nil {
fmt.Fprintf(stderr, "audita process: debug directory: %s\n", llm.RedactError(err, inv.Config))
}
}
debugPath := ""
if runDir != nil {
debugPath = runDir.Path()
}
debugWrite := func(label string, write func(*diagnostics.RunDirectory) error) {
if runDir == nil {
return
}
if err := write(runDir); err != nil {
fmt.Fprintf(stderr, "audita process: debug %s: %s\n", label, llm.RedactError(err, inv.Config))
}
}
fail := func(phase string, err error, runOutput *runner.RunOutput) (*normalization.NormalizationSummary, *chunking.Summary, *runner.RunOutput, *diagnostics.RunDirectory, error) {
debugWrite("error log", func(d *diagnostics.RunDirectory) error {
return d.WriteErrorLog(fmt.Sprintf("%s: %s", phase, llm.RedactError(err, inv.Config)))
})
return nil, nil, runOutput, runDir, fmt.Errorf("%s: %w", phase, err)
}
debugWrite("invocation", func(d *diagnostics.RunDirectory) error {
return d.WriteInvocationMetadata(diagnostics.InvocationMetadata{
Operation: "process",
TranscriptPath: inv.TranscriptPath,
GlossaryPath: inv.GlossaryPath,
OutputPath: inv.OutputPath,
ConfigPath: inv.ConfigPath,
ConfigSource: inv.ConfigSource,
Overlay: inv.Overlay,
ConfigVersion: inv.ConfigVersion,
TranscriptDescription: inv.Config.Context.Description,
Modules: append([]string(nil), inv.Config.Pipeline.Modules...),
})
})
if runDir != nil && inv.ConfigResult != nil {
var effective bytes.Buffer
if err := inv.ConfigResult.WriteEffective(&effective); err == nil {
debugWrite("effective settings", func(d *diagnostics.RunDirectory) error { return d.WriteEffectiveConfig(effective.Bytes()) })
}
if origins, err := llm.ExplanationYAML(inv.ConfigResult); err == nil {
debugWrite("configuration origins", func(d *diagnostics.RunDirectory) error {
return d.WriteYAMLArtifact(diagnostics.ArtifactConfigOrigins, origins)
})
}
}
if runDir != nil && inv.LLMRuntime != nil {
if profiles, err := inv.LLMRuntime.SelectedProfilesYAML(); err == nil {
debugWrite("selected profiles", func(d *diagnostics.RunDirectory) error {
return d.WriteYAMLArtifact(diagnostics.ArtifactSelectedProfiles, profiles)
})
}
}
transcriptBytes, err := coreio.ReadRequiredFile(inv.TranscriptPath, "transcript")
if err != nil {
return fail("transcript_read", err, nil)
}
glossaryBytes, err := coreio.ReadRequiredFile(inv.GlossaryPath, "glossary")
if err != nil {
return fail("glossary_read", err, nil)
}
sourceTranscript, err := schema.ParseSourceTranscriptJSON(transcriptBytes)
if err != nil {
return fail("transcript_schema", err, nil)
}
glossary, err := schema.ParseGlossaryYAML(glossaryBytes)
if err != nil {
return fail("glossary_schema", err, nil)
}
debugWrite("source transcript", func(d *diagnostics.RunDirectory) error {
return d.WriteSourceTranscript(sourceTranscript, transcriptBytes)
})
canonical := sourceToCanonicalTranscript(sourceTranscript)
normalizer := normalization.NewNormalizer(normalization.NormalizationConfig{
MaxSegmentGap: inv.Config.Normalization.MaxSegmentGap,
EllipsisGap: inv.Config.Normalization.EllipsisGap,
MaxSegmentDuration: inv.Config.Normalization.MaxSegmentDuration,
MaxSegmentTokens: inv.Config.Normalization.MaxSegmentTokens,
})
normalizedTranscript, normSummary := normalizer.Normalize(canonical)
debugWrite("normalized transcript", func(d *diagnostics.RunDirectory) error { return d.WriteNormalizedTranscript(normalizedTranscript) })
debugWrite("normalization summary", func(d *diagnostics.RunDirectory) error { return d.WriteNormalizationSummary(normSummary) })
// Compute chunks after normalization
chunker := chunking.NewChunker(chunking.ChunkingConfig{
MaxSectionTokens: inv.Config.Chunking.MaxSectionTokens,
MinSectionTokens: inv.Config.Chunking.MinSectionTokens,
TargetSections: inv.Config.Chunking.TargetSections,
})
sections, chunkErr := chunker.ChunkTranscript(normalizedTranscript)
if chunkErr != nil {
return fail("chunking", chunkErr, nil)
}
chunkConfig := chunking.ChunkingConfig{
MaxSectionTokens: inv.Config.Chunking.MaxSectionTokens,
MinSectionTokens: inv.Config.Chunking.MinSectionTokens,
TargetSections: inv.Config.Chunking.TargetSections,
}
chunkSummary := chunking.ComputeSummary(sections, chunkConfig)
chunkDetailedSummary := chunking.ComputeDetailedSummary(sections, chunkConfig)
debugWrite("chunking summary", func(d *diagnostics.RunDirectory) error { return d.WriteChunkingSummary(&chunkDetailedSummary) })
workingTranscript := normalizedTranscript
var runOutput *runner.RunOutput
observations := llm.NewObservationCollector()
moduleFactory := modules.NewFactory(modules.Dependencies{Config: &inv.Config, Glossary: glossary, DiagnosticsDir: debugPath})
if inv.LLMRuntime == nil {
return fail("runner_setup", fmt.Errorf("LLM runtime is required"), nil)
}
proposalLLMClient, err := inv.LLMRuntime.NewRoleClient("proposal", observations.Hooks())
if err != nil {
return fail("runner_setup", err, nil)
}
validationLLMClient, err := inv.LLMRuntime.NewRoleClient("validation", observations.Hooks())
if err != nil {
return fail("runner_setup", err, nil)
}
moduleSpecs, err := contracts.ResolveModuleRunSpecs(inv.Config.Pipeline.Modules)
if err != nil {
return fail("runner_setup", err, nil)
}
runnerResult, runErr := runner.New(moduleFactory).Run(ctx, runner.RunInput{
Config: &inv.Config,
Transcript: normalizedTranscript,
Glossary: glossary,
ModuleSpecs: moduleSpecs,
ProposalLLMClient: proposalLLMClient,
ProposalDiagnosticsDir: debugPath,
ValidationLLMClient: validationLLMClient,
ValidationDiagnosticsDir: debugPath,
CaptureCandidates: runDir != nil,
})
runnerResult.Observations = observations.Snapshot()
runnerResult.ObservationsAvailable = true
runnerResult.LLMSummary = llm.Summarize(runnerResult.Observations)
runOutput = &runnerResult
if runErr != nil {
return fail("runner_execution", runErr, runOutput)
}
workingTranscript = runnerResult.FinalTranscript
encoderDef, err := outputschema.Resolve(inv.Config.Output.Schema)
if err != nil {
return fail("output_schema", err, runOutput)
}
outputBytes, err := encoderDef.Encoder(workingTranscript)
if err != nil {
return fail("serialization", err, runOutput)
}
if strings.TrimSpace(inv.OutputPath) != "" {
if err := coreio.WriteFile(inv.OutputPath, outputBytes); err != nil {
return fail("output_write", err, runOutput)
}
return normSummary, &chunkSummary, runOutput, runDir, nil
}
if _, err := stdout.Write(outputBytes); err != nil {
return fail("stdout_write", err, runOutput)
}
return normSummary, &chunkSummary, runOutput, runDir, nil
}
func sourceToCanonicalTranscript(source *schema.SourceTranscript) *schema.Transcript {
segments := make([]schema.Segment, len(source.Segments))
for i, s := range source.Segments {
@@ -277,108 +62,6 @@ func Run(ctx context.Context, args []string, stdout, stderr io.Writer) int {
return 2
}
func runProcess(ctx context.Context, args []string, stdout, stderr io.Writer) int {
startedAt := time.Now().UTC()
input, help, err := parseConfigInput("process", args, stderr)
if err != nil {
fmt.Fprintf(stderr, "audita process: %s\n", configInputError(err))
return 2
}
if help {
writeProcessUsage(stdout)
return 0
}
selected, err := loadConfigInput(input, "process")
if err != nil {
fmt.Fprintf(stderr, "audita process: %s\n", configInputError(err))
return 2
}
reportDir, err := selectedReportDir(selected.Result.Config, input.OutputPath)
if err != nil {
fmt.Fprintf(stderr, "audita process: %s\n", configInputError(err))
return 2
}
if selected.Result.Config.Diagnostics.Debug && strings.TrimSpace(selected.Result.Config.Diagnostics.WorkDir) == "" {
fmt.Fprintln(stderr, "audita process: debug work directory is required")
return 2
}
llmRuntime, err := llm.NewRuntime(ctx, selected.Result.Config, selected.Path, nil)
if err != nil {
fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(err, selected.Result.Config))
if llm.IsAssetError(err) {
return 1
}
return 2
}
version := selected.Result.Config.Version
inv := processInvocation{
TranscriptPath: input.TranscriptPath,
GlossaryPath: input.GlossaryPath,
OutputPath: input.OutputPath,
ReportDir: reportDir,
Config: selected.Result.Config,
ConfigResult: selected.Result,
LLMRuntime: llmRuntime,
ConfigPath: selected.Path,
ConfigSource: selected.Source,
Overlay: input.Overlay,
ConfigVersion: &version,
}
normSummary, chunkSummary, runOutput, runDir, runErr := processRunner(ctx, inv, stdout, stderr)
completedAt := time.Now().UTC()
if runErr != nil {
if runDir != nil && runOutput != nil {
writeExecutionArtifacts(runDir, runOutput, inv.Config, stderr)
if debugErr := runDir.WriteJSONArtifact(diagnostics.ArtifactCorrectionLedger, processreport.BuildCorrectionLedger(processreport.CorrectionLedgerInput{
RunDirectoryPath: runDir.Path(),
RunOutput: runOutput,
})); debugErr != nil {
fmt.Fprintf(stderr, "audita process: debug correction ledger: %s\n", llm.RedactError(debugErr, inv.Config))
}
}
errorPhase, errorMessage := extractErrorPhase(runErr)
errorMessage = llm.RedactError(fmt.Errorf("%s", errorMessage), inv.Config)
report := processreport.Build(processReportInput("failed", inv, runDir, startedAt, completedAt, errorMessage, errorPhase, nil, nil, runOutput))
fmt.Fprintf(stderr, "audita process: %s\n", llm.RedactError(runErr, inv.Config))
if runDir != nil {
fmt.Fprintf(stderr, "audita process: diagnostics: %s\n", runDir.Path())
}
return 1
}
if runDir != nil && runOutput != nil {
writeExecutionArtifacts(runDir, runOutput, inv.Config, stderr)
if debugErr := runDir.WriteJSONArtifact(diagnostics.ArtifactCorrectionLedger, processreport.BuildCorrectionLedger(processreport.CorrectionLedgerInput{
RunDirectoryPath: runDir.Path(),
RunOutput: runOutput,
})); debugErr != nil {
fmt.Fprintf(stderr, "audita process: debug correction ledger: %s\n", llm.RedactError(debugErr, inv.Config))
}
}
report := processreport.Build(processReportInput("success", inv, runDir, startedAt, completedAt, "", "", normSummary, chunkSummary, runOutput))
return 0
}
func writeExecutionArtifacts(runDir *diagnostics.RunDirectory, output *runner.RunOutput, cfg config.Config, stderr io.Writer) {
if !output.ObservationsAvailable {
return
}
if err := runDir.WriteJSONArtifact(diagnostics.ArtifactLLMExecution, processreport.BuildLLMExecution(output.Observations)); err != nil {
fmt.Fprintf(stderr, "audita process: debug model execution: %s\n", llm.RedactError(err, cfg))
}
if err := runDir.WriteJSONArtifact(diagnostics.ArtifactUtilizationSummary, processreport.BuildUtilization(output, cfg.Backends)); err != nil {
fmt.Fprintf(stderr, "audita process: debug utilization: %s\n", llm.RedactError(err, cfg))
}
}
func runConfig(ctx context.Context, args []string, stdout, stderr io.Writer) int {
if len(args) == 0 || isHelpCommand(args) || args[0] == "--help" || args[0] == "-h" {
writeConfigUsage(stdout)
@@ -466,42 +149,6 @@ func runConfigPrintEffective(ctx context.Context, args []string, stdout, stderr
return 0
}
func extractErrorPhase(err error) (phase string, message string) {
msg := err.Error()
if strings.Contains(msg, ": ") {
parts := strings.SplitN(msg, ": ", 2)
if len(parts) == 2 {
return parts[0], parts[1]
}
}
return "", msg
}
func processReportInput(status string, inv processInvocation, runDir *diagnostics.RunDirectory, startedAt, completedAt time.Time, errorMessage string, errorPhase string, normalizationSummary *normalization.NormalizationSummary, chunkingSummary *chunking.Summary, runOutput *runner.RunOutput) processreport.BuildInput {
runDirectoryPath := ""
if runDir != nil {
runDirectoryPath = runDir.Path()
}
return processreport.BuildInput{
Status: status,
TranscriptPath: inv.TranscriptPath,
GlossaryPath: inv.GlossaryPath,
OutputPath: inv.OutputPath,
Modules: inv.Config.Pipeline.Modules,
OutputSchema: inv.Config.Output.Schema,
ConfigVersion: inv.ConfigVersion,
StartedAt: startedAt,
CompletedAt: completedAt,
ErrorMessage: errorMessage,
ErrorPhase: errorPhase,
RunDirectoryPath: runDirectoryPath,
NormalizationSummary: normalizationSummary,
ChunkingSummary: chunkingSummary,
RunOutput: runOutput,
Backends: inv.Config.Backends,
}
}
func isHelpCommand(args []string) bool {
if len(args) == 0 {
return false

4
internal/core/diagnostics/output_session.go
View File

@@ -151,7 +151,7 @@ func (s *OutputSession) Initialize(initial, failed ReportBundle) error {
// Finalize publishes the transcript first, then errors, recoveries, and Markdown.
// An empty transcript payload means processing failed and no transcript is written.
// Every enabled report is attempted even after an earlier write failure.
func (s *OutputSession) Finalize(transcript []byte, reports, failed ReportBundle) []FileFailure {
func (s *OutputSession) Finalize(transcript []byte, reports ReportBundle, failed func([]FileFailure) ReportBundle) []FileFailure {
if s == nil {
return []FileFailure{{Kind: "output finalization", Err: fmt.Errorf("output session is required")}}
}
@@ -168,7 +168,7 @@ func (s *OutputSession) Finalize(transcript []byte, reports, failed ReportBundle
}
}
if len(failures) > 0 {
for i, payload := range bundleParts(failed) {
for i, payload := range bundleParts(failed(failures)) {
_ = s.replace(s.reports[i], payload)
}
}

6
internal/core/diagnostics/output_session_test.go
View File

@@ -57,7 +57,7 @@ func TestPrepareOutputPreservesExistingTranscriptAndChecksParent(t *testing.T) {
if got, err := os.ReadFile(output); err != nil || string(got) != "old content" {
t.Fatalf("preflight truncated output: %s %v", got, err)
}
if failures := session.Finalize([]byte("new content"), ReportBundle{}, ReportBundle{}); len(failures) != 0 {
if failures := session.Finalize([]byte("new content"), ReportBundle{}, func([]FileFailure) ReportBundle { return ReportBundle{} }); len(failures) != 0 {
t.Fatal(failures)
}
if got, err := os.ReadFile(output); err != nil || string(got) != "new content" {
@@ -127,7 +127,7 @@ func TestStdoutModeInitializesAndFinalizesReports(t *testing.T) {
t.Fatalf("initial %s: %s %v", path, got, err)
}
}
if failures := session.Finalize(nil, finished, ReportBundle{}); len(failures) != 0 {
if failures := session.Finalize(nil, finished, func([]FileFailure) ReportBundle { return ReportBundle{} }); len(failures) != 0 {
t.Fatal(failures)
}
for i, path := range session.ReportPaths() {
@@ -186,7 +186,7 @@ func TestFinalizationCollectsFailuresAndRetainsPublishedTranscript(t *testing.T)
}
complete := ReportBundle{Errors: []byte("complete"), Recoveries: []byte("complete"), Markdown: []byte("complete")}
failed := ReportBundle{Errors: []byte("failed"), Recoveries: []byte("failed"), Markdown: []byte("failed")}
failures := session.Finalize([]byte("transcript"), complete, failed)
failures := session.Finalize([]byte("transcript"), complete, func([]FileFailure) ReportBundle { return failed })
if len(failures) != 1 || failures[0].Path != filepath.Join(dir, ErrorsFile) {
t.Fatalf("failures: %+v", failures)
}

2
internal/framework/contracts/contracts.go
View File

@@ -45,6 +45,7 @@ type SectionMetadata struct {
StartSegmentID int `json:"start_segment_id"`
EndSegmentID int `json:"end_segment_id"`
EstimatedTokens int `json:"estimated_tokens"`
SegmentCount int `json:"segment_count"`
}
// SectionMetadataFromSection converts a chunking section into framework metadata.
@@ -54,6 +55,7 @@ func SectionMetadataFromSection(section chunking.Section) SectionMetadata {
StartSegmentID: section.StartSegmentID,
EndSegmentID: section.EndSegmentID,
EstimatedTokens: section.EstimatedTokens,
SegmentCount: len(section.Segments),
}
}

1
internal/framework/outcomes/outcomes.go
View File

@@ -50,6 +50,7 @@ const (
ReasonInvalidOutput ReasonCode = "invalid_output"
ReasonValidationInputTooLarge ReasonCode = "validation_input_too_large"
ReasonValidationDecisionsInvalid ReasonCode = "validation_decisions_invalid"
ReasonInputInvalid ReasonCode = "input_invalid"
ReasonSetupFailure ReasonCode = "setup_failure"
ReasonCanceled ReasonCode = "canceled"
ReasonInternalFailure ReasonCode = "internal_failure"

4
internal/framework/processreport/report.go
View File

@@ -43,7 +43,7 @@ func Build(input BuildInput) reporting.Snapshot {
for i, spec := range input.ModuleSpecs {
snapshot.StageNames[i] = spec.InstanceName
}
if input.RunOutput != nil {
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 {
@@ -200,6 +200,8 @@ func safeMessage(reason outcomes.ReasonCode) string {
return "Processing was canceled."
case outcomes.ReasonSetupFailure:
return "Processing setup failed."
case outcomes.ReasonInputInvalid:
return "Input was invalid."
default:
return "Processing failed."
}

4
internal/framework/processreport/report_test.go
View File

@@ -23,7 +23,7 @@ func TestBuildPartialUnitsAndCombinedRecovery(t *testing.T) {
batch := 1
validation := outcomes.Operation{Reference: outcomes.Reference{Kind: outcomes.OperationValidation, ValidatorName: "review", ValidatorOrder: intPtr(1), BatchIndex: &batch}, FailureReason: outcomes.ReasonTransportFailed, ProposalIndices: []int{9, 3}}
inputFailure := outcomes.Operation{Reference: outcomes.Reference{Kind: outcomes.OperationValidationInput, ValidatorName: "review", ValidatorOrder: intPtr(1)}, FailureReason: outcomes.ReasonValidationInputTooLarge, ProposalIndices: []int{7}}
output := &runner.RunOutput{WorkUnits: []runner.WorkUnitOutcome{
output := &runner.RunOutput{InventoryAvailable: true, WorkUnits: []runner.WorkUnitOutcome{
{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}},
}}
@@ -64,7 +64,7 @@ func TestBuildInitializationNoWorkAndRunFailure(t *testing.T) {
if initialDoc.CompletedAt != nil || initialDoc.Summary.WorkUnitsTotal != nil || initialDoc.Errors == nil {
t.Fatalf("initial: %+v", initialDoc)
}
finished := Build(BuildInput{RunID: "r", Status: reporting.StatusCompleted, StartedAt: start, RunOutput: &runner.RunOutput{}})
finished := Build(BuildInput{RunID: "r", Status: reporting.StatusCompleted, StartedAt: start, RunOutput: &runner.RunOutput{InventoryAvailable: true}})
if finished.Summary.WorkUnitsTotal == nil || *finished.Summary.WorkUnitsTotal != 0 || finished.RecoveriesDocument().Recoveries == nil {
t.Fatalf("empty: %+v", finished)
}

3
internal/framework/runner/runner.go
View File

@@ -103,6 +103,7 @@ type RunOutput struct {
FinalTranscript *schema.Transcript `json:"-"`
ModuleResults []ModuleResult `json:"module_results"`
Sections []contracts.SectionMetadata `json:"sections"`
InventoryAvailable bool `json:"-"`
WorkUnits []WorkUnitOutcome `json:"work_units"`
Candidates []proposals.EnrichedCorrectionProposal `json:"-"`
Observations llm.ObservationSnapshot `json:"-"`
@@ -154,7 +155,7 @@ func (r *Runner) run(ctx context.Context, input RunInput) (RunOutput, error) {
snapshot := func() RunOutput {
return RunOutput{
FinalTranscript: transcriptFromSections(sections), ModuleResults: results,
Sections: inventory, WorkUnits: workUnits, Candidates: candidates,
Sections: inventory, InventoryAvailable: true, WorkUnits: workUnits, Candidates: candidates,
}
}