From 426864eedb1f367dbb63a3cf71cb8c736f9eb06d Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 11 May 2026 20:09:34 -0500 Subject: [PATCH] Add Phase 9 LLM scheduler, config resolution, diagnostics primitives --- docs/architecture.md | 14 +- docs/rewrite-notes.md | 28 ++-- internal/framework/llm/diagnostics.go | 126 ++++++++++++++++++ internal/framework/llm/diagnostics_test.go | 82 ++++++++++++ internal/framework/llm/effective_config.go | 56 ++++++++ .../framework/llm/effective_config_test.go | 103 ++++++++++++++ internal/framework/llm/scheduler.go | 48 +++++++ internal/framework/llm/scheduler_test.go | 113 ++++++++++++++++ 8 files changed, 555 insertions(+), 15 deletions(-) create mode 100644 internal/framework/llm/diagnostics.go create mode 100644 internal/framework/llm/diagnostics_test.go create mode 100644 internal/framework/llm/effective_config.go create mode 100644 internal/framework/llm/effective_config_test.go create mode 100644 internal/framework/llm/scheduler.go create mode 100644 internal/framework/llm/scheduler_test.go diff --git a/docs/architecture.md b/docs/architecture.md index da03cf4..11a6bfb 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -23,6 +23,9 @@ Implemented today: - Deterministic validator-chain execution in the runner with cardinality enforcement. - Module-level validator decision/rejection reporting. - Internal structured LLM client contract plus an `instructor-go`-backed adapter package. +- Bounded LLM scheduler/semaphore infrastructure with context-aware permit handling. +- Runtime primary/validation LLM effective-config resolution helpers with validation inheritance. +- Generic JSON prompt/response diagnostics writer primitives with secret redaction. Not implemented in CLI runtime path today: - Real module execution pipeline (`glossary`, `homophones`, `spoken_word`, `grammar`). @@ -32,7 +35,8 @@ Not implemented in CLI runtime path today: - End-to-end transcript polishing with real module behavior. Phase sequencing note: -- structured LLM client infrastructure is implemented, but scheduler and runtime wiring remain Phase 9 follow-up work; +- structured LLM client, scheduler, config-resolution helpers, and diagnostics primitives are implemented; +- runtime wiring from modules/validators/runner into this LLM infrastructure remains Phase 9 follow-up work; - LLM-backed validators remain Phase 10 work. ## Actual Go package layout @@ -92,6 +96,9 @@ internal/framework/validators/ internal/framework/llm/ instructor_client.go + scheduler.go + effective_config.go + diagnostics.go ``` ## Current CLI behavior @@ -189,6 +196,11 @@ Current runtime boundary: - the CLI/runner runtime path does not instantiate this adapter yet; - no production LLM requests are performed by `audita process`. +`internal/framework/llm` also provides: +- a bounded `Scheduler` for controlled concurrent LLM calls with reliable permit release; +- primary/validation effective-config resolution helpers, including validation inheritance fallback to primary settings; +- generic interaction diagnostics primitives that write machine-readable JSON artifacts for request metadata, request payload, response payload, and optional error payload with secret redaction. + ## Implemented normalization behavior Normalization (`internal/core/normalization`) currently: - sorts by segment start time; diff --git a/docs/rewrite-notes.md b/docs/rewrite-notes.md index 5ff8aad..f364250 100644 --- a/docs/rewrite-notes.md +++ b/docs/rewrite-notes.md @@ -58,13 +58,18 @@ Implemented: - optional API key support for local-compatible endpoints - API-key redaction in returned errors - typed structured decode into caller-provided outputs. +- LLM scheduler/semaphore infrastructure for bounded concurrency with context-aware acquisition and reliable release. +- LLM effective-config resolution helpers: + - primary config resolution + - validation config inheritance from primary when validation fields are unset + - validation override behavior when validation fields are set. +- Generic JSON diagnostics primitives for LLM interactions (request metadata, request payload, response payload, optional error payload) with secret redaction. Not yet implemented in runtime pipeline: - Real correction modules. -- Structured LLM scheduler/concurrency orchestration. - Runtime wiring from production runner/modules into the structured LLM adapter. - LLM-backed validators. -- Prompt/response diagnostics for LLM calls. +- Module/validator call-site wiring to emit LLM prompt/response diagnostics artifacts. - End-to-end transcript polishing behavior. ## Completed phases @@ -196,14 +201,14 @@ Implemented: Not implemented in Phase 8 (by design): - LLM-backed validators (Phase 10). -- Structured LLM scheduler behavior or runtime wiring (Phase 9 follow-up). +- Structured LLM runtime wiring (Phase 9 follow-up). - Real correction modules. -- Prompt/response diagnostics. +- Prompt/response diagnostics runtime wiring. - End-to-end transcript polishing. ## Remaining work plan -Next recommended phase: **Phase 9 follow-up (scheduler + runtime LLM wiring, still no real modules)**. +Next recommended phase: **Phase 9 follow-up (runtime LLM wiring, still no real modules)**. ## Phase 9: Structured LLM client and scheduler infrastructure @@ -218,10 +223,9 @@ Implemented in this phase so far: - Added adapter unit tests for model/base URL handling, retries, context cancellation, optional API key behavior, and error redaction. Still pending in Phase 9: -- Scheduler/semaphore behavior for bounded concurrency. - Runtime wiring in runner/module infrastructure (without introducing real modules yet). -- Prompt/response diagnostics writer primitives for LLM call artifacts. -- Full primary vs validation LLM config-resolution plumbing into runtime LLM call sites. +- Wiring prompt/response diagnostics primitives into future module/validator call sites. +- Wiring effective primary/validation LLM config resolution into runtime LLM call sites. ### Purpose @@ -230,11 +234,8 @@ Implement the provider-neutral LLM infrastructure needed by both proposal genera ### Scope Implement (remaining): -- Primary LLM config resolution. -- Validation LLM config resolution and inheritance from primary settings. -- Redaction of credentials in all diagnostics and reports. -- LLM scheduler/semaphore for bounded backend concurrency. -- Prompt/response diagnostics writer primitives that can later be used by modules and validators. +- Runtime wiring for primary/validation effective LLM config resolution. +- Runtime usage of diagnostics primitives with credential redaction. Do not implement: - Real correction modules. @@ -249,7 +250,6 @@ The codebase has a tested OpenAI-compatible structured-output client adapter, bu ### Definition of done Remaining checklist to close Phase 9: -- Scheduler enforces configured concurrency. - Primary and validation LLM settings resolve correctly in runtime wiring. - Prompt/response diagnostic primitives exist. - API keys are not leaked. diff --git a/internal/framework/llm/diagnostics.go b/internal/framework/llm/diagnostics.go new file mode 100644 index 0000000..7748e17 --- /dev/null +++ b/internal/framework/llm/diagnostics.go @@ -0,0 +1,126 @@ +package llm + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "regexp" + "strings" +) + +const redactedSecret = "[REDACTED]" + +var stageSanitizer = regexp.MustCompile(`[^a-zA-Z0-9_.-]+`) + +// InteractionArtifacts contains file paths for written prompt/response +// diagnostics. +type InteractionArtifacts struct { + RequestMetadataPath string + RequestPayloadPath string + ResponsePayloadPath string + ErrorPayloadPath string +} + +// DiagnosticsWriter writes generic machine-readable LLM interaction artifacts. +type DiagnosticsWriter struct { + dir string + secrets []string +} + +// 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, + } +} + +// WriteInteraction writes request/response/error artifacts for a generic LLM +// interaction stage. +func (w *DiagnosticsWriter) WriteInteraction(stage string, requestMetadata any, requestPayload any, responsePayload any, errorPayload any) (InteractionArtifacts, error) { + if strings.TrimSpace(w.dir) == "" { + return InteractionArtifacts{}, fmt.Errorf("diagnostics directory must not be empty") + } + if err := os.MkdirAll(w.dir, 0o755); err != nil { + return InteractionArtifacts{}, fmt.Errorf("create diagnostics directory: %w", err) + } + + base := sanitizeStageName(stage) + paths := InteractionArtifacts{ + RequestMetadataPath: filepath.Join(w.dir, base+"-request-metadata.json"), + RequestPayloadPath: filepath.Join(w.dir, base+"-request-payload.json"), + ResponsePayloadPath: filepath.Join(w.dir, base+"-response-payload.json"), + } + + if err := w.writeJSON(paths.RequestMetadataPath, requestMetadata); err != nil { + return InteractionArtifacts{}, err + } + if err := w.writeJSON(paths.RequestPayloadPath, requestPayload); err != nil { + return InteractionArtifacts{}, err + } + if err := w.writeJSON(paths.ResponsePayloadPath, responsePayload); err != nil { + return InteractionArtifacts{}, err + } + if errorPayload != nil { + paths.ErrorPayloadPath = filepath.Join(w.dir, base+"-error-payload.json") + if err := w.writeJSON(paths.ErrorPayloadPath, errorPayload); err != nil { + return InteractionArtifacts{}, err + } + } + + return paths, nil +} + +func sanitizeStageName(stage string) string { + stage = strings.TrimSpace(stage) + if stage == "" { + return "llm" + } + safe := stageSanitizer.ReplaceAllString(stage, "_") + safe = strings.Trim(safe, "._-") + if safe == "" { + return "llm" + } + return safe +} + +func (w *DiagnosticsWriter) writeJSON(path string, payload any) error { + raw, err := json.MarshalIndent(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) +} diff --git a/internal/framework/llm/diagnostics_test.go b/internal/framework/llm/diagnostics_test.go new file mode 100644 index 0000000..8e99674 --- /dev/null +++ b/internal/framework/llm/diagnostics_test.go @@ -0,0 +1,82 @@ +package llm + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" +) + +func TestDiagnosticsWriterProducesValidJSONArtifacts(t *testing.T) { + dir := t.TempDir() + writer := NewDiagnosticsWriter(dir, nil) + + artifacts, err := writer.WriteInteraction( + "validation:batch-1", + map[string]any{"stage": "validation", "attempt": 1}, + map[string]any{"messages": []map[string]any{{"role": "user", "content": "hello"}}}, + map[string]any{"result": "ok"}, + nil, + ) + if err != nil { + t.Fatalf("WriteInteraction: %v", err) + } + if artifacts.ErrorPayloadPath != "" { + t.Fatalf("expected no error payload path, got %q", artifacts.ErrorPayloadPath) + } + + assertValidJSONFile(t, artifacts.RequestMetadataPath) + assertValidJSONFile(t, artifacts.RequestPayloadPath) + assertValidJSONFile(t, artifacts.ResponsePayloadPath) +} + +func TestDiagnosticsWriterRedactsSecrets(t *testing.T) { + dir := t.TempDir() + secret := "top-secret-token" + writer := NewDiagnosticsWriter(dir, []string{secret}) + + artifacts, err := writer.WriteInteraction( + "module glossary / request", + map[string]any{"authorization": "Bearer " + secret}, + map[string]any{"api_key": secret, "nested": map[string]any{"value": secret}}, + map[string]any{"echo": secret}, + map[string]any{"error": "failed with " + secret}, + ) + if err != nil { + t.Fatalf("WriteInteraction: %v", err) + } + + for _, path := range []string{ + artifacts.RequestMetadataPath, + artifacts.RequestPayloadPath, + artifacts.ResponsePayloadPath, + artifacts.ErrorPayloadPath, + } { + raw, readErr := os.ReadFile(path) + if readErr != nil { + t.Fatalf("read %s: %v", path, readErr) + } + content := string(raw) + if strings.Contains(content, secret) { + t.Fatalf("secret leaked in %s: %s", filepath.Base(path), content) + } + if !strings.Contains(content, redactedSecret) { + t.Fatalf("redaction marker missing in %s: %s", filepath.Base(path), content) + } + if !json.Valid(raw) { + t.Fatalf("invalid json in %s", filepath.Base(path)) + } + } +} + +func assertValidJSONFile(t *testing.T, path string) { + t.Helper() + raw, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read %s: %v", path, err) + } + if !json.Valid(raw) { + t.Fatalf("invalid json in %s", filepath.Base(path)) + } +} diff --git a/internal/framework/llm/effective_config.go b/internal/framework/llm/effective_config.go new file mode 100644 index 0000000..6517f00 --- /dev/null +++ b/internal/framework/llm/effective_config.go @@ -0,0 +1,56 @@ +package llm + +import ( + "net/http" + "strings" + "time" + + "gitea.maximumdirect.net/eric/audita/internal/core/config" +) + +// EffectiveConfig is the runtime LLM configuration consumed by adapters and +// schedulers after primary/validation inheritance is resolved. +type EffectiveConfig struct { + APIKey string + Model string + BaseURL string + RequestTimeout time.Duration + MaxRetries int + Concurrency int +} + +// ResolvePrimaryConfig resolves the primary LLM settings into runtime form. +func ResolvePrimaryConfig(cfg config.Config) EffectiveConfig { + return resolveFromLLMConfig(cfg.PrimaryLLM) +} + +// ResolveValidationConfig resolves validation LLM settings with inheritance +// from primary values when validation overrides are unset. +func ResolveValidationConfig(cfg config.Config) EffectiveConfig { + return resolveFromLLMConfig(cfg.EffectiveValidationLLMConfig()) +} + +// ToInstructorClientConfig converts an effective runtime config into adapter +// config while keeping instructor-go types fully internal to this package. +func (c EffectiveConfig) ToInstructorClientConfig(mode Mode, httpClient *http.Client) InstructorClientConfig { + return InstructorClientConfig{ + BaseURL: c.BaseURL, + Model: c.Model, + APIKey: c.APIKey, + MaxRetries: c.MaxRetries, + Mode: mode, + HTTPClient: httpClient, + RequestTimeout: c.RequestTimeout, + } +} + +func resolveFromLLMConfig(raw config.LLMConfig) EffectiveConfig { + return EffectiveConfig{ + APIKey: strings.TrimSpace(raw.APIKey), + Model: strings.TrimSpace(raw.Model), + BaseURL: strings.TrimSpace(raw.BaseURL), + RequestTimeout: time.Duration(raw.TimeoutSeconds) * time.Second, + MaxRetries: raw.MaxRetries, + Concurrency: raw.Concurrency, + } +} diff --git a/internal/framework/llm/effective_config_test.go b/internal/framework/llm/effective_config_test.go new file mode 100644 index 0000000..f39c9c3 --- /dev/null +++ b/internal/framework/llm/effective_config_test.go @@ -0,0 +1,103 @@ +package llm + +import ( + "testing" + "time" + + "gitea.maximumdirect.net/eric/audita/internal/core/config" +) + +func TestResolvePrimaryConfig(t *testing.T) { + cfg := config.Default() + cfg.PrimaryLLM.APIKey = " primary-key " + cfg.PrimaryLLM.Model = " model-x " + cfg.PrimaryLLM.BaseURL = " https://example.test/v1 " + cfg.PrimaryLLM.TimeoutSeconds = 42 + cfg.PrimaryLLM.MaxRetries = 7 + cfg.PrimaryLLM.Concurrency = 3 + + effective := ResolvePrimaryConfig(cfg) + if effective.APIKey != "primary-key" { + t.Fatalf("unexpected api key: %q", effective.APIKey) + } + if effective.Model != "model-x" { + t.Fatalf("unexpected model: %q", effective.Model) + } + if effective.BaseURL != "https://example.test/v1" { + t.Fatalf("unexpected base url: %q", effective.BaseURL) + } + if effective.RequestTimeout != 42*time.Second { + t.Fatalf("unexpected timeout: %s", effective.RequestTimeout) + } + if effective.MaxRetries != 7 { + t.Fatalf("unexpected max retries: %d", effective.MaxRetries) + } + if effective.Concurrency != 3 { + t.Fatalf("unexpected concurrency: %d", effective.Concurrency) + } +} + +func TestResolveValidationConfigInheritsPrimaryWhenUnset(t *testing.T) { + cfg := config.Default() + cfg.PrimaryLLM.APIKey = "primary-key" + cfg.PrimaryLLM.Model = "primary-model" + cfg.PrimaryLLM.BaseURL = "https://primary.example/v1" + cfg.PrimaryLLM.TimeoutSeconds = 90 + cfg.PrimaryLLM.MaxRetries = 4 + cfg.PrimaryLLM.Concurrency = 2 + + effective := ResolveValidationConfig(cfg) + if effective.APIKey != "primary-key" || + effective.Model != "primary-model" || + effective.BaseURL != "https://primary.example/v1" || + effective.RequestTimeout != 90*time.Second || + effective.MaxRetries != 4 || + effective.Concurrency != 2 { + t.Fatalf("unexpected inherited config: %+v", effective) + } +} + +func TestResolveValidationConfigOverridesPrimaryWhenSet(t *testing.T) { + cfg := config.Default() + cfg.PrimaryLLM.APIKey = "primary-key" + cfg.PrimaryLLM.Model = "primary-model" + cfg.PrimaryLLM.BaseURL = "https://primary.example/v1" + cfg.PrimaryLLM.TimeoutSeconds = 90 + cfg.PrimaryLLM.MaxRetries = 4 + cfg.PrimaryLLM.Concurrency = 2 + + timeout := 12 + retries := 9 + concurrency := 6 + cfg.ValidationLLM.APIKey = "validation-key" + cfg.ValidationLLM.Model = "validation-model" + cfg.ValidationLLM.BaseURL = "https://validation.example/v1" + cfg.ValidationLLM.TimeoutSeconds = &timeout + cfg.ValidationLLM.MaxRetries = &retries + cfg.ValidationLLM.Concurrency = &concurrency + + effective := ResolveValidationConfig(cfg) + if effective.APIKey != "validation-key" || + effective.Model != "validation-model" || + effective.BaseURL != "https://validation.example/v1" || + effective.RequestTimeout != 12*time.Second || + effective.MaxRetries != 9 || + effective.Concurrency != 6 { + t.Fatalf("unexpected override config: %+v", effective) + } +} + +func TestResolveConfigOptionalAPIKey(t *testing.T) { + cfg := config.Default() + cfg.PrimaryLLM.APIKey = "" + cfg.ValidationLLM.APIKey = "" + + primary := ResolvePrimaryConfig(cfg) + validation := ResolveValidationConfig(cfg) + if primary.APIKey != "" { + t.Fatalf("expected empty primary api key, got %q", primary.APIKey) + } + if validation.APIKey != "" { + t.Fatalf("expected empty validation api key, got %q", validation.APIKey) + } +} diff --git a/internal/framework/llm/scheduler.go b/internal/framework/llm/scheduler.go new file mode 100644 index 0000000..78f80bc --- /dev/null +++ b/internal/framework/llm/scheduler.go @@ -0,0 +1,48 @@ +package llm + +import ( + "context" + "fmt" + "sync" +) + +// Scheduler bounds concurrent backend LLM calls. +type Scheduler struct { + permits chan struct{} +} + +// NewScheduler creates a scheduler with a fixed concurrency limit. +func NewScheduler(maxConcurrency int) (*Scheduler, error) { + if maxConcurrency <= 0 { + return nil, fmt.Errorf("max concurrency must be greater than zero") + } + return &Scheduler{ + permits: make(chan struct{}, maxConcurrency), + }, nil +} + +// Acquire blocks until a permit is available or the context is canceled. +// The returned release function is safe to call multiple times. +func (s *Scheduler) Acquire(ctx context.Context) (func(), error) { + select { + case s.permits <- struct{}{}: + var once sync.Once + return func() { + once.Do(func() { + <-s.permits + }) + }, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +// Run acquires a permit, executes fn, and releases the permit. +func (s *Scheduler) Run(ctx context.Context, fn func(context.Context) error) error { + release, err := s.Acquire(ctx) + if err != nil { + return err + } + defer release() + return fn(ctx) +} diff --git a/internal/framework/llm/scheduler_test.go b/internal/framework/llm/scheduler_test.go new file mode 100644 index 0000000..1981029 --- /dev/null +++ b/internal/framework/llm/scheduler_test.go @@ -0,0 +1,113 @@ +package llm + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestSchedulerEnforcesMaxConcurrency(t *testing.T) { + s, err := NewScheduler(2) + if err != nil { + t.Fatalf("NewScheduler: %v", err) + } + + var inFlight int32 + var maxInFlight int32 + var wg sync.WaitGroup + + for i := 0; i < 12; i++ { + wg.Add(1) + go func() { + defer wg.Done() + err := s.Run(context.Background(), func(ctx context.Context) error { + _ = ctx + current := atomic.AddInt32(&inFlight, 1) + for { + seen := atomic.LoadInt32(&maxInFlight) + if current <= seen || atomic.CompareAndSwapInt32(&maxInFlight, seen, current) { + break + } + } + time.Sleep(20 * time.Millisecond) + atomic.AddInt32(&inFlight, -1) + return nil + }) + if err != nil { + t.Errorf("Run error: %v", err) + } + }() + } + + wg.Wait() + if got := atomic.LoadInt32(&maxInFlight); got > 2 { + t.Fatalf("expected max in-flight <= 2, got %d", got) + } +} + +func TestSchedulerReleasesPermitOnSuccess(t *testing.T) { + s, err := NewScheduler(1) + if err != nil { + t.Fatalf("NewScheduler: %v", err) + } + + for i := 0; i < 3; i++ { + if err := s.Run(context.Background(), func(ctx context.Context) error { + _ = ctx + return nil + }); err != nil { + t.Fatalf("Run[%d] error: %v", i, err) + } + } +} + +func TestSchedulerReleasesPermitOnError(t *testing.T) { + s, err := NewScheduler(1) + if err != nil { + t.Fatalf("NewScheduler: %v", err) + } + + expectedErr := errors.New("boom") + err = s.Run(context.Background(), func(ctx context.Context) error { + _ = ctx + return expectedErr + }) + if !errors.Is(err, expectedErr) { + t.Fatalf("expected %v, got %v", expectedErr, err) + } + + // Must be able to run again after an error, proving permit release. + if err := s.Run(context.Background(), func(ctx context.Context) error { + _ = ctx + return nil + }); err != nil { + t.Fatalf("second Run error: %v", err) + } +} + +func TestSchedulerRespectsContextCancellation(t *testing.T) { + s, err := NewScheduler(1) + if err != nil { + t.Fatalf("NewScheduler: %v", err) + } + + release, err := s.Acquire(context.Background()) + if err != nil { + t.Fatalf("Acquire: %v", err) + } + defer release() + + ctx, cancel := context.WithTimeout(context.Background(), 25*time.Millisecond) + defer cancel() + + _, err = s.Acquire(ctx) + if err == nil { + t.Fatalf("expected context cancellation error") + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("expected deadline exceeded, got %v", err) + } +}