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

Complete Phase 7 runner orchestration

Browse Source
This commit is contained in:
Eric Rakestraw
2026-05-11 14:22:22 +00:00
parent 30606f5c49
commit 28fe899aa1
8 changed files with 760 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

114
internal/cli/run.go
View File

@@ -1,6 +1,7 @@
package cli
import (
"context"
"errors"
"flag"
"fmt"
@@ -16,6 +17,8 @@ import (
"gitea.maximumdirect.net/eric/audita/internal/core/normalization"
"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/runner"
)
type processInvocation struct {
@@ -26,15 +29,17 @@ type processInvocation struct {
Config config.Config
}
var processRunner = func(inv processInvocation, stdout io.Writer) (*normalization.NormalizationSummary, *chunking.Summary, *diagnostics.RunDirectory, error) {
var processModuleFactory runner.ModuleFactory
var processRunner = func(inv processInvocation, stdout io.Writer) (*normalization.NormalizationSummary, *chunking.Summary, *runner.RunOutput, *diagnostics.RunDirectory, error) {
runDir, err := diagnostics.NewRunDirectory(inv.Config.WorkDir, string(inv.Config.WorkDirRetention))
if err != nil {
return nil, nil, nil, fmt.Errorf("run_dir_creation: %w", err)
return nil, nil, nil, nil, fmt.Errorf("run_dir_creation: %w", err)
}
fail := func(phase string, err error) (*normalization.NormalizationSummary, *chunking.Summary, *diagnostics.RunDirectory, error) {
fail := func(phase string, err error, runOutput *runner.RunOutput) (*normalization.NormalizationSummary, *chunking.Summary, *runner.RunOutput, *diagnostics.RunDirectory, error) {
_ = runDir.WriteErrorLog(fmt.Sprintf("%s: %v", phase, err))
return nil, nil, runDir, fmt.Errorf("%s: %w", phase, err)
return nil, nil, runOutput, runDir, fmt.Errorf("%s: %w", phase, err)
}
if err := runDir.WriteInvocationMetadata(diagnostics.InvocationMetadata{
@@ -53,21 +58,22 @@ var processRunner = func(inv processInvocation, stdout io.Writer) (*normalizatio
transcriptBytes, err := coreio.ReadRequiredFile(inv.TranscriptPath, "transcript")
if err != nil {
return fail("transcript_read", err)
return fail("transcript_read", err, nil)
}
glossaryBytes, err := coreio.ReadRequiredFile(inv.GlossaryPath, "glossary")
if err != nil {
return fail("glossary_read", err)
return fail("glossary_read", err, nil)
}
sourceTranscript, err := schema.ParseSourceTranscriptJSON(transcriptBytes)
if err != nil {
return fail("transcript_schema", err)
return fail("transcript_schema", err, nil)
}
if _, err := schema.ParseGlossaryYAML(glossaryBytes); err != nil {
return fail("glossary_schema", err)
glossary, err := schema.ParseGlossaryYAML(glossaryBytes)
if err != nil {
return fail("glossary_schema", err, nil)
}
if err := runDir.WriteSourceTranscript(sourceTranscript, transcriptBytes); err != nil {
@@ -100,7 +106,7 @@ var processRunner = func(inv processInvocation, stdout io.Writer) (*normalizatio
sections, chunkErr := chunker.ChunkTranscript(normalizedTranscript)
if chunkErr != nil {
return fail("chunking", chunkErr)
return fail("chunking", chunkErr, nil)
}
chunkConfig := chunking.ChunkingConfig{
@@ -115,23 +121,44 @@ var processRunner = func(inv processInvocation, stdout io.Writer) (*normalizatio
_ = runDir.WriteErrorLog(fmt.Sprintf("chunking_summary: %v", err))
}
outputBytes, err := schema.TranscriptToJSON(normalizedTranscript)
workingTranscript := normalizedTranscript
var runOutput *runner.RunOutput
if processModuleFactory != nil {
moduleSpecs, err := contracts.ResolveModuleRunSpecs(inv.Config.Modules)
if err != nil {
return fail("runner_setup", err, nil)
}
runnerResult, runErr := runner.New(processModuleFactory).Run(context.Background(), runner.RunInput{
Config: &inv.Config,
Transcript: normalizedTranscript,
Glossary: glossary,
ModuleSpecs: moduleSpecs,
})
runOutput = &runnerResult
if runErr != nil {
return fail("runner_execution", runErr, runOutput)
}
workingTranscript = runnerResult.FinalTranscript
}
outputBytes, err := schema.TranscriptToJSON(workingTranscript)
if err != nil {
return fail("serialization", err)
return fail("serialization", err, runOutput)
}
if strings.TrimSpace(inv.OutputPath) != "" {
if err := coreio.WriteFile(inv.OutputPath, outputBytes); err != nil {
return fail("output_write", err)
return fail("output_write", err, runOutput)
}
return normSummary, &chunkSummary, runDir, nil
return normSummary, &chunkSummary, runOutput, runDir, nil
}
if _, err := stdout.Write(outputBytes); err != nil {
return fail("stdout_write", err)
return fail("stdout_write", err, runOutput)
}
return normSummary, &chunkSummary, runDir, nil
return normSummary, &chunkSummary, runOutput, runDir, nil
}
func sourceToCanonicalTranscript(source *schema.SourceTranscript) *schema.Transcript {
@@ -292,12 +319,12 @@ func runProcess(args []string, stdout, stderr io.Writer) int {
Config: cfg,
}
normSummary, chunkSummary, runDir, runErr := processRunner(inv, stdout)
normSummary, chunkSummary, runOutput, runDir, runErr := processRunner(inv, stdout)
completedAt := time.Now().UTC()
if runErr != nil {
errorPhase, errorMessage := extractErrorPhase(runErr)
report := buildProcessReport("failed", inv, runDir, startedAt, completedAt, errorMessage, errorPhase, nil, nil)
report := buildProcessReport("failed", inv, runDir, startedAt, completedAt, errorMessage, errorPhase, nil, nil, runOutput)
if strings.TrimSpace(inv.ReportJSONPath) != "" {
if err := reporting.WriteProcessReport(inv.ReportJSONPath, report); err != nil {
@@ -316,7 +343,7 @@ func runProcess(args []string, stdout, stderr io.Writer) int {
return 1
}
report := buildProcessReport("success", inv, runDir, startedAt, completedAt, "", "", normSummary, chunkSummary)
report := buildProcessReport("success", inv, runDir, startedAt, completedAt, "", "", normSummary, chunkSummary, runOutput)
if strings.TrimSpace(inv.ReportJSONPath) != "" {
if err := reporting.WriteProcessReport(inv.ReportJSONPath, report); err != nil {
@@ -331,11 +358,21 @@ func runProcess(args []string, stdout, stderr io.Writer) int {
}
}
hasSkippedCorrections := false
if runOutput != nil {
for _, mr := range runOutput.ModuleResults {
if len(mr.SkippedChanges) > 0 {
hasSkippedCorrections = true
break
}
}
}
if runDir != nil {
_ = runDir.WriteReport(report)
if err := runDir.ApplyRetention(diagnostics.RetentionDecisionInput{
RunSucceeded: true,
HasSkippedCorrections: false, // Hook for future module-level skipped-correction reporting.
HasSkippedCorrections: hasSkippedCorrections,
}); err != nil {
fmt.Fprintf(stderr, "audita process: failed to apply work-dir retention: %v\n", err)
return 1
@@ -356,9 +393,9 @@ func extractErrorPhase(err error) (phase string, message string) {
return "", msg
}
func buildProcessReport(status string, inv processInvocation, runDir *diagnostics.RunDirectory, startedAt, completedAt time.Time, errorMessage string, errorPhase string, normalizationSummary *normalization.NormalizationSummary, chunkingSummary *chunking.Summary) reporting.ProcessReport {
func buildProcessReport(status string, inv processInvocation, runDir *diagnostics.RunDirectory, startedAt, completedAt time.Time, errorMessage string, errorPhase string, normalizationSummary *normalization.NormalizationSummary, chunkingSummary *chunking.Summary, runOutput *runner.RunOutput) reporting.ProcessReport {
report := reporting.ProcessReport{
Phase: "phase3-chunking",
Phase: "phase7-runner",
Status: status,
Operation: "process",
TranscriptPath: inv.TranscriptPath,
@@ -409,9 +446,42 @@ func buildProcessReport(status string, inv processInvocation, runDir *diagnostic
MinSectionTokens: chunkingSummary.MinSectionTokens,
}
}
report.ModulesSummary, report.ModuleResults = buildModuleReporting(runOutput)
return report
}
func buildModuleReporting(runOutput *runner.RunOutput) (*reporting.ModulesSummary, []reporting.ModuleReport) {
if runOutput == nil || len(runOutput.ModuleResults) == 0 {
return nil, nil
}
moduleReports := make([]reporting.ModuleReport, 0, len(runOutput.ModuleResults))
summary := &reporting.ModulesSummary{ModuleCount: len(runOutput.ModuleResults)}
for _, r := range runOutput.ModuleResults {
startedAt := r.StartedAt
completedAt := r.CompletedAt
moduleReports = append(moduleReports, reporting.ModuleReport{
ModuleKey: r.ModuleKey,
ModuleInstance: r.ModuleInstance,
ReplacementPolicy: string(r.ReplacementPolicy),
Status: r.Status,
ProposalCount: r.ProposalCount,
AppliedChanges: r.AppliedChanges,
SkippedChanges: r.SkippedChanges,
ErrorMessage: r.ErrorMessage,
StartedAt: &startedAt,
CompletedAt: &completedAt,
})
summary.TotalAppliedChanges += len(r.AppliedChanges)
summary.TotalSkippedChanges += len(r.SkippedChanges)
if r.Status == runner.ModuleStatusFailed && summary.FailedModuleInstance == "" {
summary.FailedModuleInstance = r.ModuleInstance
}
}
return summary, moduleReports
}
type processFlags struct {
glossaryPath *string
outputPath *string

188
internal/cli/run_test.go
View File

@@ -2,7 +2,9 @@ package cli
import (
"bytes"
"context"
"encoding/json"
"errors"
"os"
"path/filepath"
"strings"
@@ -11,6 +13,8 @@ import (
"gitea.maximumdirect.net/eric/audita/internal/core/normalization"
"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/proposals"
)
func TestRunRootHelp(t *testing.T) {
@@ -609,8 +613,188 @@ func TestRunProcessReportJSONIncludesChunkingSummary(t *testing.T) {
if report.Chunking.MaxSectionTokens == 0 {
t.Errorf("expected max_section_tokens in report")
}
if report.Phase != "phase3-chunking" {
t.Errorf("expected phase 'phase3-chunking', got %q", report.Phase)
if report.Phase != "phase7-runner" {
t.Errorf("expected phase 'phase7-runner', got %q", report.Phase)
}
}
type fakeModuleFactory struct {
modules map[string]contracts.TranscriptModule
}
func (f fakeModuleFactory) ModuleForSpec(spec contracts.ModuleRunSpec) (contracts.TranscriptModule, error) {
m, ok := f.modules[spec.InstanceName]
if !ok {
return nil, errors.New("module not registered")
}
return m, nil
}
type fakeModule struct {
key string
policy proposals.ReplacementPolicy
proposeF func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error)
}
func (m fakeModule) Key() string { return m.key }
func (m fakeModule) ReplacementPolicy() proposals.ReplacementPolicy { return m.policy }
func (m fakeModule) Validators() []contracts.Validator { return nil }
func (m fakeModule) Propose(ctx context.Context, req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) {
if m.proposeF == nil {
return nil, nil
}
return m.proposeF(req)
}
func TestRunProcessInjectedFactoryExecutesRunnerAndReportsModules(t *testing.T) {
processModuleFactory = fakeModuleFactory{modules: map[string]contracts.TranscriptModule{
"m1": fakeModule{key: "m1", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) {
return []proposals.CorrectionProposal{
{TargetSegmentID: 1, OriginalText: "Hello", CorrectedText: "Hi", Confidence: 1},
}, nil
}},
"m2": fakeModule{key: "m2", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) {
if req.WorkingTranscript.Segments[0].Text != "Hi world" {
t.Fatalf("expected module 2 to see module 1 changes, got %q", req.WorkingTranscript.Segments[0].Text)
}
return []proposals.CorrectionProposal{
{TargetSegmentID: 1, OriginalText: "Hi", CorrectedText: "Hey", Confidence: 1},
{TargetSegmentID: 1, OriginalText: "missing", CorrectedText: "noop", Confidence: 1},
}, nil
}},
}}
t.Cleanup(func() { processModuleFactory = nil })
var stdout, stderr bytes.Buffer
workDir := t.TempDir()
reportPath := filepath.Join(t.TempDir(), "report.json")
outputPath := filepath.Join(t.TempDir(), "out.json")
transcriptPath := writeFile(t, "transcript.json", `[
{"id":1,"speaker":"Alice","start":0.0,"end":1.0,"text":"Hello world"}
]`)
exitCode := Run([]string{
"process", transcriptPath,
"--glossary", fixturePath("tiny_glossary.yaml"),
"--modules", "m1,m2",
"--output", outputPath,
"--report-json", reportPath,
"--work-dir", workDir,
"--work-dir-retention", "always",
}, &stdout, &stderr)
if exitCode != 0 {
t.Fatalf("expected success, got exit %d stderr=%q", exitCode, stderr.String())
}
if stdout.Len() != 0 {
t.Fatalf("expected empty stdout with --output, got %q", stdout.String())
}
out := readFile(t, outputPath)
parsed, err := schema.ParseTranscriptJSON(out)
if err != nil {
t.Fatalf("parse output transcript: %v", err)
}
if parsed.Segments[0].Text != "Hey world" {
t.Fatalf("expected runner-applied output text, got %q", parsed.Segments[0].Text)
}
report := readProcessReport(t, reportPath)
if report.ModulesSummary == nil || report.ModulesSummary.ModuleCount != 2 {
t.Fatalf("expected module summary for 2 modules, got %+v", report.ModulesSummary)
}
if report.ModulesSummary.TotalAppliedChanges != 2 || report.ModulesSummary.TotalSkippedChanges != 1 {
t.Fatalf("unexpected module totals: %+v", report.ModulesSummary)
}
if len(report.ModuleResults) != 2 {
t.Fatalf("expected 2 module results, got %d", len(report.ModuleResults))
}
if len(report.ModuleResults[1].SkippedChanges) != 1 {
t.Fatalf("expected skipped change for module 2, got %+v", report.ModuleResults[1].SkippedChanges)
}
runDirReport := readProcessReport(t, filepath.Join(onlyRunDir(t, workDir), "report.json"))
if len(runDirReport.ModuleResults) != 2 {
t.Fatalf("expected module results in run-dir report")
}
}
func TestRunProcessInjectedFactorySkippedKeepsAutoRetention(t *testing.T) {
processModuleFactory = fakeModuleFactory{modules: map[string]contracts.TranscriptModule{
"m1": fakeModule{key: "m1", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) {
return []proposals.CorrectionProposal{
{TargetSegmentID: 1, OriginalText: "word", CorrectedText: "term", Confidence: 1},
}, nil
}},
}}
t.Cleanup(func() { processModuleFactory = nil })
var stdout, stderr bytes.Buffer
workDir := t.TempDir()
transcriptPath := writeFile(t, "transcript.json", `[
{"id":1,"speaker":"Alice","start":0.0,"end":1.0,"text":"word word"}
]`)
exitCode := Run([]string{
"process", transcriptPath,
"--glossary", fixturePath("tiny_glossary.yaml"),
"--modules", "m1",
"--work-dir", workDir,
"--work-dir-retention", "auto",
}, &stdout, &stderr)
if exitCode != 0 {
t.Fatalf("expected success, got %d stderr=%q", exitCode, stderr.String())
}
if _, err := os.Stat(onlyRunDir(t, workDir)); err != nil {
t.Fatalf("expected run dir retained for skipped corrections under auto: %v", err)
}
}
func TestRunProcessInjectedFactoryFailureWritesFailedReport(t *testing.T) {
processModuleFactory = fakeModuleFactory{modules: map[string]contracts.TranscriptModule{
"m1": fakeModule{key: "m1", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) {
return nil, errors.New("test failure")
}},
}}
t.Cleanup(func() { processModuleFactory = nil })
var stdout, stderr bytes.Buffer
workDir := t.TempDir()
reportPath := filepath.Join(t.TempDir(), "report.json")
transcriptPath := writeFile(t, "transcript.json", `[
{"id":1,"speaker":"Alice","start":0.0,"end":1.0,"text":"Hello"}
]`)
exitCode := Run([]string{
"process", transcriptPath,
"--glossary", fixturePath("tiny_glossary.yaml"),
"--modules", "m1",
"--work-dir", workDir,
"--work-dir-retention", "always",
"--report-json", reportPath,
}, &stdout, &stderr)
if exitCode == 0 {
t.Fatal("expected failure exit code")
}
if stdout.Len() != 0 {
t.Fatalf("expected empty stdout on failure, got %q", stdout.String())
}
if !strings.Contains(stderr.String(), "runner_execution") {
t.Fatalf("expected runner_execution failure on stderr, got %q", stderr.String())
}
report := readProcessReport(t, reportPath)
if report.Status != "failed" {
t.Fatalf("expected failed report status, got %q", report.Status)
}
if report.ModulesSummary == nil || report.ModulesSummary.FailedModuleInstance == "" {
t.Fatalf("expected failed module summary, got %+v", report.ModulesSummary)
}
if len(report.ModuleResults) != 1 || report.ModuleResults[0].Status != "failed" {
t.Fatalf("expected failed module result, got %+v", report.ModuleResults)
}
runPath := onlyRunDir(t, workDir)
if _, err := os.Stat(filepath.Join(runPath, "error.log")); err != nil {
t.Fatalf("expected error.log for failed runner module: %v", err)
}
}