Wire production CLI catalog and scheduled LLM client

internal/cli/catalog.go

@@ -0,0 +1,172 @@
+package cli
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
+	"gitea.maximumdirect.net/eric/notarius/internal/core/config"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/llm"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+	"gitea.maximumdirect.net/eric/notarius/internal/modules/chunk/generic"
+	"gitea.maximumdirect.net/eric/notarius/internal/modules/extract/dnd/spells"
+	"gitea.maximumdirect.net/eric/notarius/internal/modules/input/seriatim"
+	"gitea.maximumdirect.net/eric/notarius/internal/modules/merge/appendorder"
+	"gitea.maximumdirect.net/eric/notarius/internal/modules/normalize/noop"
+	jsonoutput "gitea.maximumdirect.net/eric/notarius/internal/modules/output/json"
+)
+
+func productionRegistries() (pipeline.Registries, error) {
+	registries := pipeline.Registries{
+		Inputs:      pipeline.NewInputAdapterRegistry(),
+		Chunkers:    pipeline.NewChunkerRegistry(),
+		Extractors:  pipeline.NewExtractorRegistry(),
+		Mergers:     pipeline.NewMergerRegistry(),
+		Normalizers: pipeline.NewNormalizerRegistry(),
+		Validators:  pipeline.NewValidatorRegistry(),
+		Outputs:     pipeline.NewOutputEncoderRegistry(),
+	}
+	if err := seriatim.Register(registries.Inputs); err != nil {
+		return pipeline.Registries{}, fmt.Errorf("register seriatim input: %w", err)
+	}
+	if err := generic.Register(registries.Chunkers); err != nil {
+		return pipeline.Registries{}, fmt.Errorf("register generic chunker: %w", err)
+	}
+	if err := spells.Register(registries.Extractors); err != nil {
+		return pipeline.Registries{}, fmt.Errorf("register dnd spells extractor: %w", err)
+	}
+	if err := appendorder.Register(registries.Mergers); err != nil {
+		return pipeline.Registries{}, fmt.Errorf("register appendorder merger: %w", err)
+	}
+	if err := noop.Register(registries.Normalizers); err != nil {
+		return pipeline.Registries{}, fmt.Errorf("register noop normalizer: %w", err)
+	}
+	if err := jsonoutput.Register(registries.Outputs); err != nil {
+		return pipeline.Registries{}, fmt.Errorf("register json output encoder: %w", err)
+	}
+	return registries, nil
+}
+
+func productionCatalog() (pipeline.ModuleCatalog, error) {
+	registries, err := productionRegistries()
+	if err != nil {
+		return pipeline.ModuleCatalog{}, err
+	}
+	return catalogFromRegistries(registries), nil
+}
+
+func effectiveCatalog(opts Options) (pipeline.ModuleCatalog, error) {
+	if !isEmptyCatalog(opts.Catalog) {
+		return opts.Catalog, nil
+	}
+	if !isEmptyRegistries(opts.Registries) {
+		return catalogFromRegistries(opts.Registries), nil
+	}
+	return productionCatalog()
+}
+
+func effectiveRegistries(opts Options) (pipeline.Registries, error) {
+	if !isEmptyRegistries(opts.Registries) {
+		return opts.Registries, nil
+	}
+	if !isEmptyCatalog(opts.Catalog) {
+		return registriesFromCatalog(opts.Catalog), nil
+	}
+	return productionRegistries()
+}
+
+func catalogFromRegistries(registries pipeline.Registries) pipeline.ModuleCatalog {
+	return pipeline.ModuleCatalog{
+		Inputs:      registries.Inputs,
+		Chunkers:    registries.Chunkers,
+		Extractors:  registries.Extractors,
+		Mergers:     registries.Mergers,
+		Normalizers: registries.Normalizers,
+		Validators:  registries.Validators,
+		Outputs:     registries.Outputs,
+	}
+}
+
+func registriesFromCatalog(catalog pipeline.ModuleCatalog) pipeline.Registries {
+	return pipeline.Registries{
+		Inputs:      catalog.Inputs,
+		Chunkers:    catalog.Chunkers,
+		Extractors:  catalog.Extractors,
+		Mergers:     catalog.Mergers,
+		Normalizers: catalog.Normalizers,
+		Validators:  catalog.Validators,
+		Outputs:     catalog.Outputs,
+	}
+}
+
+func isEmptyCatalog(catalog pipeline.ModuleCatalog) bool {
+	return catalog.Inputs == nil &&
+		catalog.Chunkers == nil &&
+		catalog.Extractors == nil &&
+		catalog.Mergers == nil &&
+		catalog.Normalizers == nil &&
+		catalog.Validators == nil &&
+		catalog.Outputs == nil
+}
+
+func isEmptyRegistries(registries pipeline.Registries) bool {
+	return registries.Inputs == nil &&
+		registries.Chunkers == nil &&
+		registries.Extractors == nil &&
+		registries.Mergers == nil &&
+		registries.Normalizers == nil &&
+		registries.Validators == nil &&
+		registries.Outputs == nil
+}
+
+func productionLLMClientFactory(ctx context.Context, cfg config.Config, profileID string) (contracts.StructuredLLMClient, []artifacts.LLMProfileManifest, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, nil, err
+	}
+	trimmedID := strings.TrimSpace(profileID)
+	if trimmedID == "" {
+		trimmedID = pipeline.DefaultLLMProfile
+	}
+
+	profile, ok := cfg.LLMProfile(trimmedID)
+	if !ok {
+		return nil, nil, fmt.Errorf("LLM profile %q is not configured", trimmedID)
+	}
+	clientCfg, err := cfg.OpenAICompatibleClientConfig(trimmedID)
+	if err != nil {
+		return nil, nil, err
+	}
+	client, err := llm.NewOpenAICompatibleClient(clientCfg)
+	if err != nil {
+		return nil, nil, fmt.Errorf("create LLM client for profile %q: %w", trimmedID, err)
+	}
+
+	scheduler, err := llm.NewScheduler(effectiveLLMConcurrency(cfg, profile))
+	if err != nil {
+		return nil, nil, fmt.Errorf("create LLM scheduler for profile %q: %w", trimmedID, err)
+	}
+	provider := strings.TrimSpace(profile.Provider)
+	if provider == "" {
+		provider = "openai-compatible"
+	}
+	metadata := []artifacts.LLMProfileManifest{
+		{
+			ID:       trimmedID,
+			Provider: provider,
+			Model:    strings.TrimSpace(profile.Model),
+		},
+	}
+	return llm.NewScheduledClient(client, scheduler), metadata, nil
+}
+
+func effectiveLLMConcurrency(cfg config.Config, profile config.LLMProfile) int {
+	if profile.MaxConcurrency > 0 {
+		return profile.MaxConcurrency
+	}
+	if cfg.Concurrency.TotalLLM > 0 {
+		return cfg.Concurrency.TotalLLM
+	}
+	return 1
+}

