From ff4ed82239833ebd4fedfc15ed3c42abc285e187 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Wed, 13 May 2026 19:03:37 +0000 Subject: [PATCH] Add utilization diagnostics and correction ledger --- README.md | 3 + docs/architecture.md | 57 ++++ docs/diagnostics.md | 98 ++++++ docs/public-contract.md | 4 + docs/roadmap.md | 11 + docs/subprocess-operations.md | 3 +- docs/validators.md | 1 + internal/cli/review_artifacts.go | 157 +++++++++ internal/cli/run.go | 28 +- internal/cli/run_test.go | 193 +++++++++++ internal/cli/subprocess_test_hooks.go | 8 +- internal/core/diagnostics/run_dir.go | 13 + internal/core/reporting/report.go | 2 + internal/framework/runner/observability.go | 354 +++++++++++++++++++++ internal/framework/runner/runner.go | 50 ++- 15 files changed, 969 insertions(+), 13 deletions(-) create mode 100644 docs/diagnostics.md create mode 100644 internal/cli/review_artifacts.go create mode 100644 internal/framework/runner/observability.go diff --git a/README.md b/README.md index b3593b4..dc2c2cf 100644 --- a/README.md +++ b/README.md @@ -283,6 +283,8 @@ Per-run diagnostics include: - normalized transcript artifact - normalization summary - chunking summary +- utilization diagnostics summary +- correction ledger - invocation metadata - redacted effective config - module/validator prompt-response diagnostics @@ -295,5 +297,6 @@ Optional external report output: ## Documentation - Architecture: [`docs/architecture.md`](docs/architecture.md) +- Diagnostics: [`docs/diagnostics.md`](docs/diagnostics.md) - Structured LLM adapter: [`docs/structured-llm.md`](docs/structured-llm.md) - Subprocess operations: [`docs/subprocess-operations.md`](docs/subprocess-operations.md) diff --git a/docs/architecture.md b/docs/architecture.md index 23656d7..4ff6c29 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -646,6 +646,8 @@ Current per-run artifacts include: - `normalized-transcript.json` - `normalization-summary.json` - `chunking-summary.json` +- `utilization-diagnostics.json` +- `correction-ledger.json` - `invocation.json` - `effective-config.json` (redacted credentials) - `report.json` @@ -660,6 +662,8 @@ Current process reports include diagnostics metadata references for: - normalized transcript artifact path; - normalization summary artifact path; - chunking summary artifact path; +- utilization diagnostics artifact path; +- correction ledger artifact path; - invocation metadata artifact path; - redacted effective-config artifact path; - error-log artifact path on failure. @@ -675,6 +679,59 @@ Current process reports also include: - report schema version; - selected output schema; - config file version when config file input is used. +- review/observability artifacts: + - run-level and module-level utilization/timing summaries; + - flattened correction ledger entries for applied/rejected/skipped/failed correction dispositions. + +Utilization diagnostics collection: +- collection is performed in the runner path via lightweight instrumentation around LLM scheduler and structured-client execution (`internal/framework/runner`); +- instrumentation is observational only and does not change scheduler acquisition/release semantics or module execution order; +- serialized artifact: `utilization-diagnostics.json`. + +Utilization diagnostics high-level shape: +- `effective_concurrency`: + - `total_llm`, `proposal_llm`, `validation_llm`; +- `run_timing`: + - `run_wall_time_ms`; + - `scheduler_queue_wait_ms`; + - `llm_execution_time_ms`; + - `deterministic_validation_time_ms`; + - `max_in_flight_llm_calls`; + - `average_in_flight_llm_calls`; +- `llm_calls`: + - `total_proposal_calls`; + - `total_validation_calls`; +- `modules`: + - per-module key/instance timing summaries including module wall time and per-module call counts; +- `validators`: + - per-validator summaries keyed by stable validator key with elapsed time and LLM-backed marker. + +Correction ledger construction: +- ledger entries are built from runner module results in the CLI report/diagnostics path (`internal/cli/review_artifacts.go`); +- serialized artifact: `correction-ledger.json`; +- one flattened record per applied/validator-rejected/application-skipped outcome where data is available, plus module-failed records for failed module instances. + +Correction ledger high-level shape: +- run/module/proposal identity: + - `run_id`, `module_key`, `module_instance`, `proposal_index`, `segment_id`; +- correction payload: + - `original_text`, `proposed_corrected_text`, `applied_corrected_text` (when applied), `replacement_policy`; +- disposition: + - `disposition` in `{applied,rejected,skipped,failed}`; + - `disposition_reason_code`, `disposition_message`; +- validator decision snapshots: + - `deterministic_validator_decisions[]`; + - `llm_validator_decisions[]`; + - each decision uses stable validator keys and reason codes. + +Identity and metadata boundaries: +- stable module keys/instance names and stable validator keys are included directly in ledger records; +- prompt metadata and structured response schema metadata remain in LLM interaction diagnostics payloads and are not duplicated into every ledger row; +- reports reference artifact paths for utilization and ledger files through diagnostics metadata. + +Redaction and retention: +- secret redaction guarantees continue to apply to diagnostics/report artifacts; +- utilization and ledger artifacts are emitted within the existing run-directory retention model (`auto|always|never`) and are retained/removed with the run directory. Current report schema metadata values: - `report_metadata.report_schema_name = "audita-process-report"` diff --git a/docs/diagnostics.md b/docs/diagnostics.md new file mode 100644 index 0000000..594c982 --- /dev/null +++ b/docs/diagnostics.md @@ -0,0 +1,98 @@ +# Audita Diagnostics + +This document describes the run-directory diagnostics artifacts produced by `audita process`. + +## Purpose + +Diagnostics provide machine-readable run context and execution artifacts for: +- failure debugging; +- validator/correction review; +- post-run performance analysis. + +Diagnostics are written under the configured work directory (`--work-dir`) when run-directory initialization succeeds. + +## Core artifacts + +Typical artifacts in each run directory: +- `source-transcript.json` +- `source-transcript-parsed.json` +- `normalized-transcript.json` +- `normalization-summary.json` +- `chunking-summary.json` +- `invocation.json` +- `effective-config.json` (redacted) +- module/validator LLM interaction artifacts +- `report.json` +- `error.log` on failure + +## Utilization diagnostics artifact + +Artifact: +- `utilization-diagnostics.json` + +High-level fields: +- `effective_concurrency`: + - total/proposal/validation LLM concurrency limits in effect. +- `run_timing`: + - run wall time; + - scheduler queue wait time; + - LLM execution time; + - deterministic validator time; + - max/average in-flight LLM calls. +- `llm_calls`: + - total proposal and validation LLM call counts. +- `modules`: + - module-level timing summaries. +- `validators`: + - per-validator timing summaries keyed by stable validator key. + +## Correction ledger artifact + +Artifact: +- `correction-ledger.json` + +Ledger records are flattened review entries derived from module results and include: +- module/proposal identity (`module_key`, `module_instance`, `proposal_index`, `segment_id`); +- correction text fields and replacement policy when available; +- disposition: + - `applied` + - `rejected` + - `skipped` + - `failed` +- stable reason codes/messages; +- deterministic and LLM validator decision snapshots using stable validator keys. + +Validator rejection and proposal-application skip are distinct dispositions. + +## Report references + +`report.json` and optional `--report-json` output include diagnostics metadata paths for: +- utilization diagnostics artifact; +- correction ledger artifact; +- existing transcript/normalization/chunking/invocation/effective-config artifacts. + +## Retention behavior + +Run-directory retention follows configured policy: +- `always`: keep all run directories; +- `never`: keep successful run directories; +- `auto`: keep failed runs and successful runs with skipped/rejected corrections. + +## Redaction guarantees + +API keys and other configured secrets are redacted from: +- `effective-config.json`; +- LLM interaction diagnostics artifacts; +- reports and surfaced errors. + +## Debugging guide + +When debugging: +- slow runs: + - inspect `utilization-diagnostics.json` (`run_timing`, `modules`, `validators`, in-flight metrics). +- validator rejections: + - inspect `correction-ledger.json` rejected entries and matching validator decisions; + - inspect validator response diagnostics payloads. +- application skips: + - inspect `correction-ledger.json` skipped entries and skip reason codes; + - compare with validator decisions to distinguish validation rejection vs apply-time skip. diff --git a/docs/public-contract.md b/docs/public-contract.md index 7a01ab5..ffcc548 100644 --- a/docs/public-contract.md +++ b/docs/public-contract.md @@ -13,6 +13,7 @@ This contract covers: - stable validator key identifiers in report/diagnostics records - prompt metadata identifiers in diagnostics - diagnostics directory behavior +- utilization diagnostics and correction-ledger artifact presence/pathing in diagnostics metadata - stdout/stderr and exit-code behavior - secret redaction guarantees - compatibility and deprecation policy @@ -92,6 +93,7 @@ Current values: `--report-json` output and diagnostics run-dir `report.json` use the same report schema metadata. Validator decision/rejection records in reports use stable validator keys in `validator_name`. +Report diagnostics metadata includes artifact-path fields for utilization diagnostics and correction ledger when diagnostics initialization succeeds. ## Diagnostics directory behavior @@ -99,6 +101,8 @@ When diagnostics directory creation succeeds, Audita writes run artifacts includ - invocation metadata - redacted effective config - transcript/normalization/chunking artifacts +- utilization diagnostics (`utilization-diagnostics.json`) +- correction ledger (`correction-ledger.json`) - report and failure error log (when applicable) - module/LLM diagnostics artifacts as available diff --git a/docs/roadmap.md b/docs/roadmap.md index 68f1fd6..8aaa872 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -811,6 +811,17 @@ Improve operational observability so Audita runs can be debugged and performance This phase should follow the registry/prompt work so diagnostics can include stable module, validator, prompt, and schema identifiers. +## Implementation status (2026-05-13) + +This workstream is now implemented: +- run diagnostics include `utilization-diagnostics.json` with run-level and module-level timing/utilization data; +- utilization diagnostics include per-validator timing summaries keyed by stable validator keys; +- process reports include diagnostics references for utilization and correction-ledger artifacts; +- run diagnostics include `correction-ledger.json` with flattened correction disposition records (`applied`, `rejected`, `skipped`, `failed`) keyed by stable module instance and proposal index; +- successful runs emit both artifacts; failure paths emit partial artifacts when runner state exists. + +This status update applies only to utilization diagnostics and correction-ledger artifacts. Release-hardening evaluation fixtures and later roadmap items remain planned. + ## Scheduler utilization diagnostics Add module-level and run-level metrics for LLM scheduling and execution. diff --git a/docs/subprocess-operations.md b/docs/subprocess-operations.md index 5fb7131..d36ff44 100644 --- a/docs/subprocess-operations.md +++ b/docs/subprocess-operations.md @@ -46,11 +46,12 @@ Recommended additions: - Run-directory `report.json` is written independently under diagnostics. - Best-effort failure reports are emitted when possible without masking the primary failure. - Report write failures return nonzero with clear stderr messaging. +- Report diagnostics metadata references run-directory artifacts including utilization diagnostics and correction ledger paths when available. ## Diagnostics directory behavior - Each run creates (when possible) a per-run diagnostics directory. -- Typical artifacts include transcript, normalization, chunking, invocation, effective config, LLM diagnostics, `report.json`, and `error.log` on failure. +- Typical artifacts include transcript, normalization, chunking, invocation, effective config, LLM diagnostics, `utilization-diagnostics.json`, `correction-ledger.json`, `report.json`, and `error.log` on failure. - Failed runs retain diagnostics. - Under `auto` retention, successful runs with skipped/rejected corrections are retained; clean successful runs may be removed. diff --git a/docs/validators.md b/docs/validators.md index a6b496d..b693c33 100644 --- a/docs/validators.md +++ b/docs/validators.md @@ -103,6 +103,7 @@ These are separate outcomes and are reported separately. - report validator decision/rejection entries use stable validator keys in `validator_name`. - validator LLM diagnostics include validator identity in interaction metadata and structured response schema metadata. +- correction ledger entries include deterministic and LLM validator decision snapshots keyed by the same stable validator keys, and keep validator rejection distinct from application-level skip. ## Configurable knobs that remain supported diff --git a/internal/cli/review_artifacts.go b/internal/cli/review_artifacts.go new file mode 100644 index 0000000..3c57545 --- /dev/null +++ b/internal/cli/review_artifacts.go @@ -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 +} diff --git a/internal/cli/run.go b/internal/cli/run.go index 26af62a..32f4687 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -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"), } diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index 1e8e848..d22f045 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -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) diff --git a/internal/cli/subprocess_test_hooks.go b/internal/cli/subprocess_test_hooks.go index a3c8e02..35306f2 100644 --- a/internal/cli/subprocess_test_hooks.go +++ b/internal/cli/subprocess_test_hooks.go @@ -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 } diff --git a/internal/core/diagnostics/run_dir.go b/internal/core/diagnostics/run_dir.go index 064cb92..1c03303 100644 --- a/internal/core/diagnostics/run_dir.go +++ b/internal/core/diagnostics/run_dir.go @@ -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 == "" { diff --git a/internal/core/reporting/report.go b/internal/core/reporting/report.go index 3a300fc..faf8244 100644 --- a/internal/core/reporting/report.go +++ b/internal/core/reporting/report.go @@ -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"` diff --git a/internal/framework/runner/observability.go b/internal/framework/runner/observability.go new file mode 100644 index 0000000..8eb82d7 --- /dev/null +++ b/internal/framework/runner/observability.go @@ -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 + }) +} diff --git a/internal/framework/runner/runner.go b/internal/framework/runner/runner.go index 0cbd1ba..401f277 100644 --- a/internal/framework/runner/runner.go +++ b/internal/framework/runner/runner.go @@ -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,