diff --git a/internal/cli/run.go b/internal/cli/run.go index 23e3a40..d9705b6 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -21,6 +21,7 @@ const defaultConfigPath = "/usr/local/etc/notarius/config.yml" const usage = `Usage: notarius help + notarius run --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b] notarius config validate --config path/to/config.yml [--pipeline pipeline-id] [--only lane-a,lane-b] notarius pipelines list --config path/to/config.yml [--json] ` @@ -55,6 +56,8 @@ func RunWithOptions(args []string, stdout, stderr io.Writer, opts Options) int { return runConfig(args[1:], stdout, stderr, opts) case "pipelines": return runPipelines(args[1:], stdout, stderr, opts) + case "run": + return runPipelineCommand(args[1:], stdout, stderr, opts) default: fmt.Fprintf(stderr, "notarius: unknown command %q\n", args[0]) writeUsage(stderr) @@ -79,6 +82,185 @@ func normalizeOptions(opts Options) Options { return opts } +func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) int { + fs := flag.NewFlagSet("run", flag.ContinueOnError) + fs.SetOutput(io.Discard) + configPath := fs.String("config", "", "config file path") + inputPath := fs.String("input", "", "source input file path") + onlyRaw := fs.String("only", "", "comma-separated artifact lanes") + outputDir := fs.String("output-dir", "", "output directory") + diagnosticsDir := fs.String("diagnostics-dir", "", "diagnostics directory") + llmProfile := fs.String("llm-profile", "", "LLM profile override") + if err := fs.Parse(reorderRunArgs(args)); err != nil { + fmt.Fprintf(stderr, "notarius: %v\n", err) + return 2 + } + if fs.NArg() == 0 { + fmt.Fprintln(stderr, "notarius: run requires a pipeline ID") + return 2 + } + if fs.NArg() > 1 { + fmt.Fprintf(stderr, "notarius: unexpected argument %q\n", fs.Arg(1)) + return 2 + } + pipelineID := strings.TrimSpace(fs.Arg(0)) + if pipelineID == "" { + fmt.Fprintln(stderr, "notarius: run requires a pipeline ID") + return 2 + } + if strings.TrimSpace(*inputPath) == "" { + fmt.Fprintln(stderr, "notarius: run requires --input") + return 2 + } + only, err := parseOnly(*onlyRaw) + if err != nil { + fmt.Fprintf(stderr, "notarius: %v\n", err) + return 2 + } + + cfg, _, err := loadConfig(*configPath, opts) + if err != nil { + fmt.Fprintf(stderr, "notarius: %v\n", err) + return 1 + } + if dir := strings.TrimSpace(*diagnosticsDir); dir != "" { + cfg.Diagnostics.WorkDir = dir + } + + catalog, err := effectiveCatalog(opts) + if err != nil { + fmt.Fprintf(stderr, "notarius: %v\n", err) + return 1 + } + effective, err := cfg.Resolve(config.ResolveInput{ + PipelineID: pipelineID, + Only: only, + Catalog: catalog, + LLMProfileOverride: *llmProfile, + }) + if err != nil { + fmt.Fprintf(stderr, "notarius: %v\n", err) + return 1 + } + + profileIDs := effectiveLLMProfileIDs(effective.ResolvedPipeline) + if len(profileIDs) != 1 { + fmt.Fprintf(stderr, "notarius: pipeline %q uses %d distinct LLM profiles; current runs require exactly one: %s\n", pipelineID, len(profileIDs), strings.Join(profileIDs, ", ")) + return 1 + } + + rawInput, err := os.ReadFile(strings.TrimSpace(*inputPath)) + if err != nil { + fmt.Fprintf(stderr, "notarius: read input %q: %v\n", strings.TrimSpace(*inputPath), err) + return 1 + } + + registries, err := effectiveRegistries(opts) + if err != nil { + fmt.Fprintf(stderr, "notarius: %v\n", err) + return 1 + } + + ctx := context.Background() + llmClient, llmProfiles, err := opts.LLMClientFactory(ctx, effective.Config, profileIDs[0]) + if err != nil { + fmt.Fprintf(stderr, "notarius: create LLM client for profile %q: %v\n", profileIDs[0], err) + return 1 + } + + output, err := pipeline.New(registries).Run(ctx, pipeline.RunInput{ + Pipeline: effective.ResolvedPipeline, + Path: strings.TrimSpace(*inputPath), + RawInput: rawInput, + LLMClient: llmClient, + StartedAt: opts.Now().UTC(), + LLMProfiles: llmProfiles, + Metadata: runMetadata(*outputDir, *diagnosticsDir), + }) + if err != nil { + fmt.Fprintf(stderr, "notarius: run pipeline %q: %v\n", pipelineID, err) + return 1 + } + + fmt.Fprintf(stdout, "pipeline %q complete: approved=%d rejected=%d\n", effective.PipelineID, len(output.Approved), len(output.Rejected)) + if len(output.Warnings) > 0 { + fmt.Fprintf(stderr, "notarius: run completed with %d warning(s)\n", len(output.Warnings)) + } + return 0 +} + +func reorderRunArgs(args []string) []string { + var flags []string + var positionals []string + for i := 0; i < len(args); i++ { + arg := args[i] + if arg == "--" { + positionals = append(positionals, args[i+1:]...) + break + } + if strings.HasPrefix(arg, "-") { + flags = append(flags, arg) + if runFlagTakesValue(arg) && !strings.Contains(arg, "=") && i+1 < len(args) { + i++ + flags = append(flags, args[i]) + } + continue + } + positionals = append(positionals, arg) + } + return append(flags, positionals...) +} + +func runFlagTakesValue(arg string) bool { + switch arg { + case "--config", "--input", "--only", "--output-dir", "--diagnostics-dir", "--llm-profile": + return true + default: + return false + } +} + +func effectiveLLMProfileIDs(resolved pipeline.ResolvedPipeline) []string { + seen := make(map[string]struct{}) + add := func(binding pipeline.ModuleBinding) { + id := strings.TrimSpace(binding.LLMProfile) + if id != "" { + seen[id] = struct{}{} + } + } + add(resolved.Input) + add(resolved.Chunk) + add(resolved.Output) + for _, lane := range resolved.ArtifactLanes { + add(lane.Extract) + add(lane.Merge) + add(lane.Normalize) + for _, validator := range lane.Validators { + add(validator) + } + } + ids := make([]string, 0, len(seen)) + for id := range seen { + ids = append(ids, id) + } + sort.Strings(ids) + return ids +} + +func runMetadata(outputDir, diagnosticsDir string) map[string]any { + metadata := make(map[string]any) + if dir := strings.TrimSpace(outputDir); dir != "" { + metadata["output_dir"] = dir + } + if dir := strings.TrimSpace(diagnosticsDir); dir != "" { + metadata["diagnostics_dir"] = dir + } + if len(metadata) == 0 { + return nil + } + return metadata +} + func runConfig(args []string, stdout, stderr io.Writer, opts Options) int { if len(args) == 0 { fmt.Fprintln(stderr, "notarius: config requires a subcommand") diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index 7f70a1a..427e21a 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -3,12 +3,15 @@ package cli import ( "bytes" "context" + "encoding/json" + "errors" "os" "path/filepath" "reflect" "strings" "testing" + "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" @@ -495,6 +498,211 @@ func TestProductionLLMClientFactoryReturnsScheduledClientAndManifestMetadata(t * } } +func TestRunPipelineMissingPipelineID(t *testing.T) { + configPath := writeTestConfig(t, mvpConfigYAML("dnd-session", "dnd/spells")) + inputPath := writeSeriatimInput(t) + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{"run", "--config", configPath, "--input", inputPath}, &stdout, &stderr, Options{}) + + if code != 2 { + t.Fatalf("RunWithOptions() code = %d, want 2", code) + } + if !strings.Contains(stderr.String(), "pipeline ID") { + t.Fatalf("stderr = %q, want missing pipeline ID error", stderr.String()) + } +} + +func TestRunPipelineMissingInputFlag(t *testing.T) { + configPath := writeTestConfig(t, mvpConfigYAML("dnd-session", "dnd/spells")) + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{"run", "dnd-session", "--config", configPath}, &stdout, &stderr, Options{}) + + if code != 2 { + t.Fatalf("RunWithOptions() code = %d, want 2", code) + } + if !strings.Contains(stderr.String(), "--input") { + t.Fatalf("stderr = %q, want missing input error", stderr.String()) + } +} + +func TestRunPipelineRejectsUnknownFlag(t *testing.T) { + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{"run", "dnd-session", "--extractor", "dnd/spells"}, &stdout, &stderr, Options{}) + + if code != 2 { + t.Fatalf("RunWithOptions() code = %d, want 2", code) + } + if !strings.Contains(stderr.String(), "flag provided but not defined") { + t.Fatalf("stderr = %q, want invalid flag error", stderr.String()) + } +} + +func TestRunPipelineUnknownPipeline(t *testing.T) { + configPath := writeTestConfig(t, mvpConfigYAML("dnd-session", "dnd/spells")) + inputPath := writeSeriatimInput(t) + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{"run", "missing", "--config", configPath, "--input", inputPath}, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(newFakeRunLLMClient(false), nil), + }) + + if code != 1 { + t.Fatalf("RunWithOptions() code = %d, want 1", code) + } + if !strings.Contains(stderr.String(), "pipeline \"missing\" is not configured") { + t.Fatalf("stderr = %q, want unknown pipeline error", stderr.String()) + } +} + +func TestRunPipelineUnknownOnlyLane(t *testing.T) { + configPath := writeTestConfig(t, mvpConfigYAML("dnd-session", "dnd/spells")) + inputPath := writeSeriatimInput(t) + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{"run", "dnd-session", "--config", configPath, "--input", inputPath, "--only", "missing"}, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(newFakeRunLLMClient(false), nil), + }) + + if code != 1 { + t.Fatalf("RunWithOptions() code = %d, want 1", code) + } + if !strings.Contains(stderr.String(), "selected artifact lane") { + t.Fatalf("stderr = %q, want selected lane error", stderr.String()) + } +} + +func TestRunPipelineInvalidInputPath(t *testing.T) { + configPath := writeTestConfig(t, mvpConfigYAML("dnd-session", "dnd/spells")) + inputPath := filepath.Join(t.TempDir(), "missing.json") + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{"run", "dnd-session", "--config", configPath, "--input", inputPath}, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(newFakeRunLLMClient(false), nil), + }) + + if code != 1 { + t.Fatalf("RunWithOptions() code = %d, want 1", code) + } + if !strings.Contains(stderr.String(), "read input") { + t.Fatalf("stderr = %q, want input read error", stderr.String()) + } +} + +func TestRunPipelineSuccessUsesProductionRegistriesAndFakeLLM(t *testing.T) { + configPath := writeTestConfig(t, mvpConfigYAML("dnd-session", "dnd/spells")) + inputPath := writeSeriatimInput(t) + client := newFakeRunLLMClient(false) + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{"run", "dnd-session", "--config", configPath, "--input", inputPath}, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(client, nil), + }) + + if code != 0 { + t.Fatalf("RunWithOptions() code = %d, stderr=%q", code, stderr.String()) + } + if client.calls != 1 { + t.Fatalf("LLM calls = %d, want 1", client.calls) + } + for _, want := range []string{"dnd-session", "approved=1", "rejected=0"} { + if !strings.Contains(stdout.String(), want) { + t.Fatalf("stdout = %q, want substring %q", stdout.String(), want) + } + } + if stderr.Len() != 0 { + t.Fatalf("stderr = %q, want empty", stderr.String()) + } +} + +func TestRunPipelineOnlySelectsRequestedLane(t *testing.T) { + configPath := writeTestConfig(t, mvpConfigYAMLForLanes("dnd-session", "spells", "rituals")) + inputPath := writeSeriatimInput(t) + client := newFakeRunLLMClient(false) + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{"run", "dnd-session", "--config", configPath, "--input", inputPath, "--only", "spells"}, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(client, nil), + }) + + if code != 0 { + t.Fatalf("RunWithOptions() code = %d, stderr=%q", code, stderr.String()) + } + if client.calls != 1 { + t.Fatalf("LLM calls = %d, want only selected lane to run once", client.calls) + } + if !strings.Contains(stdout.String(), "approved=1") { + t.Fatalf("stdout = %q, want approved count", stdout.String()) + } +} + +func TestRunPipelineLLMFactoryFailure(t *testing.T) { + configPath := writeTestConfig(t, mvpConfigYAML("dnd-session", "dnd/spells")) + inputPath := writeSeriatimInput(t) + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{"run", "dnd-session", "--config", configPath, "--input", inputPath}, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(newFakeRunLLMClient(false), errors.New("factory unavailable")), + }) + + if code != 1 { + t.Fatalf("RunWithOptions() code = %d, want 1", code) + } + if !strings.Contains(stderr.String(), "create LLM client") || !strings.Contains(stderr.String(), "factory unavailable") { + t.Fatalf("stderr = %q, want LLM factory error", stderr.String()) + } +} + +func TestRunPipelineValidationRejectionCompletesSuccessfully(t *testing.T) { + configPath := writeTestConfig(t, mvpConfigYAML("dnd-session", "dnd/spells")) + inputPath := writeSeriatimInput(t) + client := newFakeRunLLMClient(true) + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{"run", "dnd-session", "--config", configPath, "--input", inputPath}, &stdout, &stderr, Options{ + LLMClientFactory: fakeLLMFactory(client, nil), + }) + + if code != 0 { + t.Fatalf("RunWithOptions() code = %d, stderr=%q", code, stderr.String()) + } + if !strings.Contains(stdout.String(), "approved=0") || !strings.Contains(stdout.String(), "rejected=1") { + t.Fatalf("stdout = %q, want rejection counts", stdout.String()) + } +} + +func TestRunPipelineLLMProfileOverrideSelectsFactoryProfile(t *testing.T) { + configPath := writeTestConfig(t, mvpConfigYAMLWithProfiles("dnd-session")) + inputPath := writeSeriatimInput(t) + client := newFakeRunLLMClient(false) + factory := &recordingLLMFactory{client: client} + var stdout bytes.Buffer + var stderr bytes.Buffer + + code := RunWithOptions([]string{"run", "dnd-session", "--config", configPath, "--input", inputPath, "--llm-profile", "runtime"}, &stdout, &stderr, Options{ + LLMClientFactory: factory.build, + }) + + if code != 0 { + t.Fatalf("RunWithOptions() code = %d, stderr=%q", code, stderr.String()) + } + if got, want := strings.Join(factory.profileIDs, ","), "runtime"; got != want { + t.Fatalf("factory profile IDs = %q, want %q", got, want) + } +} + func writeTestConfig(t *testing.T, content string) string { t.Helper() return writeFile(t, "config.yml", content) @@ -540,6 +748,115 @@ pipelines: ` } +func mvpConfigYAMLForLanes(pipelineID string, laneIDs ...string) string { + var b strings.Builder + b.WriteString("version: 1\n") + b.WriteString("pipelines:\n") + b.WriteString(" " + pipelineID + ":\n") + b.WriteString(" input: seriatim\n") + b.WriteString(" artifacts:\n") + for _, laneID := range laneIDs { + b.WriteString(" " + laneID + ":\n") + b.WriteString(" extract: dnd/spells\n") + } + return b.String() +} + +func mvpConfigYAMLWithProfiles(pipelineID string) string { + return `version: 1 +llm_profiles: + default: + provider: openai-compatible + runtime: + provider: openai-compatible +pipelines: + ` + pipelineID + `: + input: seriatim + artifacts: + spells: + extract: dnd/spells +` +} + +func writeSeriatimInput(t *testing.T) string { + t.Helper() + return writeFile(t, "source.json", `{ + "metadata": { + "id": "session-alpha" + }, + "segments": [ + { + "id": "seg-001", + "start": 0, + "end": 1, + "speaker": "Aria", + "text": "Aria casts Cure Wounds." + } + ] +}`) +} + +type fakeRunLLMClient struct { + invalidSourceRef bool + calls int +} + +func newFakeRunLLMClient(invalidSourceRef bool) *fakeRunLLMClient { + return &fakeRunLLMClient{invalidSourceRef: invalidSourceRef} +} + +func (client *fakeRunLLMClient) CompleteStructured(ctx context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) { + client.calls++ + startUnitID := "seg-001" + if client.invalidSourceRef { + startUnitID = "missing-segment" + } + payload := map[string]any{ + "spell_casts": []map[string]any{ + { + "caster": "Aria", + "spell": "Cure Wounds", + "effect": "Heals a wounded ally.", + "narrative_description": "Aria casts Cure Wounds.", + "source_refs": []map[string]string{ + { + "source_id": "session-alpha", + "start_unit_id": startUnitID, + "end_unit_id": "seg-001", + }, + }, + }, + }, + } + encoded, err := json.Marshal(payload) + if err != nil { + return contracts.StructuredCompletionResponse{}, err + } + if err := json.Unmarshal(encoded, out); err != nil { + return contracts.StructuredCompletionResponse{}, err + } + return contracts.StructuredCompletionResponse{Content: encoded}, nil +} + +func fakeLLMFactory(client contracts.StructuredLLMClient, err error) LLMClientFactory { + return func(ctx context.Context, cfg config.Config, profileID string) (contracts.StructuredLLMClient, []artifacts.LLMProfileManifest, error) { + if err != nil { + return nil, nil, err + } + return client, []artifacts.LLMProfileManifest{{ID: strings.TrimSpace(profileID)}}, nil + } +} + +type recordingLLMFactory struct { + client contracts.StructuredLLMClient + profileIDs []string +} + +func (factory *recordingLLMFactory) build(ctx context.Context, cfg config.Config, profileID string) (contracts.StructuredLLMClient, []artifacts.LLMProfileManifest, error) { + factory.profileIDs = append(factory.profileIDs, strings.TrimSpace(profileID)) + return factory.client, []artifacts.LLMProfileManifest{{ID: strings.TrimSpace(profileID)}}, nil +} + func fakeCatalog(t *testing.T) pipeline.ModuleCatalog { t.Helper() inputs := pipeline.NewInputAdapterRegistry() diff --git a/internal/core/config/effective_config.go b/internal/core/config/effective_config.go index 8de750a..cc959ca 100644 --- a/internal/core/config/effective_config.go +++ b/internal/core/config/effective_config.go @@ -10,9 +10,10 @@ import ( ) type ResolveInput struct { - PipelineID string - Only []string - Catalog pipeline.ModuleCatalog + PipelineID string + Only []string + Catalog pipeline.ModuleCatalog + LLMProfileOverride string } type EffectiveConfig struct { @@ -38,6 +39,12 @@ func (c Config) Resolve(input ResolveInput) (EffectiveConfig, error) { } profile = clonePipelineProfile(profile) profile.ID = pipelineID + if override := strings.TrimSpace(input.LLMProfileOverride); override != "" { + if !hasLLMProfile(c.LLMProfiles, override) { + return EffectiveConfig{}, fmt.Errorf("LLM profile override %q is not configured", override) + } + applyLLMProfileOverride(&profile, override) + } resolved, err := pipeline.ResolvePipeline(profile, pipeline.ResolveOptions{Only: input.Only}, input.Catalog) if err != nil { @@ -52,6 +59,21 @@ func (c Config) Resolve(input ResolveInput) (EffectiveConfig, error) { }, nil } +func applyLLMProfileOverride(profile *pipeline.PipelineProfile, profileID string) { + profile.Input.LLMProfile = profileID + profile.Chunk.LLMProfile = profileID + profile.Output.LLMProfile = profileID + for laneID, lane := range profile.Artifacts { + lane.Extract.LLMProfile = profileID + lane.Merge.LLMProfile = profileID + lane.Normalize.LLMProfile = profileID + for i := range lane.Validators { + lane.Validators[i].LLMProfile = profileID + } + profile.Artifacts[laneID] = lane + } +} + func lookupPipelineProfile(profiles map[string]pipeline.PipelineProfile, pipelineID string) (pipeline.PipelineProfile, bool) { pipelineID = strings.TrimSpace(pipelineID) for rawID, profile := range profiles { diff --git a/internal/core/config/effective_config_test.go b/internal/core/config/effective_config_test.go index 89ceb0b..7c120ef 100644 --- a/internal/core/config/effective_config_test.go +++ b/internal/core/config/effective_config_test.go @@ -118,6 +118,51 @@ func TestResolveDigestChangesWhenEffectiveConfigChanges(t *testing.T) { } } +func TestResolveLLMProfileOverrideAppliesBeforeDigest(t *testing.T) { + cfg := validConfig() + cfg.LLMProfiles["runtime"] = LLMProfile{Provider: "openai-compatible"} + + base, err := cfg.Resolve(ResolveInput{PipelineID: "example", Catalog: fakeCatalog(t)}) + if err != nil { + t.Fatalf("Resolve base: %v", err) + } + effective, err := cfg.Resolve(ResolveInput{ + PipelineID: "example", + Catalog: fakeCatalog(t), + LLMProfileOverride: "runtime", + }) + if err != nil { + t.Fatalf("Resolve override: %v", err) + } + + if base.ResolvedPipeline.Digest == effective.ResolvedPipeline.Digest { + t.Fatalf("expected digest to change after LLM profile override") + } + for _, binding := range resolvedBindings(effective.ResolvedPipeline) { + if binding.LLMProfile != "runtime" { + t.Fatalf("binding profile = %q, want runtime", binding.LLMProfile) + } + } + + _, err = cfg.Resolve(ResolveInput{ + PipelineID: "example", + Catalog: fakeCatalog(t), + LLMProfileOverride: "missing", + }) + if err == nil || !strings.Contains(err.Error(), "LLM profile override") { + t.Fatalf("expected override profile error, got %v", err) + } +} + +func resolvedBindings(resolved pipeline.ResolvedPipeline) []pipeline.ModuleBinding { + bindings := []pipeline.ModuleBinding{resolved.Input, resolved.Chunk, resolved.Output} + for _, lane := range resolved.ArtifactLanes { + bindings = append(bindings, lane.Extract, lane.Merge, lane.Normalize) + bindings = append(bindings, lane.Validators...) + } + return bindings +} + func TestOpenAICompatibleClientConfigRejectsIncompleteDefaultProfile(t *testing.T) { cfg := Default()