diff --git a/internal/cli/catalog.go b/internal/cli/catalog.go new file mode 100644 index 0000000..dd863ee --- /dev/null +++ b/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 +} diff --git a/internal/cli/run.go b/internal/cli/run.go index 20aaa22..23e3a40 100644 --- a/internal/cli/run.go +++ b/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" ) @@ -22,10 +26,15 @@ const usage = `Usage: ` type Options struct { - Catalog pipeline.ModuleCatalog - LookupEnv func(string) (string, bool) + 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 diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index 5e1a8a4..7f70a1a 100644 --- a/internal/cli/run_test.go +++ b/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() diff --git a/internal/framework/llm/scheduled_client.go b/internal/framework/llm/scheduled_client.go new file mode 100644 index 0000000..88d923a --- /dev/null +++ b/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 +} diff --git a/internal/framework/llm/scheduled_client_test.go b/internal/framework/llm/scheduled_client_test.go new file mode 100644 index 0000000..b4b4835 --- /dev/null +++ b/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 +}