Add Phase 9 LLM scheduler, config resolution, diagnostics primitives

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)
+}

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))
+	}
+}

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,
+	}
+}

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)
+	}
+}

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)
+}

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)
+	}
+}