Fix the five library adoption review findings

This commit is contained in:
2026-09-24 15:32:13 +00:00
parent c42bc01052
commit 665ee85215
19 changed files with 942 additions and 208 deletions

View File

@@ -3,7 +3,6 @@ package cli
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"strings"
@@ -40,7 +39,7 @@ type processInvocation struct {
}
var processRunner = func(inv processInvocation, stdout io.Writer) (*normalization.NormalizationSummary, *chunking.Summary, *runner.RunOutput, *diagnostics.RunDirectory, error) {
runDir, err := diagnostics.NewRunDirectory(inv.Config.Diagnostics.WorkDir, string(inv.Config.Diagnostics.Retention))
runDir, err := diagnostics.NewRunDirectory(inv.Config.Diagnostics.WorkDir, string(inv.Config.Diagnostics.Retention), llm.ConfiguredSecrets(&inv.Config)...)
if err != nil {
return nil, nil, nil, nil, fmt.Errorf("run_dir_creation: %w", err)
}
@@ -74,12 +73,12 @@ var processRunner = func(inv processInvocation, stdout io.Writer) (*normalizatio
_ = runDir.WriteErrorLog(fmt.Sprintf("effective_config: %v", err))
}
if origins, err := llm.ExplanationYAML(inv.ConfigResult); err == nil {
_ = runDir.WriteArtifact(diagnostics.ArtifactConfigOrigins, origins)
_ = runDir.WriteYAMLArtifact(diagnostics.ArtifactConfigOrigins, origins)
}
}
if inv.LLMRuntime != nil {
if profiles, err := inv.LLMRuntime.SelectedProfilesYAML(); err == nil {
_ = runDir.WriteArtifact(diagnostics.ArtifactSelectedProfiles, profiles)
_ = runDir.WriteYAMLArtifact(diagnostics.ArtifactSelectedProfiles, profiles)
}
}
@@ -320,10 +319,9 @@ func runProcess(args []string, stdout, stderr io.Writer) int {
errorPhase, errorMessage := extractErrorPhase(runErr)
errorMessage = llm.RedactError(fmt.Errorf("%s", errorMessage), inv.Config)
report := processreport.Build(processReportInput("failed", inv, runDir, startedAt, completedAt, errorMessage, errorPhase, nil, nil, runOutput))
report = redactProcessReport(report, inv.Config)
if strings.TrimSpace(inv.ReportJSONPath) != "" {
if err := reporting.WriteProcessReport(inv.ReportJSONPath, report); err != nil {
if err := reporting.WriteProcessReport(inv.ReportJSONPath, report, llm.ConfiguredSecrets(&inv.Config)...); err != nil {
fmt.Fprintf(stderr, "audita process: %v\n", err)
}
}
@@ -352,10 +350,9 @@ func runProcess(args []string, stdout, stderr io.Writer) int {
}
report := processreport.Build(processReportInput("success", inv, runDir, startedAt, completedAt, "", "", normSummary, chunkSummary, runOutput))
report = redactProcessReport(report, inv.Config)
if strings.TrimSpace(inv.ReportJSONPath) != "" {
if err := reporting.WriteProcessReport(inv.ReportJSONPath, report); err != nil {
if err := reporting.WriteProcessReport(inv.ReportJSONPath, report, llm.ConfiguredSecrets(&inv.Config)...); err != nil {
if runDir != nil {
_ = runDir.WriteErrorLog(fmt.Sprintf("report_write: %v", err))
_ = runDir.ApplyRetention(diagnostics.RetentionDecisionInput{
@@ -389,24 +386,6 @@ func writeExecutionArtifacts(runDir *diagnostics.RunDirectory, output *runner.Ru
_ = runDir.WriteJSONArtifact(diagnostics.ArtifactUtilizationSummary, processreport.BuildUtilization(output, cfg.Backends))
}
func redactProcessReport(report reporting.ProcessReport, cfg config.Config) reporting.ProcessReport {
raw, err := json.Marshal(report)
if err != nil {
return report
}
content := string(raw)
for _, secret := range llm.ConfiguredSecrets(&cfg) {
content = strings.ReplaceAll(content, secret, "[REDACTED]")
encoded, _ := json.Marshal(secret)
content = strings.ReplaceAll(content, string(encoded[1:len(encoded)-1]), "[REDACTED]")
}
var safe reporting.ProcessReport
if json.Unmarshal([]byte(content), &safe) != nil {
return report
}
return safe
}
func runConfig(args []string, stdout, stderr io.Writer) int {
if len(args) == 0 || isHelpCommand(args) || args[0] == "--help" || args[0] == "-h" {
writeConfigUsage(stdout)

View File

@@ -1,7 +1,6 @@
package diagnostics
import (
"encoding/json"
"fmt"
"os"
"path/filepath"
@@ -9,12 +8,14 @@ import (
"gitea.maximumdirect.net/eric/audita/internal/core/chunking"
"gitea.maximumdirect.net/eric/audita/internal/core/normalization"
"gitea.maximumdirect.net/eric/audita/internal/core/redaction"
"gitea.maximumdirect.net/eric/audita/internal/core/reporting"
"gitea.maximumdirect.net/eric/audita/internal/core/schema"
)
// RunDirectory represents a per-run diagnostics directory
type RunDirectory struct {
redactor redaction.Redactor
path string
retention string
createdAt time.Time
@@ -63,7 +64,7 @@ type InvocationMetadata struct {
}
// NewRunDirectory creates a new run directory under the configured work dir
func NewRunDirectory(workDir, retention string) (*RunDirectory, error) {
func NewRunDirectory(workDir, retention string, secrets ...string) (*RunDirectory, error) {
if workDir == "" {
workDir = ".audita-runs"
}
@@ -83,6 +84,7 @@ func NewRunDirectory(workDir, retention string) (*RunDirectory, error) {
return &RunDirectory{
path: runPath,
redactor: redaction.New(secrets),
retention: retention,
createdAt: time.Now().UTC(),
}, nil
@@ -107,7 +109,7 @@ func (r *RunDirectory) WriteInvocationMetadata(metadata InvocationMetadata) erro
}
path := filepath.Join(r.path, ArtifactInvocationMetadata)
bytes, err := json.MarshalIndent(metadata, "", " ")
bytes, err := r.redactor.JSON(metadata)
if err != nil {
return fmt.Errorf("failed to marshal invocation metadata: %w", err)
}
@@ -120,11 +122,7 @@ func (r *RunDirectory) WriteInvocationMetadata(metadata InvocationMetadata) erro
// WriteEffectiveConfig writes redacted effective config metadata for this run.
func (r *RunDirectory) WriteEffectiveConfig(effectiveYAML []byte) error {
path := filepath.Join(r.path, ArtifactEffectiveConfig)
if err := os.WriteFile(path, effectiveYAML, 0o644); err != nil {
return fmt.Errorf("failed to write effective config: %w", err)
}
return nil
return r.WriteYAMLArtifact(ArtifactEffectiveConfig, effectiveYAML)
}
// WriteSourceTranscript writes the source transcript artifact
@@ -137,7 +135,7 @@ func (r *RunDirectory) WriteSourceTranscript(transcript *schema.SourceTranscript
// Write parsed source for debugging
parsedPath := filepath.Join(r.path, ArtifactParsedSourceTranscript)
parsedBytes, err := json.MarshalIndent(transcript, "", " ")
parsedBytes, err := r.redactor.JSON(transcript)
if err != nil {
return fmt.Errorf("failed to marshal parsed source transcript: %w", err)
}
@@ -165,7 +163,7 @@ func (r *RunDirectory) WriteNormalizedTranscript(transcript *schema.Transcript)
// WriteNormalizationSummary writes the normalization summary artifact
func (r *RunDirectory) WriteNormalizationSummary(summary *normalization.NormalizationSummary) error {
summaryPath := filepath.Join(r.path, ArtifactNormalizationSummary)
bytes, err := json.MarshalIndent(summary, "", " ")
bytes, err := r.redactor.JSON(summary)
if err != nil {
return fmt.Errorf("failed to marshal normalization summary: %w", err)
}
@@ -179,7 +177,7 @@ func (r *RunDirectory) WriteNormalizationSummary(summary *normalization.Normaliz
// WriteReport writes the authoritative report artifact
func (r *RunDirectory) WriteReport(report reporting.ProcessReport) error {
reportPath := filepath.Join(r.path, ArtifactReport)
bytes, err := json.MarshalIndent(report, "", " ")
bytes, err := r.redactor.JSON(report)
if err != nil {
return fmt.Errorf("failed to marshal report: %w", err)
}
@@ -193,13 +191,13 @@ func (r *RunDirectory) WriteReport(report reporting.ProcessReport) error {
// WriteErrorLog writes an error log on failure
func (r *RunDirectory) WriteErrorLog(errorMessage string) error {
errorPath := filepath.Join(r.path, ArtifactErrorLog)
return os.WriteFile(errorPath, []byte(errorMessage+"\n"), 0o644)
return os.WriteFile(errorPath, []byte(r.redactor.String(errorMessage)+"\n"), 0o644)
}
// WriteChunkingSummary writes the chunking summary artifact
func (r *RunDirectory) WriteChunkingSummary(summary *chunking.DetailedSummary) error {
summaryPath := filepath.Join(r.path, ArtifactChunkingSummary)
bytes, err := json.MarshalIndent(summary, "", " ")
bytes, err := r.redactor.JSON(summary)
if err != nil {
return fmt.Errorf("failed to marshal chunking summary: %w", err)
}
@@ -212,7 +210,7 @@ func (r *RunDirectory) WriteChunkingSummary(summary *chunking.DetailedSummary) e
func (r *RunDirectory) WriteJSONArtifact(name string, payload any) error {
artifactPath := filepath.Join(r.path, name)
bytes, err := json.MarshalIndent(payload, "", " ")
bytes, err := r.redactor.JSON(payload)
if err != nil {
return fmt.Errorf("failed to marshal %s: %w", name, err)
}
@@ -223,7 +221,13 @@ func (r *RunDirectory) WriteJSONArtifact(name string, payload any) error {
return nil
}
func (r *RunDirectory) WriteArtifact(name string, payload []byte) error {
// WriteYAMLArtifact writes a sanitized YAML diagnostic artifact.
func (r *RunDirectory) WriteYAMLArtifact(name string, payload []byte) error {
var err error
payload, err = r.redactor.YAML(payload)
if err != nil {
return err
}
if err := os.WriteFile(filepath.Join(r.path, name), payload, 0o644); err != nil {
return fmt.Errorf("failed to write %s: %w", name, err)
}

View File

@@ -0,0 +1,123 @@
// Package redaction sanitizes diagnostic representations without modifying domain data.
package redaction
import (
"bytes"
"encoding/json"
"fmt"
"sort"
"strings"
"gopkg.in/yaml.v3"
)
// Redactor replaces configured values in a single pass, longest matches first.
// Its zero value is usable and leaves strings unchanged.
type Redactor struct{ replacer *strings.Replacer }
func New(secrets []string) Redactor {
values := map[string]bool{}
for _, secret := range secrets {
if secret == "" {
continue
}
values[secret] = true
// Provider content may itself be serialized JSON inside a string field.
encoded, _ := json.Marshal(secret)
values[string(encoded[1:len(encoded)-1])] = true
}
keys := make([]string, 0, len(values))
for value := range values {
keys = append(keys, value)
}
sort.Slice(keys, func(i, j int) bool {
if len(keys[i]) == len(keys[j]) {
return keys[i] < keys[j]
}
return len(keys[i]) > len(keys[j])
})
pairs := make([]string, 0, 2*len(keys))
for _, key := range keys {
pairs = append(pairs, key, "[REDACTED]")
}
return Redactor{replacer: strings.NewReplacer(pairs...)}
}
func (r Redactor) String(value string) string {
if r.replacer == nil {
return value
}
return r.replacer.Replace(value)
}
// JSON serializes first, then redacts only decoded string values and keys.
// Numbers retain their exact representation. Failure never returns original data.
func (r Redactor) JSON(value any) ([]byte, error) {
raw, err := json.Marshal(value)
if err != nil {
return nil, fmt.Errorf("serialize diagnostic: %s", r.String(err.Error()))
}
decoder := json.NewDecoder(bytes.NewReader(raw))
decoder.UseNumber()
var decoded any
if err := decoder.Decode(&decoded); err != nil {
return nil, fmt.Errorf("decode diagnostic: %s", r.String(err.Error()))
}
safe, err := r.walk(decoded)
if err != nil {
return nil, err
}
return json.MarshalIndent(safe, "", " ")
}
func (r Redactor) walk(value any) (any, error) {
switch v := value.(type) {
case string:
return r.String(v), nil
case []any:
for i := range v {
var err error
v[i], err = r.walk(v[i])
if err != nil {
return nil, err
}
}
case map[string]any:
safe := make(map[string]any, len(v))
for key, item := range v {
safeKey := r.String(key)
if _, exists := safe[safeKey]; exists {
return nil, fmt.Errorf("redaction would produce duplicate object keys")
}
safeItem, err := r.walk(item)
if err != nil {
return nil, err
}
safe[safeKey] = safeItem
}
return safe, nil
}
return value, nil
}
// YAML preserves scalar types while sanitizing string nodes, including map keys.
func (r Redactor) YAML(raw []byte) ([]byte, error) {
var node yaml.Node
if err := yaml.Unmarshal(raw, &node); err != nil {
return nil, fmt.Errorf("decode diagnostic YAML: %s", r.String(err.Error()))
}
var visit func(*yaml.Node)
visit = func(n *yaml.Node) {
if n.Kind == yaml.ScalarNode && n.Tag == "!!str" {
n.Value = r.String(n.Value)
}
n.HeadComment = r.String(n.HeadComment)
n.LineComment = r.String(n.LineComment)
n.FootComment = r.String(n.FootComment)
for _, child := range n.Content {
visit(child)
}
}
visit(&node)
return yaml.Marshal(&node)
}

View File

@@ -0,0 +1,92 @@
package redaction_test
import (
"encoding/json"
"math"
"os"
"path/filepath"
"strings"
"testing"
"gitea.maximumdirect.net/eric/audita/internal/core/diagnostics"
"gitea.maximumdirect.net/eric/audita/internal/core/redaction"
"gitea.maximumdirect.net/eric/audita/internal/core/reporting"
)
func TestStringRedactionPreservesSerializationAndDomainData(t *testing.T) {
secrets := []string{"2", `quoted"\key`, "overlap", "overlap-long", "REDACTED"}
r := redaction.New(secrets)
original := map[string]any{"number": 2, "large": json.Number("9007199254740993"), "bool": true, "text": `echo:2 quoted"\key overlap-long overlap`}
raw, err := r.JSON(original)
if err != nil {
t.Fatal(err)
}
var decoded map[string]any
if err := json.Unmarshal(raw, &decoded); err != nil {
t.Fatal(err)
}
if decoded["number"] != float64(2) || decoded["bool"] != true || !strings.Contains(string(raw), "9007199254740993") {
t.Fatalf("changed scalar meanings: %s", raw)
}
if decoded["text"] != "echo:[REDACTED] [REDACTED] [REDACTED] [REDACTED]" {
t.Fatalf("redaction: %s", raw)
}
if original["text"] != `echo:2 quoted"\key overlap-long overlap` {
t.Fatal("mutated domain data")
}
if raw, err := r.JSON(math.NaN()); err == nil || raw != nil {
t.Fatalf("unsafe fallback: %s %v", raw, err)
}
if raw, err := r.JSON(map[string]string{"overlap": "first", "overlap-long": "second"}); err == nil || raw != nil {
t.Fatalf("colliding redacted keys: %s %v", raw, err)
}
nested, _ := json.Marshal(map[string]string{"reason": `quoted"\key`})
raw, err = r.JSON(map[string]string{"final_content": string(nested)})
if err != nil || strings.Contains(string(raw), "key") {
t.Fatalf("nested content: %s %v", raw, err)
}
yaml, err := r.YAML([]byte("version: 2\nenabled: true\nmessage: echo:2\n"))
if err != nil || !strings.Contains(string(yaml), "version: 2") || !strings.Contains(string(yaml), "echo:[REDACTED]") {
t.Fatalf("YAML: %s %v", yaml, err)
}
}
func TestReportWritersRedactStringsAndFailClosed(t *testing.T) {
dir := t.TempDir()
version := 2
report := reporting.ProcessReport{ErrorMessage: "echo:2"}
report.ReportMetadata.ConfigVersion = &version
run, err := diagnostics.NewRunDirectory(dir, "always", "2")
if err != nil {
t.Fatal(err)
}
external := filepath.Join(dir, "external.json")
if err := reporting.WriteProcessReport(external, report, "2"); err != nil {
t.Fatal(err)
}
if err := run.WriteReport(report); err != nil {
t.Fatal(err)
}
for _, path := range []string{external, filepath.Join(run.Path(), diagnostics.ArtifactReport)} {
raw, err := os.ReadFile(path)
if err != nil {
t.Fatal(err)
}
var got reporting.ProcessReport
if err := json.Unmarshal(raw, &got); err != nil {
t.Fatal(err)
}
if got.ErrorMessage != "echo:[REDACTED]" || got.ReportMetadata.ConfigVersion == nil || *got.ReportMetadata.ConfigVersion != 2 {
t.Fatalf("report: %+v", got)
}
}
if report.ErrorMessage != "echo:2" {
t.Fatal("mutated input report")
}
if err := run.WriteJSONArtifact("invalid.json", map[string]any{"secret": "echo:2", "invalid": math.NaN()}); err == nil {
t.Fatal("expected serialization failure")
}
if _, err := os.Stat(filepath.Join(run.Path(), "invalid.json")); !os.IsNotExist(err) {
t.Fatalf("failed artifact persisted: %v", err)
}
}

View File

@@ -1,11 +1,11 @@
package reporting
import (
"encoding/json"
"fmt"
"os"
"time"
"gitea.maximumdirect.net/eric/audita/internal/core/redaction"
"gitea.maximumdirect.net/eric/audita/internal/framework/proposals"
stagewarnings "gitea.maximumdirect.net/eric/audita/internal/framework/warnings"
)
@@ -124,8 +124,8 @@ type ChunkingSummary struct {
MinSectionTokens int `json:"min_section_tokens"`
}
func WriteProcessReport(path string, report ProcessReport) error {
payload, err := json.MarshalIndent(report, "", " ")
func WriteProcessReport(path string, report ProcessReport, secrets ...string) error {
payload, err := redaction.New(secrets).JSON(report)
if err != nil {
return fmt.Errorf("failed to marshal process report JSON: %w", err)
}

View File

@@ -1,15 +1,14 @@
package llm
import (
"encoding/json"
"fmt"
"os"
"path/filepath"
"regexp"
"strings"
)
const redactedSecret = "[REDACTED]"
"gitea.maximumdirect.net/eric/audita/internal/core/redaction"
)
var stageSanitizer = regexp.MustCompile(`[^a-zA-Z0-9_.-]+`)
@@ -30,18 +29,7 @@ type DiagnosticsWriter struct {
// NewDiagnosticsWriter creates a diagnostics writer rooted at dir.
func NewDiagnosticsWriter(dir string, secrets []string) *DiagnosticsWriter {
filtered := make([]string, 0, len(secrets))
for _, secret := range secrets {
secret = strings.TrimSpace(secret)
if secret != "" {
filtered = append(filtered, secret)
}
}
return &DiagnosticsWriter{
dir: dir,
secrets: filtered,
}
return &DiagnosticsWriter{dir: dir, secrets: append([]string(nil), secrets...)}
}
// WriteInteraction writes request/response/error artifacts for a generic LLM
@@ -94,33 +82,13 @@ func sanitizeStageName(stage string) string {
}
func (w *DiagnosticsWriter) writeJSON(path string, payload any) error {
raw, err := json.MarshalIndent(payload, "", " ")
redacted, err := redaction.New(w.secrets).JSON(payload)
if err != nil {
return fmt.Errorf("marshal json artifact %q: %w", filepath.Base(path), err)
}
redacted := redactJSONSecrets(raw, w.secrets)
if !json.Valid(redacted) {
return fmt.Errorf("redacted json artifact %q is invalid", filepath.Base(path))
}
redacted = append(redacted, '\n')
if err := os.WriteFile(path, redacted, 0o644); err != nil {
return fmt.Errorf("write json artifact %q: %w", filepath.Base(path), err)
}
return nil
}
func redactJSONSecrets(raw []byte, secrets []string) []byte {
if len(secrets) == 0 {
return raw
}
result := string(raw)
for _, secret := range secrets {
if secret == "" {
continue
}
result = strings.ReplaceAll(result, "Bearer "+secret, "Bearer "+redactedSecret)
result = strings.ReplaceAll(result, secret, redactedSecret)
}
return []byte(result)
}

View File

@@ -61,7 +61,7 @@ func TestDiagnosticsWriterRedactsSecrets(t *testing.T) {
if strings.Contains(content, secret) {
t.Fatalf("secret leaked in %s: %s", filepath.Base(path), content)
}
if !strings.Contains(content, redactedSecret) {
if !strings.Contains(content, "[REDACTED]") {
t.Fatalf("redaction marker missing in %s: %s", filepath.Base(path), content)
}
if !json.Valid(raw) {

View File

@@ -12,6 +12,7 @@ import (
"sync/atomic"
"gitea.maximumdirect.net/eric/audita/internal/core/config"
"gitea.maximumdirect.net/eric/audita/internal/core/redaction"
"gitea.maximumdirect.net/eric/audita/internal/framework/responseschema"
"gitea.maximumdirect.net/eric/audita/internal/prompts"
pk "gitea.maximumdirect.net/eric/promptkit"
@@ -95,16 +96,5 @@ func RedactError(err error, c config.Config) string {
if err == nil {
return ""
}
message := err.Error()
var secrets []string
for _, backend := range c.Backends {
if backend.Credential != nil && backend.Credential.Value() != "" {
secrets = append(secrets, backend.Credential.Value())
}
}
sort.Slice(secrets, func(i, j int) bool { return len(secrets[i]) > len(secrets[j]) })
for _, secret := range secrets {
message = strings.ReplaceAll(message, secret, "[REDACTED]")
}
return message
return redaction.New(ConfiguredSecrets(&c)).String(err.Error())
}

View File

@@ -6,6 +6,7 @@ import (
ck "gitea.maximumdirect.net/eric/ConfigKit"
"gitea.maximumdirect.net/eric/audita/internal/core/config"
"gitea.maximumdirect.net/eric/audita/internal/core/redaction"
"gopkg.in/yaml.v3"
)
@@ -45,10 +46,14 @@ func (r *Runtime) SelectedProfilesYAML() ([]byte, error) {
if err != nil {
return nil, err
}
return yaml.Marshal(struct {
raw, err := yaml.Marshal(struct {
Proposal selectedProfileView `yaml:"proposal"`
Validation selectedProfileView `yaml:"validation"`
}{proposal, validation})
if err != nil {
return nil, err
}
return redaction.New(ConfiguredSecrets(&r.Config)).YAML(raw)
}
type derivationView struct {

View File

@@ -29,86 +29,74 @@ func BuildUtilization(output *runner.RunOutput, backends map[string]config.Backe
return result
}
result.RunWallTimeMS = output.RunWallTimeMS
for id, backend := range backends {
group := llm.ObservationSnapshot{}
for _, call := range output.Observations.Calls {
if call.Prompt.BackendID == id {
group.Calls = append(group.Calls, call)
addModuleCall(&result, call)
groups := make(map[string]*llm.ObservationSnapshot, len(backends))
callBackends := make(map[uint64]string, len(output.Observations.Calls))
moduleIndices := make(map[string]int)
moduleIndex := func(key, instance string) int {
if i, ok := moduleIndices[instance]; ok {
return i
}
i := len(result.Modules)
moduleIndices[instance] = i
result.Modules = append(result.Modules, reporting.ModuleUtilization{ModuleKey: key, ModuleInstance: instance})
return i
}
// Seed configured order, including modules that made no LLM calls.
for _, module := range output.ModuleResults {
i := moduleIndex(module.ModuleKey, module.ModuleInstance)
result.Modules[i].ModuleWallTimeMS = module.CompletedAt.Sub(module.StartedAt).Milliseconds()
}
for _, call := range output.Observations.Calls {
id := call.Prompt.BackendID
if groups[id] == nil {
groups[id] = &llm.ObservationSnapshot{}
}
groups[id].Calls = append(groups[id].Calls, call)
callBackends[call.CallID] = id
if call.Context.ModuleInstance != "" {
i := moduleIndex(call.Context.ModuleKey, call.Context.ModuleInstance)
switch call.Context.Role {
case "proposal":
result.Modules[i].ProposalLLMCalls++
case "validation":
result.Modules[i].ValidationLLMCalls++
}
}
}
for _, event := range output.Observations.Events {
if id, ok := callBackends[event.CallID]; ok {
groups[id].Events = append(groups[id].Events, event)
}
}
for id, backend := range backends {
group := groups[id]
if group == nil {
group = &llm.ObservationSnapshot{}
}
var wait, generation, retry, wall time.Duration
for _, call := range group.Calls {
wall += call.WallDuration
}
for _, event := range output.Observations.Events {
if hasCall(group.Calls, event.CallID) {
group.Events = append(group.Events, event)
if event.Kind == string(pk.ExecutionFinished) {
wait += event.CapacityWait
generation += event.GenerationDuration
retry += event.RetryDelay
}
for _, event := range group.Events {
if event.Kind == string(pk.ExecutionFinished) {
wait += event.CapacityWait
generation += event.GenerationDuration
retry += event.RetryDelay
}
}
result.Backends[id] = reporting.BackendUtilization{
Concurrency: backend.Concurrency, QueueCapacity: backend.QueueCapacity,
LLMSummary: mapLLMSummary(llm.Summarize(group)),
LLMSummary: mapLLMSummary(llm.Summarize(*group)),
CapacityWaitMS: wait.Milliseconds(), GenerationMS: generation.Milliseconds(),
RetryDelayMS: retry.Milliseconds(), LogicalWallMS: wall.Milliseconds(),
}
}
for _, module := range output.ModuleResults {
found := false
for i := range result.Modules {
if result.Modules[i].ModuleInstance == module.ModuleInstance {
result.Modules[i].ModuleWallTimeMS = module.CompletedAt.Sub(module.StartedAt).Milliseconds()
found = true
break
}
}
if !found {
result.Modules = append(result.Modules, reporting.ModuleUtilization{ModuleKey: module.ModuleKey, ModuleInstance: module.ModuleInstance, ModuleWallTimeMS: module.CompletedAt.Sub(module.StartedAt).Milliseconds()})
}
}
for _, timing := range output.ValidatorTimings {
result.Validators = append(result.Validators, reporting.ValidatorUtilization{ModuleKey: timing.ModuleKey, ModuleInstance: timing.ModuleInstance, ValidatorKey: timing.ValidatorName, LLMBacked: timing.LLMBacked, Calls: timing.Calls, ElapsedMS: timing.ElapsedMS})
}
return result
}
func hasCall(calls []llm.LogicalCall, id uint64) bool {
for _, call := range calls {
if call.CallID == id {
return true
}
}
return false
}
func addModuleCall(result *reporting.UtilizationDiagnostics, call llm.LogicalCall) {
if call.Context.ModuleInstance == "" {
return
}
for i := range result.Modules {
if result.Modules[i].ModuleInstance == call.Context.ModuleInstance {
if call.Context.Role == "proposal" {
result.Modules[i].ProposalLLMCalls++
} else if call.Context.Role == "validation" {
result.Modules[i].ValidationLLMCalls++
}
return
}
}
module := reporting.ModuleUtilization{ModuleKey: call.Context.ModuleKey, ModuleInstance: call.Context.ModuleInstance}
if call.Context.Role == "proposal" {
module.ProposalLLMCalls++
} else if call.Context.Role == "validation" {
module.ValidationLLMCalls++
}
result.Modules = append(result.Modules, module)
}
type executionCall struct {
CallID uint64 `json:"call_id"`
Context structuredoutput.DiagnosticContext `json:"context"`
@@ -150,13 +138,12 @@ type executionEvent struct {
func BuildLLMExecution(snapshot llm.ObservationSnapshot) any {
calls := make([]executionCall, 0, len(snapshot.Calls))
events := make([]executionEvent, 0, len(snapshot.Events))
eventsByCall := make(map[uint64][]llm.ExecutionObservation, len(snapshot.Calls))
for _, event := range snapshot.Events {
eventsByCall[event.CallID] = append(eventsByCall[event.CallID], event)
}
for _, call := range snapshot.Calls {
group := llm.ObservationSnapshot{Calls: []llm.LogicalCall{call}}
for _, event := range snapshot.Events {
if event.CallID == call.CallID {
group.Events = append(group.Events, event)
}
}
group := llm.ObservationSnapshot{Calls: []llm.LogicalCall{call}, Events: eventsByCall[call.CallID]}
var finished *time.Time
if !call.Finished.IsZero() {
value := call.Finished

View File

@@ -2,6 +2,7 @@ package processreport
import (
"encoding/json"
"fmt"
"strings"
"testing"
"time"
@@ -54,3 +55,72 @@ func TestReportSummaryAvailability(t *testing.T) {
t.Fatalf("initialized empty summary: %+v", got)
}
}
func BenchmarkExecutionReports(b *testing.B) {
for _, n := range []int{1000, 10000, 20000} {
snapshot := llm.ObservationSnapshot{}
for i := 0; i < n; i++ {
snapshot.Calls = append(snapshot.Calls, llm.LogicalCall{CallID: uint64(i + 1), Prompt: structuredoutput.PromptMetadata{BackendID: "local"}, Context: structuredoutput.DiagnosticContext{ModuleInstance: "grammar", Role: "proposal"}})
snapshot.Events = append(snapshot.Events, llm.ExecutionObservation{CallID: uint64(i + 1), Kind: string(pk.ExecutionFinished), Attempts: 1})
}
output := &runner.RunOutput{Observations: snapshot}
b.Run(fmt.Sprintf("execution/%d", n), func(b *testing.B) {
for i := 0; i < b.N; i++ {
BuildLLMExecution(snapshot)
}
})
b.Run(fmt.Sprintf("utilization/%d", n), func(b *testing.B) {
for i := 0; i < b.N; i++ {
BuildUtilization(output, map[string]config.Backend{"local": {Concurrency: 1}})
}
})
}
}
func TestIndexedReportsPreserveOrderAndPartialUsage(t *testing.T) {
snapshot := llm.ObservationSnapshot{
Calls: []llm.LogicalCall{
{CallID: 3, Context: structuredoutput.DiagnosticContext{Role: "validation", ModuleInstance: "grammar_2"}, Prompt: structuredoutput.PromptMetadata{BackendID: "b"}},
{CallID: 1, Context: structuredoutput.DiagnosticContext{Role: "proposal", ModuleInstance: "grammar_1"}, Prompt: structuredoutput.PromptMetadata{BackendID: "a"}},
{CallID: 2, Context: structuredoutput.DiagnosticContext{Role: "validation", ModuleInstance: "grammar_1"}, Prompt: structuredoutput.PromptMetadata{BackendID: "a"}},
},
Events: []llm.ExecutionObservation{
{CallID: 1, Kind: string(pk.ExecutionFinished), Attempts: 1, Usage: llm.Usage{Total: llm.TokenCount{Value: 7, Known: true, Complete: true}}},
{CallID: 3, Kind: string(pk.ExecutionFinished), Attempts: 1},
{CallID: 2, Kind: string(pk.ExecutionFinished), Attempts: 1},
},
}
raw, err := json.Marshal(BuildLLMExecution(snapshot))
if err != nil {
t.Fatal(err)
}
var execution struct {
Calls []executionCall `json:"calls"`
Events []executionEvent `json:"events"`
}
if err := json.Unmarshal(raw, &execution); err != nil {
t.Fatal(err)
}
for i, call := range snapshot.Calls {
if execution.Calls[i].CallID != call.CallID {
t.Fatal("call order changed")
}
}
for i, event := range snapshot.Events {
if execution.Events[i].CallID != event.CallID {
t.Fatal("event order changed")
}
}
if execution.Calls[1].Summary.Usage.Total.Value != 7 || !execution.Calls[1].Summary.Usage.Total.Complete || execution.Calls[0].Summary.Usage.Total.Known {
t.Fatalf("call usage: %+v", execution.Calls)
}
output := &runner.RunOutput{Observations: snapshot}
utilization := BuildUtilization(output, map[string]config.Backend{"a": {}, "b": {}})
total := utilization.Backends["a"].Usage.Total
if total.Value != 7 || !total.Known || total.Complete || utilization.Backends["a"].ProviderAttempts != 2 {
t.Fatalf("partial backend usage: %+v", utilization)
}
if len(utilization.Modules) != 2 || utilization.Modules[1].ProposalLLMCalls != 1 || utilization.Modules[1].ValidationLLMCalls != 1 {
t.Fatalf("module grouping: %+v", utilization.Modules)
}
}

View File

@@ -0,0 +1,145 @@
package runner
import (
"context"
"errors"
"net/http"
"os"
"path/filepath"
"sync/atomic"
"testing"
"time"
ck "gitea.maximumdirect.net/eric/ConfigKit"
"gitea.maximumdirect.net/eric/audita/internal/core/chunking"
"gitea.maximumdirect.net/eric/audita/internal/core/config"
"gitea.maximumdirect.net/eric/audita/internal/core/schema"
"gitea.maximumdirect.net/eric/audita/internal/framework/contracts"
"gitea.maximumdirect.net/eric/audita/internal/framework/llm"
"gitea.maximumdirect.net/eric/audita/internal/framework/proposals"
"gitea.maximumdirect.net/eric/audita/internal/framework/responseschema"
"gitea.maximumdirect.net/eric/audita/internal/framework/structuredoutput"
"gitea.maximumdirect.net/eric/audita/internal/framework/validators"
"gitea.maximumdirect.net/eric/audita/internal/prompts"
pk "gitea.maximumdirect.net/eric/promptkit"
)
type contextValidator struct {
validate func(context.Context, validators.Request) (validators.Result, error)
}
func (v contextValidator) Name() string { return "cancel-test" }
func (v contextValidator) Validate(ctx context.Context, req validators.Request) (validators.Result, error) {
return v.validate(ctx, req)
}
type cancelTransport struct {
entered chan struct{}
calls atomic.Int32
}
func (t *cancelTransport) RoundTrip(req *http.Request) (*http.Response, error) {
t.calls.Add(1)
t.entered <- struct{}{}
<-req.Context().Done()
return nil, req.Context().Err()
}
func TestValidatorFailureCancelsActiveAndPendingProviderWork(t *testing.T) {
dir := t.TempDir()
if err := os.Mkdir(filepath.Join(dir, "profiles"), 0700); err != nil {
t.Fatal(err)
}
main := filepath.Join(dir, "config.yml")
body := "version: 2\nbackends:\n shared:\n endpoint: http://provider.invalid/v1\n concurrency: 1\n queue_capacity: 1\nllm:\n proposal_profile: base\n transient_retries: 0\n repair_attempts: 0\n"
if err := os.WriteFile(main, []byte(body), 0600); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(dir, "profiles/base.yaml"), []byte("id: base\nbackend: shared\nmodel: local\ntimeout_seconds: 30\n"), 0600); err != nil {
t.Fatal(err)
}
loaded, err := config.Load(ck.File(main), "")
if err != nil {
t.Fatal(err)
}
transport := &cancelTransport{entered: make(chan struct{}, 3)}
runtime, err := llm.NewRuntime(context.Background(), loaded.Config, main, &http.Client{Transport: transport})
if err != nil {
t.Fatal(err)
}
collector := llm.NewObservationCollector()
hooks := collector.Hooks()
started := make(chan struct{}, 2)
hooks.OnEvent = func(event pk.ExecutionEvent) {
if event.Kind == pk.ExecutionStarted {
started <- struct{}{}
}
}
client, err := runtime.NewRoleClient("validation", hooks)
if err != nil {
t.Fatal(err)
}
origin := errors.New("originating validator failure")
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
validator := contextValidator{validate: func(ctx context.Context, req validators.Request) (validators.Result, error) {
if *req.SectionIndex == 0 {
// Hold the failure until one call is active and both peer executions started.
for i := 0; i < 2; i++ {
select {
case <-started:
case <-ctx.Done():
return validators.Result{}, ctx.Err()
}
}
select {
case <-transport.entered:
case <-ctx.Done():
return validators.Result{}, ctx.Err()
}
return validators.Result{}, origin
}
var out map[string]any
_, err := client.CompleteStructured(ctx, structuredoutput.Request{
PromptID: prompts.PromptIDValidatorEditorialReview, PromptVersion: prompts.VersionV1,
Variables: map[string]string{"TranscriptDescriptionBlock": "", "PayloadJSON": "[]"}, Schema: responseschema.ValidatorDecisionSetKey,
Context: structuredoutput.DiagnosticContext{ModuleKey: req.ModuleKey, ModuleInstance: req.ModuleInstance, SectionIndex: req.SectionIndex},
}, &out)
return validators.Result{}, err
}}
module := fakeModule{key: "grammar", policy: proposals.ReplacementPolicyRequireUnique, validators: []contracts.Validator{validator}, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) {
return []proposals.CorrectionProposal{{TargetSegmentID: 1, OriginalText: "hello", CorrectedText: "Hello", Confidence: 1}}, nil
}}
sections := []chunking.Section{}
for i := 0; i < 3; i++ {
sections = append(sections, chunking.Section{Index: i, Segments: []schema.Segment{{ID: 1, Text: "hello"}}})
}
done := make(chan error, 1)
go func() {
_, err := runModulePipeline(ctx, collectSectionProposalsInput{Module: module, Spec: contracts.ModuleRunSpec{ModuleKey: "grammar", InstanceName: "grammar"}, Policy: module.policy, Config: &loaded.Config, Sections: sections})
done <- err
}()
select {
case err := <-done:
if !errors.Is(err, origin) {
t.Fatalf("lost originating failure: %v", err)
}
case <-time.After(2 * time.Second):
t.Fatal("validator failure did not cancel and join peers")
}
if transport.calls.Load() != 1 {
t.Fatalf("pending call dispatched: %d", transport.calls.Load())
}
snapshot := collector.Snapshot()
if len(snapshot.Calls) != 2 {
t.Fatalf("partial observations missing: %+v", snapshot)
}
for _, call := range snapshot.Calls {
if call.Finished.IsZero() || call.Failure != structuredoutput.FailureRuntime {
t.Fatalf("unfinished/cause lost: %+v", call)
}
}
if got := llm.Summarize(snapshot); got.ProviderAttempts != 1 {
t.Fatalf("attempt count: %+v", got)
}
}

View File

@@ -332,7 +332,7 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput)
var (
firstErr error
errOnce sync.Once
errMu sync.Mutex
vwg sync.WaitGroup
)
@@ -340,10 +340,17 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput)
if err == nil {
return
}
errOnce.Do(func() {
errMu.Lock()
defer errMu.Unlock()
if firstErr == nil {
firstErr = err
cancel()
})
}
}
getErr := func() error {
errMu.Lock()
defer errMu.Unlock()
return firstErr
}
for result := range proposalResults {
@@ -351,13 +358,16 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput)
setErr(result.err)
continue
}
if firstErr != nil {
if getErr() != nil {
continue
}
out.Warnings = append(out.Warnings, result.warnings...)
pending[result.sectionPos] = result
for {
if getErr() != nil {
break
}
sectionResult, ok := pending[nextSectionToProcess]
if !ok {
break
@@ -385,18 +395,19 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput)
go func(sectionPos int, enriched []proposals.EnrichedCorrectionProposal, sectionMetadata contracts.SectionMetadata) {
defer vwg.Done()
validated, err := validateSectionCandidates(runCtx, validateSectionCandidatesInput{
Spec: input.Spec,
Policy: input.Policy,
Glossary: input.Glossary,
Config: input.Config,
WorkingTranscript: transcriptFromSection(sectionResult.section),
ModuleInstanceForStages: fmt.Sprintf("%s:section-%04d", input.Spec.InstanceName, sectionMetadata.Index),
Validators: validatorsOrdered,
SectionEnriched: enriched,
ValidationLLMClient: input.ValidationClient,
DiagnosticsDir: input.ValidationDiagnosticsDir,
Timings: input.Timings,
Spec: input.Spec,
Policy: input.Policy,
Glossary: input.Glossary,
Config: input.Config,
WorkingTranscript: transcriptFromSection(sectionResult.section),
SectionIndex: &sectionMetadata.Index,
Validators: validatorsOrdered,
SectionEnriched: enriched,
ValidationLLMClient: input.ValidationClient,
DiagnosticsDir: input.ValidationDiagnosticsDir,
Timings: input.Timings,
})
setErr(err)
validationResults <- sectionValidationResult{
sectionPos: sectionPos,
enriched: enriched,
@@ -477,25 +488,25 @@ func runModulePipeline(ctx context.Context, input collectSectionProposalsInput)
return out.Warnings[i].Message < out.Warnings[j].Message
})
if firstErr != nil {
return out, firstErr
if err := getErr(); err != nil {
return out, err
}
return out, nil
}
type validateSectionCandidatesInput struct {
Spec contracts.ModuleRunSpec
Policy proposals.ReplacementPolicy
Glossary *schema.Glossary
Config *config.Config
WorkingTranscript *schema.Transcript
ModuleInstanceForStages string
Validators []contracts.Validator
SectionEnriched []proposals.EnrichedCorrectionProposal
ValidationLLMClient contracts.StructuredLLMClient
DiagnosticsDir string
Timings *validatorTimingCollector
Spec contracts.ModuleRunSpec
Policy proposals.ReplacementPolicy
Glossary *schema.Glossary
Config *config.Config
WorkingTranscript *schema.Transcript
SectionIndex *int
Validators []contracts.Validator
SectionEnriched []proposals.EnrichedCorrectionProposal
ValidationLLMClient contracts.StructuredLLMClient
DiagnosticsDir string
Timings *validatorTimingCollector
}
type validateSectionCandidatesResult struct {
@@ -527,7 +538,8 @@ func validateSectionCandidates(ctx context.Context, input validateSectionCandida
WorkingTranscript: input.WorkingTranscript,
CandidateProposal: eligible,
ModuleKey: input.Spec.ModuleKey,
ModuleInstance: input.ModuleInstanceForStages,
ModuleInstance: input.Spec.InstanceName,
SectionIndex: input.SectionIndex,
ReplacementPolicy: input.Policy,
Glossary: input.Glossary,
Config: input.Config,

View File

@@ -438,8 +438,8 @@ func (c *captureContractStructuredClient) CompleteStructured(ctx context.Context
func (c *stageAwareStructuredClient) CompleteStructured(ctx context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) {
section := 0
if req.Context.BatchIndex != nil {
section = *req.Context.BatchIndex
if req.Context.SectionIndex != nil {
section = *req.Context.SectionIndex
}
if c.startedSection != nil {
select {

View File

@@ -107,11 +107,15 @@ func (v *LLMBackedValidator) Validate(ctx context.Context, req Request) (Result,
var completion StructuredCompletionResponse
responseSchema := responseschema.MustLookup(responseschema.ValidatorDecisionSetKey)
promptMetadata := validatorPromptMetadata(v.validatorType)
stage := stagename.ValidatorBatch(req.ModuleInstance, v.name, batch.BatchIndex)
stageInstance := req.ModuleInstance
if req.SectionIndex != nil {
stageInstance = fmt.Sprintf("%s:section-%04d", stageInstance, *req.SectionIndex)
}
stage := stagename.ValidatorBatch(stageInstance, v.name, batch.BatchIndex)
completion, err = req.LLMClient.CompleteStructured(ctx, StructuredCompletionRequest{
PromptID: promptMetadata.PromptID, PromptVersion: promptMetadata.PromptVersion,
Variables: variables, Schema: responseschema.ValidatorDecisionSetKey,
Context: structuredoutput.DiagnosticContext{Role: "validation", ModuleKey: req.ModuleKey, ModuleInstance: req.ModuleInstance, BatchIndex: &batch.BatchIndex, ValidatorName: v.name},
Context: structuredoutput.DiagnosticContext{Role: "validation", ModuleKey: req.ModuleKey, ModuleInstance: req.ModuleInstance, SectionIndex: req.SectionIndex, BatchIndex: &batch.BatchIndex, ValidatorName: v.name},
}, &response)
artifacts := InteractionArtifacts{}

View File

@@ -29,6 +29,7 @@ const (
// Request is the runtime input shared by deterministic validators.
type Request struct {
SectionIndex *int `json:"section_index,omitempty"`
WorkingTranscript *schema.Transcript `json:"-"`
CandidateProposal []proposals.EnrichedCorrectionProposal `json:"candidate_proposals"`
ModuleKey string `json:"module_key"`