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

Add utilization diagnostics and correction ledger

Browse Source
This commit is contained in:
Eric Rakestraw
2026-05-13 19:03:37 +00:00
parent 037121e9ce
commit ff4ed82239
15 changed files with 969 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

157
internal/cli/review_artifacts.go Normal file
View File

@@ -0,0 +1,157 @@
package cli
import (
"path/filepath"
"sort"
"gitea.maximumdirect.net/eric/audita/internal/framework/runner"
)
const (
correctionDispositionApplied = "applied"
correctionDispositionRejected = "rejected"
correctionDispositionSkipped = "skipped"
correctionDispositionFailed = "failed"
)
type correctionLedgerEntry struct {
RunID string `json:"run_id,omitempty"`
ModuleKey string `json:"module_key"`
ModuleInstance string `json:"module_instance"`
ProposalIndex int `json:"proposal_index"`
SegmentID int `json:"segment_id,omitempty"`
OriginalText string `json:"original_text,omitempty"`
ProposedCorrectedText string `json:"proposed_corrected_text,omitempty"`
AppliedCorrectedText string `json:"applied_corrected_text,omitempty"`
ReplacementPolicy string `json:"replacement_policy,omitempty"`
Disposition string `json:"disposition"`
DispositionReasonCode string `json:"disposition_reason_code,omitempty"`
DispositionMessage string `json:"disposition_message,omitempty"`
DeterministicValidatorResults []ledgerValidatorDecisionRecord `json:"deterministic_validator_decisions,omitempty"`
LLMValidatorResults []ledgerValidatorDecisionRecord `json:"llm_validator_decisions,omitempty"`
}
type ledgerValidatorDecisionRecord struct {
ValidatorKey string `json:"validator_key"`
Approved bool `json:"approved"`
ReasonCode string `json:"reason_code"`
Message string `json:"message,omitempty"`
}
func buildCorrectionLedger(runDirPath string, runOutput *runner.RunOutput) []correctionLedgerEntry {
if runOutput == nil || len(runOutput.ModuleResults) == 0 {
return nil
}
runID := ""
if runDirPath != "" {
runID = filepath.Base(runDirPath)
}
entries := make([]correctionLedgerEntry, 0)
llmBacked := map[string]bool{
"spoken_form_plausibility": true,
"meaning_reversal_review": true,
"editorial_review": true,
"grammar_review": true,
"spoken_word_review": true,
}
for _, module := range runOutput.ModuleResults {
decisionsByProposal := make(map[int][]runner.ValidatorDecisionRecord)
for _, decision := range module.ValidatorDecisions {
decisionsByProposal[decision.ProposalIndex] = append(decisionsByProposal[decision.ProposalIndex], decision)
}
for _, change := range module.AppliedChanges {
entries = append(entries, correctionLedgerEntry{
RunID: runID,
ModuleKey: module.ModuleKey,
ModuleInstance: module.ModuleInstance,
ProposalIndex: change.ProposalIndex,
SegmentID: change.TargetSegmentID,
OriginalText: change.OriginalText,
ProposedCorrectedText: change.CorrectedText,
AppliedCorrectedText: change.CorrectedText,
ReplacementPolicy: string(module.ReplacementPolicy),
Disposition: correctionDispositionApplied,
DeterministicValidatorResults: filterLedgerDecisions(decisionsByProposal[change.ProposalIndex], false, llmBacked),
LLMValidatorResults: filterLedgerDecisions(decisionsByProposal[change.ProposalIndex], true, llmBacked),
})
}
for _, change := range module.SkippedChanges {
entries = append(entries, correctionLedgerEntry{
RunID: runID,
ModuleKey: module.ModuleKey,
ModuleInstance: module.ModuleInstance,
ProposalIndex: change.ProposalIndex,
SegmentID: change.TargetSegmentID,
OriginalText: change.OriginalText,
ProposedCorrectedText: change.CorrectedText,
ReplacementPolicy: string(module.ReplacementPolicy),
Disposition: correctionDispositionSkipped,
DispositionReasonCode: string(change.SkipReason),
DispositionMessage: change.Message,
DeterministicValidatorResults: filterLedgerDecisions(decisionsByProposal[change.ProposalIndex], false, llmBacked),
LLMValidatorResults: filterLedgerDecisions(decisionsByProposal[change.ProposalIndex], true, llmBacked),
})
}
for _, rejection := range module.ValidatorRejected {
entries = append(entries, correctionLedgerEntry{
RunID: runID,
ModuleKey: module.ModuleKey,
ModuleInstance: module.ModuleInstance,
ProposalIndex: rejection.ProposalIndex,
SegmentID: rejection.TargetSegmentID,
OriginalText: rejection.OriginalText,
ProposedCorrectedText: rejection.CorrectedText,
ReplacementPolicy: string(module.ReplacementPolicy),
Disposition: correctionDispositionRejected,
DispositionReasonCode: rejection.ReasonCode,
DispositionMessage: rejection.Message,
DeterministicValidatorResults: filterLedgerDecisions(decisionsByProposal[rejection.ProposalIndex], false, llmBacked),
LLMValidatorResults: filterLedgerDecisions(decisionsByProposal[rejection.ProposalIndex], true, llmBacked),
})
}
if module.Status == runner.ModuleStatusFailed {
entries = append(entries, correctionLedgerEntry{
RunID: runID,
ModuleKey: module.ModuleKey,
ModuleInstance: module.ModuleInstance,
Disposition: correctionDispositionFailed,
DispositionReasonCode: "module_failed",
DispositionMessage: module.ErrorMessage,
})
}
}
sort.SliceStable(entries, func(i, j int) bool {
if entries[i].ModuleInstance != entries[j].ModuleInstance {
return entries[i].ModuleInstance < entries[j].ModuleInstance
}
if entries[i].ProposalIndex != entries[j].ProposalIndex {
return entries[i].ProposalIndex < entries[j].ProposalIndex
}
return entries[i].Disposition < entries[j].Disposition
})
return entries
}
func filterLedgerDecisions(in []runner.ValidatorDecisionRecord, wantLLM bool, llmBacked map[string]bool) []ledgerValidatorDecisionRecord {
if len(in) == 0 {
return nil
}
out := make([]ledgerValidatorDecisionRecord, 0, len(in))
for _, decision := range in {
if llmBacked[decision.ValidatorName] != wantLLM {
continue
}
out = append(out, ledgerValidatorDecisionRecord{
ValidatorKey: decision.ValidatorName,
Approved: decision.Approved,
ReasonCode: decision.ReasonCode,
Message: decision.Message,
})
}
return out
}