internal/cli/run.go

@@ -1,6 +1,7 @@
 package cli
 
 import (
+	"context"
 	"encoding/json"
 	"flag"
 	"fmt"
@@ -8,8 +9,11 @@ import (
 	"os"
 	"sort"
 	"strings"
+	"time"
 
+	"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
 	"gitea.maximumdirect.net/eric/notarius/internal/core/config"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
 	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
 )
 
@@ -23,9 +27,14 @@ const usage = `Usage:
 
 type Options struct {
 	Catalog          pipeline.ModuleCatalog
+	Registries       pipeline.Registries
+	LLMClientFactory LLMClientFactory
 	LookupEnv        func(string) (string, bool)
+	Now              func() time.Time
 }
 
+type LLMClientFactory func(ctx context.Context, cfg config.Config, profileID string) (contracts.StructuredLLMClient, []artifacts.LLMProfileManifest, error)
+
 // Run executes the command-line interface and returns a process exit code.
 func Run(args []string, stdout, stderr io.Writer) int {
 	return RunWithOptions(args, stdout, stderr, Options{})
@@ -61,6 +70,12 @@ func normalizeOptions(opts Options) Options {
 	if opts.LookupEnv == nil {
 		opts.LookupEnv = os.LookupEnv
 	}
+	if opts.Now == nil {
+		opts.Now = time.Now
+	}
+	if opts.LLMClientFactory == nil {
+		opts.LLMClientFactory = productionLLMClientFactory
+	}
 	return opts
 }
 
@@ -111,10 +126,15 @@ func runConfigValidate(args []string, stdout, stderr io.Writer, opts Options) in
 	}
 
 	if strings.TrimSpace(*pipelineID) != "" {
+		catalog, err := effectiveCatalog(opts)
+		if err != nil {
+			fmt.Fprintf(stderr, "notarius: %v\n", err)
+			return 1
+		}
 		if _, err := cfg.Resolve(config.ResolveInput{
 			PipelineID: *pipelineID,
 			Only:       only,
-			Catalog:    opts.Catalog,
+			Catalog:    catalog,
 		}); err != nil {
 			fmt.Fprintf(stderr, "notarius: %v\n", err)
 			return 1

internal/cli/run_test.go

@@ -2,13 +2,22 @@ package cli
 
 import (
 	"bytes"
+	"context"
 	"os"
 	"path/filepath"
+	"reflect"
 	"strings"
 	"testing"
 
+	"gitea.maximumdirect.net/eric/notarius/internal/core/config"
 	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
 	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+	"gitea.maximumdirect.net/eric/notarius/internal/modules/chunk/generic"
+	"gitea.maximumdirect.net/eric/notarius/internal/modules/extract/dnd/spells"
+	"gitea.maximumdirect.net/eric/notarius/internal/modules/input/seriatim"
+	"gitea.maximumdirect.net/eric/notarius/internal/modules/merge/appendorder"
+	"gitea.maximumdirect.net/eric/notarius/internal/modules/normalize/noop"
+	jsonoutput "gitea.maximumdirect.net/eric/notarius/internal/modules/output/json"
 )
 
 func TestRunNoArgsWritesUsageToStdout(t *testing.T) {
@@ -102,6 +111,95 @@ func TestRunConfigValidateSuccessWithFakeCatalog(t *testing.T) {
 	}
 }
 
+func TestProductionCatalogIncludesDefaultModules(t *testing.T) {
+	catalog, err := productionCatalog()
+	if err != nil {
+		t.Fatalf("productionCatalog() error = %v, want nil", err)
+	}
+
+	tests := []struct {
+		name string
+		got  func() (pipeline.ModuleSpec, bool)
+		want pipeline.ModuleSpec
+	}{
+		{
+			name: "seriatim input",
+			got:  func() (pipeline.ModuleSpec, bool) { return catalog.Inputs.Spec(seriatim.Key) },
+			want: seriatim.ModuleSpec(),
+		},
+		{
+			name: "generic chunker",
+			got:  func() (pipeline.ModuleSpec, bool) { return catalog.Chunkers.Spec(generic.Key) },
+			want: generic.ModuleSpec(),
+		},
+		{
+			name: "dnd spells extractor",
+			got:  func() (pipeline.ModuleSpec, bool) { return catalog.Extractors.Spec(spells.Key) },
+			want: spells.ModuleSpec(),
+		},
+		{
+			name: "appendorder merger",
+			got:  func() (pipeline.ModuleSpec, bool) { return catalog.Mergers.Spec(appendorder.Key) },
+			want: appendorder.ModuleSpec(),
+		},
+		{
+			name: "noop normalizer",
+			got:  func() (pipeline.ModuleSpec, bool) { return catalog.Normalizers.Spec(noop.Key) },
+			want: noop.ModuleSpec(),
+		},
+		{
+			name: "json output",
+			got:  func() (pipeline.ModuleSpec, bool) { return catalog.Outputs.Spec(jsonoutput.Key) },
+			want: jsonoutput.ModuleSpec(),
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			got, ok := test.got()
+			if !ok {
+				t.Fatalf("module spec ok = false, want true")
+			}
+			if !reflect.DeepEqual(got, test.want) {
+				t.Fatalf("module spec = %#v, want %#v", got, test.want)
+			}
+		})
+	}
+}
+
+func TestRunConfigValidateUsesProductionCatalogByDefault(t *testing.T) {
+	configPath := writeTestConfig(t, mvpConfigYAML("dnd-session", "dnd/spells"))
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+
+	code := RunWithOptions([]string{"config", "validate", "--config", configPath, "--pipeline", "dnd-session"}, &stdout, &stderr, Options{})
+
+	if code != 0 {
+		t.Fatalf("RunWithOptions() code = %d, stderr=%q", code, stderr.String())
+	}
+	if !strings.Contains(stdout.String(), "is valid for pipeline") {
+		t.Fatalf("stdout = %q, want validation success", stdout.String())
+	}
+}
+
+func TestRunConfigValidateUnknownProductionModuleIncludesContext(t *testing.T) {
+	configPath := writeTestConfig(t, mvpConfigYAML("dnd-session", "missing/extract"))
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+
+	code := RunWithOptions([]string{"config", "validate", "--config", configPath, "--pipeline", "dnd-session"}, &stdout, &stderr, Options{})
+
+	if code != 1 {
+		t.Fatalf("RunWithOptions() code = %d, want 1", code)
+	}
+	got := stderr.String()
+	for _, want := range []string{"dnd-session", "extract", "missing/extract", "not registered"} {
+		if !strings.Contains(got, want) {
+			t.Fatalf("stderr = %q, want substring %q", got, want)
+		}
+	}
+}
+
 func TestRunConfigValidateReportsParseErrors(t *testing.T) {
 	configPath := writeFile(t, "config.yml", "version: 2\n")
 	var stdout bytes.Buffer
@@ -320,6 +418,83 @@ func TestRunInvalidFlagsExitTwo(t *testing.T) {
 	}
 }
 
+func TestProductionLLMClientFactoryRejectsMissingProfile(t *testing.T) {
+	cfg := config.Default()
+
+	_, _, err := productionLLMClientFactory(context.Background(), cfg, "missing")
+	if err == nil {
+		t.Fatal("productionLLMClientFactory() error = nil, want error")
+	}
+	if !strings.Contains(err.Error(), "LLM profile") || !strings.Contains(err.Error(), "missing") {
+		t.Fatalf("error = %q, want missing profile context", err.Error())
+	}
+}
+
+func TestProductionLLMClientFactoryRejectsInvalidProfile(t *testing.T) {
+	tests := []struct {
+		name    string
+		profile config.LLMProfile
+		want    string
+	}{
+		{
+			name:    "unsupported provider",
+			profile: config.LLMProfile{Provider: "other", BaseURL: "https://example.test", Model: "model"},
+			want:    "not supported",
+		},
+		{
+			name:    "missing base url",
+			profile: config.LLMProfile{Provider: "openai-compatible", Model: "model"},
+			want:    "base URL",
+		},
+		{
+			name:    "missing model",
+			profile: config.LLMProfile{Provider: "openai-compatible", BaseURL: "https://example.test"},
+			want:    "model",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			cfg := config.Default()
+			cfg.LLMProfiles = map[string]config.LLMProfile{"default": test.profile}
+
+			_, _, err := productionLLMClientFactory(context.Background(), cfg, "default")
+			if err == nil {
+				t.Fatal("productionLLMClientFactory() error = nil, want error")
+			}
+			if !strings.Contains(err.Error(), test.want) {
+				t.Fatalf("error = %q, want substring %q", err.Error(), test.want)
+			}
+		})
+	}
+}
+
+func TestProductionLLMClientFactoryReturnsScheduledClientAndManifestMetadata(t *testing.T) {
+	cfg := config.Default()
+	cfg.LLMProfiles = map[string]config.LLMProfile{
+		"default": {
+			Provider:       "openai-compatible",
+			BaseURL:        "https://example.test",
+			Model:          "model-a",
+			MaxConcurrency: 2,
+		},
+	}
+
+	client, metadata, err := productionLLMClientFactory(context.Background(), cfg, "default")
+	if err != nil {
+		t.Fatalf("productionLLMClientFactory() error = %v, want nil", err)
+	}
+	if client == nil {
+		t.Fatal("client = nil, want scheduled client")
+	}
+	if len(metadata) != 1 {
+		t.Fatalf("len(metadata) = %d, want 1", len(metadata))
+	}
+	if metadata[0].ID != "default" || metadata[0].Provider != "openai-compatible" || metadata[0].Model != "model-a" {
+		t.Fatalf("metadata = %#v, want profile-safe model metadata", metadata)
+	}
+}
+
 func writeTestConfig(t *testing.T, content string) string {
 	t.Helper()
 	return writeFile(t, "config.yml", content)
@@ -354,6 +529,17 @@ func testConfigYAMLForPipelines(pipelines map[string][]string) string {
 	return b.String()
 }
 
+func mvpConfigYAML(pipelineID string, extractor string) string {
+	return `version: 1
+pipelines:
+  ` + pipelineID + `:
+    input: seriatim
+    artifacts:
+      spells:
+        extract: ` + extractor + `
+`
+}
+
 func fakeCatalog(t *testing.T) pipeline.ModuleCatalog {
 	t.Helper()
 	inputs := pipeline.NewInputAdapterRegistry()

internal/framework/llm/scheduled_client.go

@@ -0,0 +1,43 @@
+package llm
+
+import (
+	"context"
+	"fmt"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+)
+
+type scheduledClient struct {
+	client    contracts.StructuredLLMClient
+	scheduler *Scheduler
+}
+
+func NewScheduledClient(client contracts.StructuredLLMClient, scheduler *Scheduler) contracts.StructuredLLMClient {
+	return &scheduledClient{
+		client:    client,
+		scheduler: scheduler,
+	}
+}
+
+func (c *scheduledClient) CompleteStructured(ctx context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) {
+	if c == nil {
+		return contracts.StructuredCompletionResponse{}, fmt.Errorf("scheduled LLM client must not be nil")
+	}
+	if c.client == nil {
+		return contracts.StructuredCompletionResponse{}, fmt.Errorf("scheduled LLM client inner client must not be nil")
+	}
+	if c.scheduler == nil {
+		return contracts.StructuredCompletionResponse{}, fmt.Errorf("scheduled LLM client scheduler must not be nil")
+	}
+
+	var response contracts.StructuredCompletionResponse
+	err := c.scheduler.Run(ctx, func(ctx context.Context) error {
+		var callErr error
+		response, callErr = c.client.CompleteStructured(ctx, req, out)
+		return callErr
+	})
+	if err != nil {
+		return contracts.StructuredCompletionResponse{}, err
+	}
+	return response, nil
+}

internal/framework/llm/scheduled_client_test.go

@@ -0,0 +1,115 @@
+package llm
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+)
+
+func TestScheduledClientEnforcesSchedulerLimit(t *testing.T) {
+	scheduler, err := NewScheduler(1)
+	if err != nil {
+		t.Fatalf("NewScheduler() error = %v, want nil", err)
+	}
+	inner := &blockingStructuredClient{
+		release: make(chan struct{}),
+	}
+	client := NewScheduledClient(inner, scheduler)
+
+	var wg sync.WaitGroup
+	for i := 0; i < 3; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			var out map[string]any
+			if _, err := client.CompleteStructured(context.Background(), contracts.StructuredCompletionRequest{}, &out); err != nil {
+				t.Errorf("CompleteStructured() error = %v, want nil", err)
+			}
+		}()
+	}
+
+	waitForAtomicAtLeast(t, &inner.calls, 1)
+	time.Sleep(20 * time.Millisecond)
+	if got := atomic.LoadInt32(&inner.maxInFlight); got > 1 {
+		t.Fatalf("max in-flight calls = %d, want <= 1", got)
+	}
+	close(inner.release)
+	wg.Wait()
+	if got := atomic.LoadInt32(&inner.calls); got != 3 {
+		t.Fatalf("calls = %d, want 3", got)
+	}
+}
+
+func TestScheduledClientPropagatesClientError(t *testing.T) {
+	scheduler, err := NewScheduler(1)
+	if err != nil {
+		t.Fatalf("NewScheduler() error = %v, want nil", err)
+	}
+	expected := errors.New("provider unavailable")
+	client := NewScheduledClient(&errorStructuredClient{err: expected}, scheduler)
+
+	_, err = client.CompleteStructured(context.Background(), contracts.StructuredCompletionRequest{}, &struct{}{})
+	if !errors.Is(err, expected) {
+		t.Fatalf("CompleteStructured() error = %v, want %v", err, expected)
+	}
+}
+
+func TestScheduledClientPropagatesSchedulerError(t *testing.T) {
+	scheduler, err := NewScheduler(1)
+	if err != nil {
+		t.Fatalf("NewScheduler() error = %v, want nil", err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	client := NewScheduledClient(&errorStructuredClient{}, scheduler)
+
+	_, err = client.CompleteStructured(ctx, contracts.StructuredCompletionRequest{}, &struct{}{})
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("CompleteStructured() error = %v, want context canceled", err)
+	}
+}
+
+type blockingStructuredClient struct {
+	release     chan struct{}
+	inFlight    int32
+	maxInFlight int32
+	calls       int32
+}
+
+func (c *blockingStructuredClient) CompleteStructured(ctx context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) {
+	atomic.AddInt32(&c.calls, 1)
+	current := atomic.AddInt32(&c.inFlight, 1)
+	for {
+		seen := atomic.LoadInt32(&c.maxInFlight)
+		if current <= seen || atomic.CompareAndSwapInt32(&c.maxInFlight, seen, current) {
+			break
+		}
+	}
+	defer atomic.AddInt32(&c.inFlight, -1)
+
+	select {
+	case <-c.release:
+	case <-ctx.Done():
+		return contracts.StructuredCompletionResponse{}, ctx.Err()
+	}
+	if target, ok := out.(*map[string]any); ok {
+		*target = map[string]any{"ok": true}
+	}
+	return contracts.StructuredCompletionResponse{
+		Content: json.RawMessage(`{"ok":true}`),
+	}, nil
+}
+
+type errorStructuredClient struct {
+	err error
+}
+
+func (c *errorStructuredClient) CompleteStructured(ctx context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) {
+	return contracts.StructuredCompletionResponse{}, c.err
+}