28
internal/cli/run.go
View File

@@ -241,10 +241,15 @@ var processRunner = func(inv processInvocation, stdout io.Writer) (*normalizatio
runCtx, cancelRun := processRunnerContext()
defer cancelRun()
runnerResult, runErr := runner.New(moduleFactory).Run(runCtx, runner.RunInput{
Config: &inv.Config,
Transcript: normalizedTranscript,
Glossary: glossary,
ModuleSpecs: moduleSpecs,
Config: &inv.Config,
Transcript: normalizedTranscript,
Glossary: glossary,
ModuleSpecs: moduleSpecs,
EffectiveConcurrency: runner.EffectiveConcurrencyLimits{
TotalLLM: inv.Config.TotalLLMConcurrency,
ProposalLLM: inv.Config.EffectiveProposalLLMConcurrency(),
ValidationLLM: inv.Config.EffectiveValidationLLMConcurrency(),
},
ProposalLLMClient: proposalLLMClient,
ProposalLLMScheduler: proposalScheduler,
ProposalDiagnosticsDir: runDir.Path(),
@@ -523,6 +528,12 @@ func runProcess(args []string, stdout, stderr io.Writer) int {
completedAt := time.Now().UTC()
if runErr != nil {
if runDir != nil && runOutput != nil {
if runOutput.Utilization != nil {
_ = runDir.WriteJSONArtifact("utilization-diagnostics.json", runOutput.Utilization)
}
_ = runDir.WriteJSONArtifact("correction-ledger.json", buildCorrectionLedger(runDir.Path(), runOutput))
}
errorPhase, errorMessage := extractErrorPhase(runErr)
report := buildProcessReport("failed", inv, runDir, startedAt, completedAt, errorMessage, errorPhase, nil, nil, runOutput)
@@ -546,6 +557,13 @@ func runProcess(args []string, stdout, stderr io.Writer) int {
return 1
}
if runDir != nil && runOutput != nil {
if runOutput.Utilization != nil {
_ = runDir.WriteJSONArtifact("utilization-diagnostics.json", runOutput.Utilization)
}
_ = runDir.WriteJSONArtifact("correction-ledger.json", buildCorrectionLedger(runDir.Path(), runOutput))
}
report := buildProcessReport("success", inv, runDir, startedAt, completedAt, "", "", normSummary, chunkSummary, runOutput)
if strings.TrimSpace(inv.ReportJSONPath) != "" {
@@ -732,6 +750,8 @@ func buildProcessReport(status string, inv processInvocation, runDir *diagnostic
NormalizedTranscriptPath: filepath.Join(diagnosticsDir, "normalized-transcript.json"),
NormalizationSummaryPath: filepath.Join(diagnosticsDir, "normalization-summary.json"),
ChunkingSummaryPath: filepath.Join(diagnosticsDir, "chunking-summary.json"),
UtilizationSummaryPath: filepath.Join(diagnosticsDir, "utilization-diagnostics.json"),
CorrectionLedgerPath: filepath.Join(diagnosticsDir, "correction-ledger.json"),
InvocationMetadataPath: filepath.Join(diagnosticsDir, "invocation.json"),
RedactedEffectiveConfigPath: filepath.Join(diagnosticsDir, "effective-config.json"),
}

193
internal/cli/run_test.go
View File

@@ -1239,6 +1239,9 @@ func TestRunProcessReportJSONBestEffortOnFailure(t *testing.T) {
if report.Diagnostics.DirectoryPath == "" || report.Diagnostics.ErrorLogPath == "" {
t.Fatalf("expected diagnostics directory and error log references in failed report: %+v", report.Diagnostics)
}
if report.Diagnostics.UtilizationSummaryPath == "" || report.Diagnostics.CorrectionLedgerPath == "" {
t.Fatalf("expected review artifact paths in failed report diagnostics: %+v", report.Diagnostics)
}
if _, err := os.Stat(report.Diagnostics.ErrorLogPath); err != nil {
t.Fatalf("expected error log file at reported path: %v", err)
}
@@ -1312,6 +1315,153 @@ func TestRunProcessReportJSONAndRunDirReportShareDiagnosticsMetadata(t *testing.
}
}
func TestRunProcessWritesUtilizationAndCorrectionLedgerArtifacts(t *testing.T) {
var stdout bytes.Buffer
var stderr bytes.Buffer
workDir := t.TempDir()
reportPath := filepath.Join(t.TempDir(), "report.json")
exitCode := Run([]string{
"process",
fixturePath("tiny_transcript.json"),
"--glossary",
fixturePath("tiny_glossary.yaml"),
"--modules",
"grammar",
"--work-dir",
workDir,
"--work-dir-retention",
"always",
"--report-json",
reportPath,
}, &stdout, &stderr)
if exitCode != 0 {
t.Fatalf("expected success, got %d stderr=%q", exitCode, stderr.String())
}
report := readProcessReport(t, reportPath)
if report.Diagnostics == nil {
t.Fatalf("expected diagnostics metadata")
}
if report.Diagnostics.UtilizationSummaryPath == "" || report.Diagnostics.CorrectionLedgerPath == "" {
t.Fatalf("expected utilization and correction ledger paths, got %+v", report.Diagnostics)
}
if _, err := os.Stat(report.Diagnostics.UtilizationSummaryPath); err != nil {
t.Fatalf("expected utilization diagnostics artifact: %v", err)
}
if _, err := os.Stat(report.Diagnostics.CorrectionLedgerPath); err != nil {
t.Fatalf("expected correction ledger artifact: %v", err)
}
var utilization struct {
EffectiveConcurrency struct {
TotalLLM int `json:"total_llm"`
ProposalLLM int `json:"proposal_llm"`
ValidationLLM int `json:"validation_llm"`
} `json:"effective_concurrency"`
Modules []struct {
ModuleInstance string `json:"module_instance"`
} `json:"modules"`
LLMCalls struct {
TotalProposalCalls int `json:"total_proposal_calls"`
TotalValidationCalls int `json:"total_validation_calls"`
} `json:"llm_calls"`
}
if err := json.Unmarshal(readFile(t, report.Diagnostics.UtilizationSummaryPath), &utilization); err != nil {
t.Fatalf("unmarshal utilization artifact: %v", err)
}
if utilization.EffectiveConcurrency.TotalLLM <= 0 || utilization.EffectiveConcurrency.ProposalLLM <= 0 || utilization.EffectiveConcurrency.ValidationLLM <= 0 {
t.Fatalf("expected positive effective concurrency limits, got %+v", utilization.EffectiveConcurrency)
}
if len(utilization.Modules) == 0 || utilization.Modules[0].ModuleInstance == "" {
t.Fatalf("expected module-level timing summaries, got %+v", utilization.Modules)
}
if utilization.LLMCalls.TotalProposalCalls == 0 {
t.Fatalf("expected proposal LLM calls to be recorded")
}
var ledger []map[string]any
if err := json.Unmarshal(readFile(t, report.Diagnostics.CorrectionLedgerPath), &ledger); err != nil {
t.Fatalf("unmarshal correction ledger: %v", err)
}
}
func TestRunProcessCorrectionLedgerKeepsValidatorRejectionDistinctFromSkip(t *testing.T) {
var stdout bytes.Buffer
var stderr bytes.Buffer
workDir := t.TempDir()
reportPath := filepath.Join(t.TempDir(), "report.json")
proposalClient := &fakeStructuredLLMClient{
proposalResponses: []proposal_generation.StructuredCorrectionSet{
{
Corrections: []proposal_generation.StructuredCorrectionProposal{
{
TargetSegmentID: 1,
OriginalText: "hello",
CorrectedText: "hi",
Confidence: 0.99,
},
},
},
},
}
processProposalLLMClient = proposalClient
processValidationLLMClient = &fakeStructuredLLMClient{
validationResponses: []validators.LLMValidationResponse{
{
Validations: []validators.LLMValidationDecision{
{CorrectionIndex: 0, Approved: false, Confidence: 0.1, Reason: "reject"},
},
},
},
}
t.Cleanup(func() {
processProposalLLMClient = nil
processValidationLLMClient = nil
})
exitCode := Run([]string{
"process",
fixturePath("tiny_transcript.json"),
"--glossary",
fixturePath("tiny_glossary.yaml"),
"--modules",
"grammar",
"--work-dir",
workDir,
"--work-dir-retention",
"always",
"--report-json",
reportPath,
}, &stdout, &stderr)
if exitCode != 0 {
t.Fatalf("expected success, got %d stderr=%q", exitCode, stderr.String())
}
report := readProcessReport(t, reportPath)
var ledger []struct {
Disposition string `json:"disposition"`
DispositionReasonCode string `json:"disposition_reason_code"`
ModuleInstance string `json:"module_instance"`
}
if err := json.Unmarshal(readFile(t, report.Diagnostics.CorrectionLedgerPath), &ledger); err != nil {
t.Fatalf("unmarshal correction ledger: %v", err)
}
foundRejected := false
for _, entry := range ledger {
if entry.ModuleInstance == "grammar" && entry.Disposition == "rejected" {
foundRejected = true
if entry.DispositionReasonCode == "" {
t.Fatalf("expected rejection reason code in ledger entry")
}
}
}
if !foundRejected {
t.Fatalf("expected rejected ledger entry, got %+v", ledger)
}
}
func TestRunProcessReportJSONIncludesConfigVersionWhenConfigFileUsed(t *testing.T) {
var stdout bytes.Buffer
var stderr bytes.Buffer
@@ -2547,6 +2697,49 @@ func TestRunProcessDefaultFullPipelineAggregatesSkipsAndAutoRetentionKeepsRunDir
if runDirReport.ModulesSummary == nil || runDirReport.ModulesSummary.TotalSkippedChanges != 2 {
t.Fatalf("expected skipped totals in retained run-dir report, got %+v", runDirReport.ModulesSummary)
}
if runDirReport.Diagnostics == nil {
t.Fatalf("expected diagnostics metadata in run-dir report")
}
var utilization struct {
RunTiming struct {
SchedulerQueueWaitMS int64 `json:"scheduler_queue_wait_ms"`
LLMExecutionTimeMS int64 `json:"llm_execution_time_ms"`
DeterministicValidationMS int64 `json:"deterministic_validation_time_ms"`
} `json:"run_timing"`
Validators []struct {
ValidatorKey string `json:"validator_key"`
} `json:"validators"`
}
if err := json.Unmarshal(readFile(t, runDirReport.Diagnostics.UtilizationSummaryPath), &utilization); err != nil {
t.Fatalf("unmarshal utilization diagnostics: %v", err)
}
if utilization.RunTiming.LLMExecutionTimeMS < 0 || utilization.RunTiming.SchedulerQueueWaitMS < 0 || utilization.RunTiming.DeterministicValidationMS < 0 {
t.Fatalf("expected non-negative run timing fields, got %+v", utilization.RunTiming)
}
if len(utilization.Validators) == 0 {
t.Fatalf("expected per-validator timing summaries")
}
var ledger []struct {
Disposition string `json:"disposition"`
}
if err := json.Unmarshal(readFile(t, runDirReport.Diagnostics.CorrectionLedgerPath), &ledger); err != nil {
t.Fatalf("unmarshal correction ledger: %v", err)
}
foundSkipped := false
foundRejected := false
for _, entry := range ledger {
if entry.Disposition == "skipped" {
foundSkipped = true
}
if entry.Disposition == "rejected" {
foundRejected = true
}
}
if !foundSkipped || !foundRejected {
t.Fatalf("expected skipped and rejected ledger dispositions, got %+v", ledger)
}
diagFiles, globErr := filepath.Glob(filepath.Join(runDir, "*", "*response-payload.json"))
if globErr != nil {
t.Fatalf("glob diagnostics payloads: %v", globErr)

8
internal/cli/subprocess_test_hooks.go
View File

@@ -22,18 +22,22 @@ const (
// ConfigureSubprocessTestHooksFromEnv enables deterministic test-only hooks for
// subprocess integration tests that run through the Go test binary helper path.
func ConfigureSubprocessTestHooksFromEnv() {
mode := strings.TrimSpace(os.Getenv(subprocessTestLLMModeEnv))
timeoutMSRaw := strings.TrimSpace(os.Getenv(subprocessTestRunTimeoutMSEnv))
// Only activate in explicit subprocess test mode.
if mode == "" && timeoutMSRaw == "" {
return
}
if !shouldUseNoOpLLMClientForTests() {
return
}
mode := strings.TrimSpace(os.Getenv(subprocessTestLLMModeEnv))
if mode != "" {
client := &subprocessTestLLMClient{mode: mode}
processProposalLLMClient = client
processValidationLLMClient = client
}
timeoutMSRaw := strings.TrimSpace(os.Getenv(subprocessTestRunTimeoutMSEnv))
if timeoutMSRaw == "" {
return
}

13
internal/core/diagnostics/run_dir.go
View File

@@ -216,6 +216,19 @@ func (r *RunDirectory) WriteChunkingSummary(summary *chunking.DetailedSummary) e
return nil
}
func (r *RunDirectory) WriteJSONArtifact(name string, payload any) error {
artifactPath := filepath.Join(r.path, name)
bytes, err := json.MarshalIndent(payload, "", " ")
if err != nil {
return fmt.Errorf("failed to marshal %s: %w", name, err)
}
bytes = append(bytes, '\n')
if err := os.WriteFile(artifactPath, bytes, 0o644); err != nil {
return fmt.Errorf("failed to write %s: %w", name, err)
}
return nil
}
func (r *RunDirectory) ApplyRetention(input RetentionDecisionInput) error {
decision := input
if decision.RetentionMode == "" {

2
internal/core/reporting/report.go
View File

@@ -95,6 +95,8 @@ type DiagnosticsMetadata struct {
NormalizedTranscriptPath string `json:"normalized_transcript_path,omitempty"`
NormalizationSummaryPath string `json:"normalization_summary_path,omitempty"`
ChunkingSummaryPath string `json:"chunking_summary_path,omitempty"`
UtilizationSummaryPath string `json:"utilization_summary_path,omitempty"`
CorrectionLedgerPath string `json:"correction_ledger_path,omitempty"`
InvocationMetadataPath string `json:"invocation_metadata_path,omitempty"`
RedactedEffectiveConfigPath string `json:"redacted_effective_config_path,omitempty"`
ErrorLogPath string `json:"error_log_path,omitempty"`

354
internal/framework/runner/observability.go Normal file
View File

@@ -0,0 +1,354 @@
package runner
import (
"context"
"sort"
"strings"
"sync"
"time"
"gitea.maximumdirect.net/eric/audita/internal/framework/contracts"
frameworkvalidators "gitea.maximumdirect.net/eric/audita/internal/framework/validators"
)
type UtilizationDiagnostics struct {
EffectiveConcurrency EffectiveConcurrencyLimits `json:"effective_concurrency"`
RunTiming RunTimingSummary `json:"run_timing"`
LLMCalls LLMCallSummary `json:"llm_calls"`
Modules []ModuleTimingSummary `json:"modules,omitempty"`
Validators []ValidatorTimingSummary `json:"validators,omitempty"`
}
type EffectiveConcurrencyLimits struct {
TotalLLM int `json:"total_llm"`
ProposalLLM int `json:"proposal_llm"`
ValidationLLM int `json:"validation_llm"`
}
type RunTimingSummary struct {
RunWallTimeMS int64 `json:"run_wall_time_ms"`
SchedulerQueueWaitMS int64 `json:"scheduler_queue_wait_ms"`
LLMExecutionTimeMS int64 `json:"llm_execution_time_ms"`
DeterministicValidationMS int64 `json:"deterministic_validation_time_ms"`
MaxInFlightLLMCalls int `json:"max_in_flight_llm_calls"`
AverageInFlightLLMCalls int `json:"average_in_flight_llm_calls"`
}
type LLMCallSummary struct {
TotalProposalCalls int `json:"total_proposal_calls"`
TotalValidationCalls int `json:"total_validation_calls"`
}
type ModuleTimingSummary struct {
ModuleKey string `json:"module_key"`
ModuleInstance string `json:"module_instance"`
ProposalLLMCalls int `json:"proposal_llm_calls"`
ValidationLLMCalls int `json:"validation_llm_calls"`
SchedulerQueueWaitMS int64 `json:"scheduler_queue_wait_ms"`
LLMExecutionTimeMS int64 `json:"llm_execution_time_ms"`
DeterministicValidationMS int64 `json:"deterministic_validation_time_ms"`
ModuleWallTimeMS int64 `json:"module_wall_time_ms"`
}
type ValidatorTimingSummary struct {
ModuleKey string `json:"module_key"`
ModuleInstance string `json:"module_instance"`
ValidatorKey string `json:"validator_key"`
LLMBacked bool `json:"llm_backed"`
Calls int `json:"calls"`
ElapsedMS int64 `json:"elapsed_ms"`
}
type runInstrumentation struct {
startedAt time.Time
mu sync.Mutex
totalProposalCalls int
totalValidationCalls int
schedulerQueueWait time.Duration
llmExecutionTime time.Duration
deterministicValidation time.Duration
perModuleProposalCalls map[string]int
perModuleValidationCalls map[string]int
perModuleSchedulerQueueWait map[string]time.Duration
perModuleLLMExecutionTime map[string]time.Duration
perModuleDeterministicValidation map[string]time.Duration
validatorSummaries map[string]*ValidatorTimingSummary
inflight int
maxInflight int
inflightArea float64
lastInflightChange time.Time
averageInflightComputed bool
}
func newRunInstrumentation(startedAt time.Time) *runInstrumentation {
return &runInstrumentation{
startedAt: startedAt,
perModuleProposalCalls: make(map[string]int),
perModuleValidationCalls: make(map[string]int),
perModuleSchedulerQueueWait: make(map[string]time.Duration),
perModuleLLMExecutionTime: make(map[string]time.Duration),
perModuleDeterministicValidation: make(map[string]time.Duration),
validatorSummaries: make(map[string]*ValidatorTimingSummary),
lastInflightChange: startedAt,
}
}
func (i *runInstrumentation) wrapProposalClient(client contracts.StructuredLLMClient) contracts.StructuredLLMClient {
if client == nil {
return nil
}
return measuredClient{
inner: client,
collector: i,
isProposal: true,
}
}
func (i *runInstrumentation) wrapValidationClient(client contracts.StructuredLLMClient) contracts.StructuredLLMClient {
if client == nil {
return nil
}
return measuredClient{
inner: client,
collector: i,
isProposal: false,
}
}
func (i *runInstrumentation) wrapScheduler(s contracts.LLMScheduler, isProposal bool) contracts.LLMScheduler {
if s == nil {
return nil
}
return measuredScheduler{
inner: s,
collector: i,
isProposal: isProposal,
}
}
func (i *runInstrumentation) recordValidatorTiming(moduleKey, moduleInstance, validatorKey string, llmBacked bool, elapsed time.Duration) {
i.mu.Lock()
defer i.mu.Unlock()
if !llmBacked {
i.deterministicValidation += elapsed
i.perModuleDeterministicValidation[moduleInstance] += elapsed
}
k := moduleInstance + "::" + validatorKey
summary, ok := i.validatorSummaries[k]
if !ok {
summary = &ValidatorTimingSummary{
ModuleKey: moduleKey,
ModuleInstance: moduleInstance,
ValidatorKey: validatorKey,
LLMBacked: llmBacked,
}
i.validatorSummaries[k] = summary
}
summary.Calls++
summary.ElapsedMS += elapsed.Milliseconds()
}
func (i *runInstrumentation) recordModuleWallTime(moduleInstance string, elapsed time.Duration) {
i.mu.Lock()
defer i.mu.Unlock()
// Derived later from started/completed, but we keep a slot for sanity.
_ = moduleInstance
_ = elapsed
}
func (i *runInstrumentation) beginInFlight() {
now := time.Now().UTC()
i.mu.Lock()
defer i.mu.Unlock()
i.inflightArea += float64(i.inflight) * now.Sub(i.lastInflightChange).Seconds()
i.lastInflightChange = now
i.inflight++
if i.inflight > i.maxInflight {
i.maxInflight = i.inflight
}
}
func (i *runInstrumentation) endInFlight() {
now := time.Now().UTC()
i.mu.Lock()
defer i.mu.Unlock()
i.inflightArea += float64(i.inflight) * now.Sub(i.lastInflightChange).Seconds()
i.lastInflightChange = now
if i.inflight > 0 {
i.inflight--
}
}
func (i *runInstrumentation) finalize(runOutput *RunOutput, completedAt time.Time, limits EffectiveConcurrencyLimits) *UtilizationDiagnostics {
i.mu.Lock()
defer i.mu.Unlock()
if !i.averageInflightComputed {
i.inflightArea += float64(i.inflight) * completedAt.Sub(i.lastInflightChange).Seconds()
i.lastInflightChange = completedAt
i.averageInflightComputed = true
}
runSeconds := completedAt.Sub(i.startedAt).Seconds()
avgInflight := 0
if runSeconds > 0 {
avgInflight = int(i.inflightArea / runSeconds)
}
moduleSummaries := make([]ModuleTimingSummary, 0)
if runOutput != nil {
moduleSummaries = make([]ModuleTimingSummary, 0, len(runOutput.ModuleResults))
for _, moduleResult := range runOutput.ModuleResults {
moduleSummaries = append(moduleSummaries, ModuleTimingSummary{
ModuleKey: moduleResult.ModuleKey,
ModuleInstance: moduleResult.ModuleInstance,
ProposalLLMCalls: i.perModuleProposalCalls[moduleResult.ModuleInstance],
ValidationLLMCalls: i.perModuleValidationCalls[moduleResult.ModuleInstance],
SchedulerQueueWaitMS: i.perModuleSchedulerQueueWait[moduleResult.ModuleInstance].Milliseconds(),
LLMExecutionTimeMS: i.perModuleLLMExecutionTime[moduleResult.ModuleInstance].Milliseconds(),
DeterministicValidationMS: i.perModuleDeterministicValidation[moduleResult.ModuleInstance].Milliseconds(),
ModuleWallTimeMS: moduleResult.CompletedAt.Sub(moduleResult.StartedAt).Milliseconds(),
})
}
}
validatorSummaries := make([]ValidatorTimingSummary, 0, len(i.validatorSummaries))
for _, summary := range i.validatorSummaries {
validatorSummaries = append(validatorSummaries, *summary)
}
sortValidatorSummaries(validatorSummaries)
return &UtilizationDiagnostics{
EffectiveConcurrency: limits,
RunTiming: RunTimingSummary{
RunWallTimeMS: completedAt.Sub(i.startedAt).Milliseconds(),
SchedulerQueueWaitMS: i.schedulerQueueWait.Milliseconds(),
LLMExecutionTimeMS: i.llmExecutionTime.Milliseconds(),
DeterministicValidationMS: i.deterministicValidation.Milliseconds(),
MaxInFlightLLMCalls: i.maxInflight,
AverageInFlightLLMCalls: avgInflight,
},
LLMCalls: LLMCallSummary{
TotalProposalCalls: i.totalProposalCalls,
TotalValidationCalls: i.totalValidationCalls,
},
Modules: moduleSummaries,
Validators: validatorSummaries,
}
}
type measuredClient struct {
inner contracts.StructuredLLMClient
collector *runInstrumentation
isProposal bool
}
func (c measuredClient) CompleteStructured(ctx context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) {
if c.collector != nil {
c.collector.beginInFlight()
}
started := time.Now().UTC()
resp, err := c.inner.CompleteStructured(ctx, req, out)
elapsed := time.Since(started)
if c.collector != nil {
c.collector.endInFlight()
c.collector.mu.Lock()
moduleInstance := moduleInstanceFromStage(req.StageName)
c.collector.llmExecutionTime += elapsed
c.collector.perModuleLLMExecutionTime[moduleInstance] += elapsed
if c.isProposal {
c.collector.totalProposalCalls++
c.collector.perModuleProposalCalls[moduleInstance]++
} else {
c.collector.totalValidationCalls++
c.collector.perModuleValidationCalls[moduleInstance]++
}
c.collector.mu.Unlock()
}
return resp, err
}
type measuredScheduler struct {
inner contracts.LLMScheduler
collector *runInstrumentation
isProposal bool
}
func (s measuredScheduler) Run(ctx context.Context, fn func(context.Context) error) error {
started := time.Now().UTC()
var execDuration time.Duration
err := s.inner.Run(ctx, func(runCtx context.Context) error {
execStart := time.Now().UTC()
callErr := fn(runCtx)
execDuration += time.Since(execStart)
return callErr
})
totalDuration := time.Since(started)
waitDuration := totalDuration - execDuration
if waitDuration < 0 {
waitDuration = 0
}
if s.collector != nil {
stageModule := moduleInstanceFromContext(ctx)
s.collector.mu.Lock()
s.collector.schedulerQueueWait += waitDuration
if stageModule != "" {
s.collector.perModuleSchedulerQueueWait[stageModule] += waitDuration
}
s.collector.mu.Unlock()
}
_ = s.isProposal
return err
}
func moduleInstanceFromStage(stage string) string {
stage = strings.TrimSpace(stage)
if stage == "" {
return ""
}
parts := strings.Split(stage, ":")
if len(parts) == 0 {
return stage
}
return strings.TrimSpace(parts[0])
}
func moduleInstanceFromContext(ctx context.Context) string {
if ctx == nil {
return ""
}
if v := ctx.Value(moduleInstanceContextKey{}); v != nil {
if s, ok := v.(string); ok {
return strings.TrimSpace(s)
}
}
return ""
}
type moduleInstanceContextKey struct{}
func withModuleInstanceContext(ctx context.Context, moduleInstance string) context.Context {
if ctx == nil {
ctx = context.Background()
}
return context.WithValue(ctx, moduleInstanceContextKey{}, strings.TrimSpace(moduleInstance))
}
func isLLMBackedValidator(v contracts.Validator) bool {
if v == nil {
return false
}
_, ok := v.(*frameworkvalidators.LLMBackedValidator)
return ok
}
func sortValidatorSummaries(in []ValidatorTimingSummary) {
sort.SliceStable(in, func(i, j int) bool {
if in[i].ModuleInstance != in[j].ModuleInstance {
return in[i].ModuleInstance < in[j].ModuleInstance
}
return in[i].ValidatorKey < in[j].ValidatorKey
})
}

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

@@ -77,6 +77,7 @@ type RunInput struct {
Transcript *schema.Transcript
Glossary *schema.Glossary
ModuleSpecs []contracts.ModuleRunSpec
EffectiveConcurrency EffectiveConcurrencyLimits
ProposalLLMClient contracts.StructuredLLMClient
ProposalLLMScheduler contracts.LLMScheduler
ProposalDiagnosticsDir string
@@ -89,6 +90,7 @@ type RunInput struct {
type RunOutput struct {
FinalTranscript *schema.Transcript `json:"-"`
ModuleResults []ModuleResult `json:"module_results"`
Utilization *UtilizationDiagnostics
}
func New(factory ModuleFactory) *Runner {
@@ -100,6 +102,14 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
return RunOutput{}, fmt.Errorf("runner module factory is required")
}
startedAt := time.Now().UTC()
instrumentation := newRunInstrumentation(startedAt)
input.ProposalLLMClient = instrumentation.wrapProposalClient(input.ProposalLLMClient)
input.ValidationLLMClient = instrumentation.wrapValidationClient(input.ValidationLLMClient)
input.ProposalLLMScheduler = instrumentation.wrapScheduler(input.ProposalLLMScheduler, true)
input.ValidationLLMScheduler = instrumentation.wrapScheduler(input.ValidationLLMScheduler, false)
working := cloneTranscript(input.Transcript)
results := make([]ModuleResult, 0, len(input.ModuleSpecs))
@@ -116,7 +126,11 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
CompletedAt: time.Now().UTC(),
}
results = append(results, failed)
return RunOutput{FinalTranscript: working, ModuleResults: results}, fmt.Errorf("module %q setup failed: %w", spec.InstanceName, err)
return RunOutput{
FinalTranscript: working,
ModuleResults: results,
Utilization: instrumentation.finalize(&RunOutput{ModuleResults: results}, time.Now().UTC(), input.EffectiveConcurrency),
}, fmt.Errorf("module %q setup failed: %w", spec.InstanceName, err)
}
policy := module.ReplacementPolicy()
@@ -132,7 +146,11 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
CompletedAt: time.Now().UTC(),
}
results = append(results, failed)
return RunOutput{FinalTranscript: working, ModuleResults: results}, fmt.Errorf("module %q chunking failed: %w", spec.InstanceName, err)
return RunOutput{
FinalTranscript: working,
ModuleResults: results,
Utilization: instrumentation.finalize(&RunOutput{ModuleResults: results}, time.Now().UTC(), input.EffectiveConcurrency),
}, fmt.Errorf("module %q chunking failed: %w", spec.InstanceName, err)
}
pipelineResult, pipelineErr := runModulePipeline(ctx, collectSectionProposalsInput{
@@ -148,6 +166,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
ValidationClient: input.ValidationLLMClient,
ValidationScheduler: input.ValidationLLMScheduler,
ValidationDiagnosticsDir: input.ValidationDiagnosticsDir,
Instrumentation: instrumentation,
})
if pipelineErr != nil {
failed := ModuleResult{
@@ -163,7 +182,11 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
CompletedAt: time.Now().UTC(),
}
results = append(results, failed)
return RunOutput{FinalTranscript: working, ModuleResults: results}, fmt.Errorf("module %q failed: %w", spec.InstanceName, pipelineErr)
return RunOutput{
FinalTranscript: working,
ModuleResults: results,
Utilization: instrumentation.finalize(&RunOutput{ModuleResults: results}, time.Now().UTC(), input.EffectiveConcurrency),
}, fmt.Errorf("module %q failed: %w", spec.InstanceName, pipelineErr)
}
applyResult := proposals.ApplyProposals(working, pipelineResult.Approved, policy)
@@ -184,7 +207,9 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
})
}
return RunOutput{FinalTranscript: working, ModuleResults: results}, nil
output := RunOutput{FinalTranscript: working, ModuleResults: results}
output.Utilization = instrumentation.finalize(&output, time.Now().UTC(), input.EffectiveConcurrency)
return output, nil
}
type collectSectionProposalsInput struct {
@@ -200,6 +225,7 @@ type collectSectionProposalsInput struct {
ValidationClient contracts.StructuredLLMClient
ValidationScheduler ValidationScheduler
ValidationDiagnosticsDir string
Instrumentation *runInstrumentation
}
type sectionProposals struct {
@@ -244,7 +270,7 @@ func collectSectionProposals(ctx context.Context, input collectSectionProposalsI
go func() {
defer wg.Done()
meta := contracts.SectionMetadataFromSection(section)
corrected, err := input.Module.Propose(runCtx, contracts.ProposalRequest{
corrected, err := input.Module.Propose(withModuleInstanceContext(runCtx, input.Spec.InstanceName), contracts.ProposalRequest{
ExecutionContext: contracts.ExecutionContext{
Config: input.Config,
WorkingTranscript: transcriptFromSection(section),
@@ -366,6 +392,7 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput)
ValidationLLMClient: input.ValidationClient,
ValidationScheduler: input.ValidationScheduler,
DiagnosticsDir: input.ValidationDiagnosticsDir,
Instrumentation: input.Instrumentation,
})
validationResults <- sectionValidationResult{
sectionPos: sectionPos,
@@ -432,6 +459,7 @@ type validateSectionCandidatesInput struct {
ValidationLLMClient contracts.StructuredLLMClient
ValidationScheduler ValidationScheduler
DiagnosticsDir string
Instrumentation *runInstrumentation
}
type validateSectionCandidatesResult struct {
@@ -456,7 +484,8 @@ func validateSectionCandidates(ctx context.Context, input validateSectionCandida
}
}
vResult, err := validator.Validate(ctx, contracts.ValidationRequest{
validatorStartedAt := time.Now().UTC()
vResult, err := validator.Validate(withModuleInstanceContext(ctx, input.Spec.InstanceName), contracts.ValidationRequest{
WorkingTranscript: input.WorkingTranscript,
CandidateProposal: eligible,
ModuleKey: input.Spec.ModuleKey,
@@ -468,6 +497,15 @@ func validateSectionCandidates(ctx context.Context, input validateSectionCandida
Scheduler: input.ValidationScheduler,
DiagnosticsWriter: diagnosticsWriter,
})
if input.Instrumentation != nil {
input.Instrumentation.recordValidatorTiming(
input.Spec.ModuleKey,
input.Spec.InstanceName,
validator.Name(),
isLLMBackedValidator(validator),
time.Since(validatorStartedAt),
)
}
if err != nil {
return validateSectionCandidatesResult{
approved: eligible,